diff --git a/src/java/org/apache/cassandra/concurrent/Shutdownable.java b/src/java/org/apache/cassandra/concurrent/Shutdownable.java index 185875b791..a72253fc87 100644 --- a/src/java/org/apache/cassandra/concurrent/Shutdownable.java +++ b/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); + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 04ef7e6355..a42f52aefe 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/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; } diff --git a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java b/src/java/org/apache/cassandra/tcm/RemoteProcessor.java index 54adbafba6..ed10512a88 100644 --- a/src/java/org/apache/cassandra/tcm/RemoteProcessor.java +++ b/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)); diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index c703310101..43e4749ba4 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/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 future : results) { try @@ -1272,6 +1278,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance { accumulate = Throwables.merge(accumulate, t); } + idx++; } return accumulate; }