Interrupt replaying hints on decommission Patch by Jeff Jirsa; Reviewed by Aleksey Yeschenko for CASSANDRA-13308
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5089e74e Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5089e74e Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5089e74e Branch: refs/heads/trunk Commit: 5089e74ef4a0eaeb1c439d57f074de1c496421f2 Parents: 3110d27 Author: Jeff Jirsa <jji...@apple.com> Authored: Wed Apr 19 08:26:02 2017 -0700 Committer: Jeff Jirsa <j...@jeffjirsa.net> Committed: Wed Apr 19 08:57:45 2017 -0700 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../apache/cassandra/hints/HintsDispatchExecutor.java | 8 ++++++++ src/java/org/apache/cassandra/hints/HintsDispatcher.java | 9 +++++++-- src/java/org/apache/cassandra/hints/HintsService.java | 11 ++++++----- 4 files changed, 22 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/5089e74e/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 918c46b..e55d4cb 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 3.0.14 * Handling partially written hint files (CASSANDRA-12728) + * Interrupt replaying hints on decommission (CASSANDRA-13308) 3.0.13 * Make reading of range tombstones more reliable (CASSANDRA-12811) http://git-wip-us.apache.org/repos/asf/cassandra/blob/5089e74e/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java b/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java index 333232d..58b30bd 100644 --- a/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java +++ b/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java @@ -117,6 +117,14 @@ final class HintsDispatchExecutor } } + void interruptDispatch(UUID hostId) + { + Future future = scheduledDispatches.remove(hostId); + + if (null != future) + future.cancel(true); + } + private final class TransferHintsTask implements Runnable { private final HintsCatalog catalog; http://git-wip-us.apache.org/repos/asf/cassandra/blob/5089e74e/src/java/org/apache/cassandra/hints/HintsDispatcher.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/hints/HintsDispatcher.java b/src/java/org/apache/cassandra/hints/HintsDispatcher.java index d7a3515..351b3fa 100644 --- a/src/java/org/apache/cassandra/hints/HintsDispatcher.java +++ b/src/java/org/apache/cassandra/hints/HintsDispatcher.java @@ -26,6 +26,8 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import com.google.common.util.concurrent.RateLimiter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.gms.FailureDetector; @@ -42,6 +44,8 @@ import org.apache.cassandra.utils.concurrent.SimpleCondition; */ final class HintsDispatcher implements AutoCloseable { + private static final Logger logger = LoggerFactory.getLogger(HintsDispatcher.class); + private enum Action { CONTINUE, ABORT } private final HintsReader reader; @@ -181,7 +185,7 @@ final class HintsDispatcher implements AutoCloseable private static final class Callback implements IAsyncCallbackWithFailure { - enum Outcome { SUCCESS, TIMEOUT, FAILURE } + enum Outcome { SUCCESS, TIMEOUT, FAILURE, INTERRUPTED } private final long start = System.nanoTime(); private final SimpleCondition condition = new SimpleCondition(); @@ -198,7 +202,8 @@ final class HintsDispatcher implements AutoCloseable } catch (InterruptedException e) { - throw new AssertionError(e); + logger.warn("Hint dispatch was interrupted", e); + return Outcome.INTERRUPTED; } return timedOut ? Outcome.TIMEOUT : outcome; http://git-wip-us.apache.org/repos/asf/cassandra/blob/5089e74e/src/java/org/apache/cassandra/hints/HintsService.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/hints/HintsService.java b/src/java/org/apache/cassandra/hints/HintsService.java index 5a32786..9cd4ed3 100644 --- a/src/java/org/apache/cassandra/hints/HintsService.java +++ b/src/java/org/apache/cassandra/hints/HintsService.java @@ -287,10 +287,11 @@ public final class HintsService implements HintsServiceMBean /** * Cleans up hints-related state after a node with id = hostId left. * - * Dispatcher should stop itself (isHostAlive() will start returning false for the leaving host), but we'll wait for - * completion anyway. + * Dispatcher can not stop itself (isHostAlive() can not start returning false for the leaving host because this + * method is called by the same thread as gossip, which blocks gossip), so we can't simply wait for + * completion. * - * We should also flush the buffer is there are any thints for the node there, and close the writer (if any), + * We should also flush the buffer if there are any hints for the node there, and close the writer (if any), * so that we don't leave any hint files lying around. * * Once that is done, we can simply delete all hint files and remove the host id from the catalog. @@ -319,8 +320,8 @@ public final class HintsService implements HintsServiceMBean throw new RuntimeException(e); } - // wait for the current dispatch session to end (if any), so that the currently dispatched file gets removed - dispatchExecutor.completeDispatchBlockingly(store); + // interrupt the current dispatch session to end (if any), so that the currently dispatched file gets removed + dispatchExecutor.interruptDispatch(store.hostId); // delete all the hints files and remove the HintsStore instance from the map in the catalog catalog.exciseStore(hostId);