Interrupt replaying hints on decommission

Patch by Jeff Jirsa; Reviewed by Aleksey Yeschenko for CASSANDRA-13308

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)

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;

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;

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);