Mailing-List: contact common-commits-help@hadoop.apache.org; run by ezmlm
Precedence: bulk
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: subru@apache.org
To: common-commits@hadoop.apache.org
Date: Tue, 08 Nov 2016 22:46:26 -0000
Message-Id: <d552e554c6b6435fbf7816780e304d03@git.apache.org>
In-Reply-To: <d083509d5c3b4e6a9e13c5e3d903f2eb@git.apache.org>
References: <d083509d5c3b4e6a9e13c5e3d903f2eb@git.apache.org>
Subject: [21/50] [abbrv] hadoop git commit: HDFS-11114. Support for running
 async disk checks in DataNode.
archived-at: Tue, 08 Nov 2016 22:46:14 -0000

HDFS-11114. Support for running async disk checks in DataNode.

This closes #153.


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/3fff1585
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/3fff1585
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/3fff1585

Branch: refs/heads/YARN-2915
Commit: 3fff1585875ad322ce6e8acb485275e6a4360823
Parents: 3dbad5d
Author: Arpit Agarwal <arp@apache.org>
Authored: Mon Nov 7 18:45:53 2016 -0800
Committer: Arpit Agarwal <arp@apache.org>
Committed: Mon Nov 7 18:45:53 2016 -0800

----------------------------------------------------------------------
 .../server/datanode/checker/AsyncChecker.java   |  63 +++++
 .../hdfs/server/datanode/checker/Checkable.java |  49 ++++
 .../datanode/checker/ThrottledAsyncChecker.java | 224 +++++++++++++++
 .../server/datanode/checker/package-info.java   |  26 ++
 .../checker/TestThrottledAsyncChecker.java      | 276 +++++++++++++++++++
 5 files changed, 638 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/3fff1585/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
