When jvm-dtest is shutting down an instance TCM retries block the shutdown causing the test to fail

patch by David Capwell; reviewed by Blake Eggleston for CASSANDRA-19514

src/java/org/apache/cassandra/concurrent/Shutdownable.java

@@ -19,7 +19,9 @@
 package org.apache.cassandra.concurrent;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
+import org.apache.cassandra.utils.ExecutorUtils;
 import org.apache.cassandra.utils.Shared;
 
 import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
@@ -29,6 +31,11 @@ public interface Shutdownable
 {
     boolean isTerminated();
 
+    default boolean isShutdown()
+    {
+        return isTerminated();
+    }
+
     /**
      * Shutdown once any remaining work has completed (however this is defined for the implementation).
      */
@@ -42,5 +49,10 @@ public interface Shutdownable
     /**
      * Await termination of this object, i.e. the cessation of all current and future work.
      */
-    public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException;
+    boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException;
+
+    default void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
+    {
+        ExecutorUtils.shutdownAndWait(timeout, unit, this);
+    }
 }

src/java/org/apache/cassandra/service/accord/AccordService.java

@@ -35,7 +35,9 @@ import com.google.common.primitives.Ints;
 import accord.coordinate.TopologyMismatch;
 import accord.impl.CoordinateDurabilityScheduling;
 import org.apache.cassandra.cql3.statements.RequestValidations;
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.service.accord.interop.AccordInteropAdapter.AccordInteropFactory;
+import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.ClusterMetadataService;
 import org.apache.cassandra.service.accord.api.*;
 import org.slf4j.Logger;
@@ -245,6 +247,14 @@ public class AccordService implements IAccordService, Shutdownable
         }
         AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmId));
         as.startup();
+        if (StorageService.instance.isReplacingSameAddress())
+        {
+            // when replacing another node but using the same ip the hostId will also match, this causes no TCM transactions
+            // to be committed...
+            // In order to bootup correctly, need to pull in the current epoch
+            ClusterMetadata current = ClusterMetadata.current();
+            as.configurationService().notifyPostCommit(current, current, false);
+        }
         instance = as;
     }
 

src/java/org/apache/cassandra/tcm/RemoteProcessor.java

@@ -93,6 +93,7 @@ public final class RemoteProcessor implements Processor
             {
                 log.waitForHighestConsecutive();
             }
+
             return result;
         }
         catch (Exception e)
@@ -257,7 +258,7 @@ public final class RemoteProcessor implements Processor
             }
 
             @Override
-            public void onFailure(InetAddressAndPort from, RequestFailure failure)
+            public void onFailure(InetAddressAndPort from, RequestFailure failureReason)
             {
                 // "success" - this lets us just try the next one in cmsIter
                 promise.setSuccess(new DiscoveredNodes(Collections.emptySet(), DiscoveredNodes.Kind.KNOWN_PEERS));

test/distributed/org/apache/cassandra/distributed/impl/Instance.java

@@ -916,6 +916,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
     {
         Future<?> future = async((ExecutorService executor) -> {
             Throwable error = null;
+            inInstancelogger.warn("Shutting down in thread {}", Thread.currentThread().getName());
 
             error = parallelRun(error, executor, SnapshotManager.instance::close);
 
@@ -1260,6 +1261,11 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
                 }
             }));
         }
+        // This is not used code, but it is here for when you run in a debugger...
+        // When shutdown gets blocked we need to be able to trace down which future is blocked, so this idx
+        // helps map the location... the reason we can't leverage here is the timeout logic is higher up, so
+        // 'idx' really only helps out in a debugger...
+        int idx = 0;
         for (Future<Throwable> future : results)
         {
             try
@@ -1272,6 +1278,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
             {
                 accumulate = Throwables.merge(accumulate, t);
             }
+            idx++;
         }
         return accumulate;
     }