hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a..@apache.org
Subject hadoop git commit: Revert "HDFS-11114. Support for running async disk checks in DataNode."
Date Tue, 08 Nov 2016 17:05:35 GMT
Repository: hadoop
Updated Branches:
  refs/heads/branch-2 a942d968c -> b77239b46


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

This reverts commit 732eaadddbe9a2682bf96ba06ce5e7ebfa58ba87.


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

Branch: refs/heads/branch-2
Commit: b77239b46d5918b3a3f4f7032e6c20a2b4fcbe56
Parents: a942d96
Author: Arpit Agarwal <arp@apache.org>
Authored: Tue Nov 8 09:04:58 2016 -0800
Committer: Arpit Agarwal <arp@apache.org>
Committed: Tue Nov 8 09:04:58 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 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/b77239b4/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
deleted file mode 100644
index 1d534a3..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/AsyncChecker.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/**
- * 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/b77239b4/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
deleted file mode 100644
index 833ebda..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/Checkable.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/**
- * 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/b77239b4/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
deleted file mode 100644
index d0ee3d2..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/ThrottledAsyncChecker.java
+++ /dev/null
@@ -1,224 +0,0 @@
-/**
- * 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/b77239b4/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
deleted file mode 100644
index 52822e9..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/checker/package-info.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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/b77239b4/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
deleted file mode 100644
index 70795ca..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/checker/TestThrottledAsyncChecker.java
+++ /dev/null
@@ -1,276 +0,0 @@
-/**
- * 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


Mime
View raw message