new file mode 100644
index 0000000..1d534a3
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.server.datanode.checker;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A class that can be used to schedule an asynchronous check on a given
+ * {@link Checkable}. If the check is successfully scheduled then a
+ * {@link ListenableFuture} is returned.
+ *
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public interface AsyncChecker<K, V> {
+
+  /**
+   * Schedule an asynchronous check for the given object.
+   *
+   * @param target object to be checked.
+   *
+   * @param context the interpretation of the context depends on the
+   *                target.
+   *
+   * @return returns a {@link ListenableFuture} that can be used to
+   *         retrieve the result of the asynchronous check.
+   */
+  ListenableFuture<V> schedule(Checkable<K, V> target, K context);
+
+  /**
+   * Cancel all executing checks and wait for them to complete.
+   * First attempts a graceful cancellation, then cancels forcefully.
+   * Waits for the supplied timeout after both attempts.
+   *
+   * See {@link ExecutorService#awaitTermination} for a description of
+   * the parameters.
+   *
+   * @throws InterruptedException
+   */
+  void shutdownAndWait(long timeout, TimeUnit timeUnit)
+      throws InterruptedException;
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3fff1585/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/Checkable.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/Checkable.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/Checkable.java
new file mode 100644
index 0000000..833ebda
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/Checkable.java
@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.server.datanode.checker;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+
+/**
+ * A Checkable is an object whose health can be probed by invoking its
+ * {@link #check} method.
+ *
+ * e.g. a {@link Checkable} instance may represent a single hardware
+ * resource.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public interface Checkable<K, V> {
+
+  /**
+   * Query the health of this object. This method may hang
+   * indefinitely depending on the status of the target resource.
+   *
+   * @param context for the probe operation. May be null depending
+   *                on the implementation.
+   *
+   * @return result of the check operation.
+   *
+   * @throws Exception encountered during the check operation. An
+   *                   exception indicates that the check failed.
+   */
+  V check(K context) throws Exception;
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3fff1585/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
new file mode 100644
index 0000000..d0ee3d2
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
@@ -0,0 +1,224 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.server.datanode.checker;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.util.Timer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.WeakHashMap;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * An implementation of {@link AsyncChecker} that skips checking recently
+ * checked objects. It will enforce at least {@link minMsBetweenChecks}
+ * milliseconds between two successive checks of any one object.
+ *
+ * It is assumed that the total number of Checkable objects in the system
+ * is small, (not more than a few dozen) since the checker uses O(Checkables)
+ * storage and also potentially O(Checkables) threads.
+ *
+ * {@link minMsBetweenChecks} should be configured reasonably
+ * by the caller to avoid spinning up too many threads frequently.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class ThrottledAsyncChecker<K, V> implements AsyncChecker<K, V> {
+  public static final Logger LOG =
+      LoggerFactory.getLogger(ThrottledAsyncChecker.class);
+
+  private final Timer timer;
+
+  /**
+   * The ExecutorService used to schedule asynchronous checks.
+   */
+  private final ListeningExecutorService executorService;
+
+  /**
+   * The minimum gap in milliseconds between two successive checks
+   * of the same object. This is the throttle.
+   */
+  private final long minMsBetweenChecks;
+
+  /**
+   * Map of checks that are currently in progress. Protected by the object
+   * lock.
+   */
+  private final Map<Checkable, ListenableFuture<V>> checksInProgress;
+
+  /**
+   * Maps Checkable objects to a future that can be used to retrieve
+   * the results of the operation.
+   * Protected by the object lock.
+   */
+  private final Map<Checkable, LastCheckResult<V>> completedChecks;
+
+  ThrottledAsyncChecker(final Timer timer,
+                        final long minMsBetweenChecks,
+                        final ExecutorService executorService) {
+    this.timer = timer;
+    this.minMsBetweenChecks = minMsBetweenChecks;
+    this.executorService = MoreExecutors.listeningDecorator(executorService);
+    this.checksInProgress = new HashMap<>();
+    this.completedChecks = new WeakHashMap<>();
+  }
+
+  /**
+   * See {@link AsyncChecker#schedule}
+   *
+   * If the object has been checked recently then the check will
+   * be skipped. Multiple concurrent checks for the same object
+   * will receive the same Future.
+   */
+  @Override
+  public synchronized ListenableFuture<V> schedule(
+      final Checkable<K, V> target,
+      final K context) {
+    LOG.debug("Scheduling a check of {}", target);
+
+    if (checksInProgress.containsKey(target)) {
+      return checksInProgress.get(target);
+    }
+
+    if (completedChecks.containsKey(target)) {
+      final LastCheckResult<V> result = completedChecks.get(target);
+      final long msSinceLastCheck = timer.monotonicNow() - result.completedAt;
+      if (msSinceLastCheck < minMsBetweenChecks) {
+        LOG.debug("Skipped checking {}. Time since last check {}ms " +
+            "is less than the min gap {}ms.",
+            target, msSinceLastCheck, minMsBetweenChecks);
+        return result.result != null ?
+            Futures.immediateFuture(result.result) :
+            Futures.immediateFailedFuture(result.exception);
+      }
+    }
+
+    final ListenableFuture<V> lf = executorService.submit(
+        new Callable<V>() {
+          @Override
+          public V call() throws Exception {
+            return target.check(context);
+          }
+        });
+    checksInProgress.put(target, lf);
+    addResultCachingCallback(target, lf);
+    return lf;
+  }
+
+  /**
+   * Register a callback to cache the result of a check.
+   * @param target
+   * @param lf
+   */
+  private void addResultCachingCallback(
+      Checkable<K, V> target, ListenableFuture<V> lf) {
+    Futures.addCallback(lf, new FutureCallback<V>() {
+      @Override
+      public void onSuccess(@Nullable V result) {
+        synchronized (ThrottledAsyncChecker.this) {
+          checksInProgress.remove(target);
+          completedChecks.put(target, new LastCheckResult<>(
+              result, timer.monotonicNow()));
+        }
+      }
+
+      @Override
+      public void onFailure(@Nonnull Throwable t) {
+        synchronized (ThrottledAsyncChecker.this) {
+          checksInProgress.remove(target);
+          completedChecks.put(target, new LastCheckResult<>(
+              t, timer.monotonicNow()));
+        }
+      }
+    });
+  }
+
+  /**
+   * {@inheritDoc}.
+   */
+  @Override
+  public void shutdownAndWait(long timeout, TimeUnit timeUnit)
+      throws InterruptedException {
+    // Try orderly shutdown.
+    executorService.shutdown();
+
+    if (!executorService.awaitTermination(timeout, timeUnit)) {
+      // Interrupt executing tasks and wait again.
+      executorService.shutdownNow();
+      executorService.awaitTermination(timeout, timeUnit);
+    }
+  }
+
+  /**
+   * Status of running a check. It can either be a result or an
+   * exception, depending on whether the check completed or threw.
+   */
+  private static final class LastCheckResult<V> {
+    /**
+     * Timestamp at which the check completed.
+     */
+    private final long completedAt;
+
+    /**
+     * Result of running the check if it completed. null if it threw.
+     */
+    @Nullable
+    private final V result;
+
+    /**
+     * Exception thrown by the check. null if it returned a result.
+     */
+    private final Throwable exception; // null on success.
+
+    /**
+     * Initialize with a result.
+     * @param result
+     */
+    private LastCheckResult(V result, long completedAt) {
+      this.result = result;
+      this.exception = null;
+      this.completedAt = completedAt;
+    }
+
+    /**
+     * Initialize with an exception.
+     * @param completedAt
+     * @param t
+     */
+    private LastCheckResult(Throwable t, long completedAt) {
+      this.result = null;
+      this.exception = t;
+      this.completedAt = completedAt;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3fff1585/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/package-info.java
new file mode 100644
index 0000000..52822e9
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/package-info.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Datanode support for running disk checks.
+ */
+@InterfaceAudience.LimitedPrivate({"HDFS"})
+@InterfaceStability.Evolving
+package org.apache.hadoop.hdfs.server.datanode.checker;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/3fff1585/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
new file mode 100644
index 0000000..70795ca
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
@@ -0,0 +1,276 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.server.datanode.checker;
+
+import com.google.common.base.Supplier;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.util.FakeTimer;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.core.Is.isA;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Verify functionality of {@link ThrottledAsyncChecker}.
+ */
+public class TestThrottledAsyncChecker {
+  public static final Logger LOG =
+      LoggerFactory.getLogger(TestThrottledAsyncChecker.class);
+  private static final long MIN_ERROR_CHECK_GAP = 1000;
+
+  @Rule
+  public ExpectedException thrown = ExpectedException.none();
+
+  /**
+   * Test various scheduling combinations to ensure scheduling and
+   * throttling behave as expected.
+   */
+  @Test(timeout=60000)
+  public void testScheduler() throws Exception {
+    final NoOpCheckable target1 = new NoOpCheckable();
+    final NoOpCheckable target2 = new NoOpCheckable();
+    final FakeTimer timer = new FakeTimer();
+    ThrottledAsyncChecker<Boolean, Boolean> checker =
+        new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
+                                    getExecutorService());
+
+    // check target1 and ensure we get back the expected result.
+    assertTrue(checker.schedule(target1, true).get());
+    assertThat(target1.numChecks.get(), is(1L));
+
+    // Check target1 again without advancing the timer. target1 should not
+    // be checked again and the cached result should be returned.
+    assertTrue(checker.schedule(target1, true).get());
+    assertThat(target1.numChecks.get(), is(1L));
+
+    // Schedule target2 scheduled without advancing the timer.
+    // target2 should be checked as it has never been checked before.
+    assertTrue(checker.schedule(target2, true).get());
+    assertThat(target2.numChecks.get(), is(1L));
+
+    // Advance the timer but just short of the min gap.
+    // Neither target1 nor target2 should be checked again.
+    timer.advance(MIN_ERROR_CHECK_GAP - 1);
+    assertTrue(checker.schedule(target1, true).get());
+    assertThat(target1.numChecks.get(), is(1L));
+    assertTrue(checker.schedule(target2, true).get());
+    assertThat(target2.numChecks.get(), is(1L));
+
+    // Advance the timer again.
+    // Both targets should be checked now.
+    timer.advance(MIN_ERROR_CHECK_GAP);
+    assertTrue(checker.schedule(target1, true).get());
+    assertThat(target1.numChecks.get(), is(2L));
+    assertTrue(checker.schedule(target2, true).get());
+    assertThat(target1.numChecks.get(), is(2L));
+  }
+
+  @Test (timeout=60000)
+  public void testCancellation() throws Exception {
+    LatchedCheckable target = new LatchedCheckable();
+    final FakeTimer timer = new FakeTimer();
+    final LatchedCallback callback = new LatchedCallback(target);
+    ThrottledAsyncChecker<Boolean, Boolean> checker =
+        new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
+                                    getExecutorService());
+
+    ListenableFuture<Boolean> lf = checker.schedule(target, true);
+    Futures.addCallback(lf, callback);
+
+    // Request immediate cancellation.
+    checker.shutdownAndWait(0, TimeUnit.MILLISECONDS);
+    try {
+      assertFalse(lf.get());
+      fail("Failed to get expected InterruptedException");
+    } catch (ExecutionException ee) {
+      assertTrue(ee.getCause() instanceof InterruptedException);
+    }
+    callback.failureLatch.await();
+  }
+
+  @Test (timeout=60000)
+  public void testConcurrentChecks() throws Exception {
+    LatchedCheckable target = new LatchedCheckable();
+    final FakeTimer timer = new FakeTimer();
+    ThrottledAsyncChecker<Boolean, Boolean> checker =
+        new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
+                                    getExecutorService());
+    final ListenableFuture<Boolean> lf1 = checker.schedule(target, true);
+    final ListenableFuture<Boolean> lf2 = checker.schedule(target, true);
+
+    // Ensure that concurrent requests return the same future object.
+    assertTrue(lf1 == lf2);
+
+    // Unblock the latch and wait for it to finish execution.
+    target.latch.countDown();
+    lf1.get();
+
+    GenericTestUtils.waitFor(new Supplier<Boolean>() {
+      @Override
+      public Boolean get() {
+        // We should not get back the same future as before.
+        // This can take a short while until the internal callback in
+        // ThrottledAsyncChecker is scheduled for execution.
+        // Also this should not trigger a new check operation as the timer
+        // was not advanced. If it does trigger a new check then the test
+        // will fail with a timeout.
+        final ListenableFuture<Boolean> lf3 = checker.schedule(target, true);
+        return lf3 != lf2;
+      }
+    }, 100, 10000);
+  }
+
+  /**
+   * Ensure that the context is passed through to the Checkable#check
+   * method.
+   * @throws Exception
+   */
+  @Test(timeout=60000)
+  public void testContextIsPassed() throws Exception {
+    final NoOpCheckable target1 = new NoOpCheckable();
+    final FakeTimer timer = new FakeTimer();
+    ThrottledAsyncChecker<Boolean, Boolean> checker =
+        new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
+            getExecutorService());
+
+    assertTrue(checker.schedule(target1, true).get());
+    assertThat(target1.numChecks.get(), is(1L));
+    timer.advance(MIN_ERROR_CHECK_GAP + 1);
+    assertFalse(checker.schedule(target1, false).get());
+    assertThat(target1.numChecks.get(), is(2L));
+  }
+
+  /**
+   * Ensure that the exeption from a failed check is cached
+   * and returned without re-running the check when the minimum
+   * gap has not elapsed.
+   *
+   * @throws Exception
+   */
+  @Test(timeout=60000)
+  public void testExceptionCaching() throws Exception {
+    final ThrowingCheckable target1 = new ThrowingCheckable();
+    final FakeTimer timer = new FakeTimer();
+    ThrottledAsyncChecker<Boolean, Boolean> checker =
+        new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP,
+            getExecutorService());
+
+    thrown.expectCause(isA(DummyException.class));
+    checker.schedule(target1, true).get();
+    assertThat(target1.numChecks.get(), is(1L));
+
+    thrown.expectCause(isA(DummyException.class));
+    checker.schedule(target1, true).get();
+    assertThat(target1.numChecks.get(), is(2L));
+  }
+
+  /**
+   * A simple ExecutorService for testing.
+   */
+  private ExecutorService getExecutorService() {
+    return new ScheduledThreadPoolExecutor(1);
+  }
+
+  /**
+   * A Checkable that just returns its input.
+   */
+  private static class NoOpCheckable
+      implements Checkable<Boolean, Boolean> {
+    private final AtomicLong numChecks = new AtomicLong(0);
+    @Override
+    public Boolean check(Boolean context) {
+      numChecks.incrementAndGet();
+      return context;
+    }
+  }
+
+  private static class ThrowingCheckable
+      implements Checkable<Boolean, Boolean> {
+    private final AtomicLong numChecks = new AtomicLong(0);
+    @Override
+    public Boolean check(Boolean context) throws DummyException {
+      numChecks.incrementAndGet();
+      throw new DummyException();
+    }
+
+  }
+
+  private static class DummyException extends Exception {
+  }
+
+  /**
+   * A checkable that hangs until signaled.
+   */
+  private static class LatchedCheckable
+      implements Checkable<Boolean, Boolean> {
+    private final CountDownLatch latch = new CountDownLatch(1);
+
+    @Override
+    public Boolean check(Boolean ignored) throws InterruptedException {
+      LOG.info("LatchedCheckable {} waiting.", this);
+      latch.await();
+      return true;  // Unreachable.
+    }
+  }
+
+  /**
+   * A {@link FutureCallback} that counts its invocations.
+   */
+  private static final class LatchedCallback
+      implements FutureCallback<Boolean> {
+    private final CountDownLatch successLatch = new CountDownLatch(1);
+    private final CountDownLatch failureLatch = new CountDownLatch(1);
+    private final Checkable target;
+
+    private LatchedCallback(Checkable target) {
+      this.target = target;
+    }
+
+    @Override
+    public void onSuccess(@Nonnull Boolean result) {
+      LOG.info("onSuccess callback invoked for {}", target);
+      successLatch.countDown();
+    }
+
+    @Override
+    public void onFailure(@Nonnull Throwable t) {
+      LOG.info("onFailure callback invoked for {} with exception", target, t);
+      failureLatch.countDown();
+    }
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org