diff --git a/CHANGES.txt b/CHANGES.txt index 7a83979055..8b59569491 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,7 @@ 4.0.5 * Optimise BTree build,update and transform operations (CASSANDRA-15510) +Merged from 3.0: + * Schema mutations may not be completed on drain (CASSANDRA-17524) 4.0.4 * Clean up schema migration coordinator and tests (CASSANDRA-17533) diff --git a/src/java/org/apache/cassandra/concurrent/Stage.java b/src/java/org/apache/cassandra/concurrent/Stage.java index e00da7b0cc..5efaf16cf6 100644 --- a/src/java/org/apache/cassandra/concurrent/Stage.java +++ b/src/java/org/apache/cassandra/concurrent/Stage.java @@ -49,28 +49,35 @@ import static java.util.stream.Collectors.toMap; public enum Stage { - READ ("ReadStage", "request", DatabaseDescriptor::getConcurrentReaders, DatabaseDescriptor::setConcurrentReaders, Stage::multiThreadedLowSignalStage), - MUTATION ("MutationStage", "request", DatabaseDescriptor::getConcurrentWriters, DatabaseDescriptor::setConcurrentWriters, Stage::multiThreadedLowSignalStage), - COUNTER_MUTATION ("CounterMutationStage", "request", DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage), - VIEW_MUTATION ("ViewMutationStage", "request", DatabaseDescriptor::getConcurrentViewWriters, DatabaseDescriptor::setConcurrentViewWriters, Stage::multiThreadedLowSignalStage), - GOSSIP ("GossipStage", "internal", () -> 1, null, Stage::singleThreadedStage), - REQUEST_RESPONSE ("RequestResponseStage", "request", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedLowSignalStage), - ANTI_ENTROPY ("AntiEntropyStage", "internal", () -> 1, null, Stage::singleThreadedStage), - MIGRATION ("MigrationStage", "internal", () -> 1, null, Stage::singleThreadedStage), - MISC ("MiscStage", "internal", () -> 1, null, Stage::singleThreadedStage), - TRACING ("TracingStage", "internal", () -> 1, null, Stage::tracingExecutor), - INTERNAL_RESPONSE ("InternalResponseStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), - IMMEDIATE ("ImmediateStage", "internal", () -> 0, null, Stage::immediateExecutor); + READ (false, "ReadStage", "request", DatabaseDescriptor::getConcurrentReaders, DatabaseDescriptor::setConcurrentReaders, Stage::multiThreadedLowSignalStage), + MUTATION (true, "MutationStage", "request", DatabaseDescriptor::getConcurrentWriters, DatabaseDescriptor::setConcurrentWriters, Stage::multiThreadedLowSignalStage), + COUNTER_MUTATION (true, "CounterMutationStage", "request", DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage), + VIEW_MUTATION (true, "ViewMutationStage", "request", DatabaseDescriptor::getConcurrentViewWriters, DatabaseDescriptor::setConcurrentViewWriters, Stage::multiThreadedLowSignalStage), + GOSSIP (true, "GossipStage", "internal", () -> 1, null, Stage::singleThreadedStage), + REQUEST_RESPONSE (false, "RequestResponseStage", "request", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedLowSignalStage), + ANTI_ENTROPY (false, "AntiEntropyStage", "internal", () -> 1, null, Stage::singleThreadedStage), + MIGRATION (false, "MigrationStage", "internal", () -> 1, null, Stage::singleThreadedStage), + MISC (false, "MiscStage", "internal", () -> 1, null, Stage::singleThreadedStage), + TRACING (false, "TracingStage", "internal", () -> 1, null, Stage::tracingExecutor), + INTERNAL_RESPONSE (false, "InternalResponseStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage), + IMMEDIATE (false, "ImmediateStage", "internal", () -> 0, null, Stage::immediateExecutor); public static final long KEEP_ALIVE_SECONDS = 60; // seconds to keep "extra" threads alive for when idle public final String jmxName; + /** Set true if this executor should be gracefully shutdown before stopping + * the commitlog allocator. Tasks on executors that issue mutations may + * block indefinitely waiting for a new commitlog segment, preventing a + * clean drain/shutdown. + */ + public final boolean shutdownBeforeCommitlog; private final Supplier initialiser; private volatile LocalAwareExecutorService executor = null; - Stage(String jmxName, String jmxType, IntSupplier numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize, ExecutorServiceInitialiser initialiser) + Stage(Boolean shutdownBeforeCommitlog, String jmxName, String jmxType, IntSupplier numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize, ExecutorServiceInitialiser initialiser) { + this.shutdownBeforeCommitlog = shutdownBeforeCommitlog; this.jmxName = jmxName; - this.initialiser = () -> initialiser.init(jmxName, jmxType, numThreads.getAsInt(), onSetMaximumPoolSize); + this.initialiser = () -> initialiser.init(jmxName,jmxType, numThreads.getAsInt(), onSetMaximumPoolSize); } private static String normalizeName(String stageName) @@ -150,6 +157,14 @@ public enum Stage .collect(Collectors.toList()); } + private static List mutatingExecutors() + { + return Stream.of(Stage.values()) + .filter(stage -> stage.shutdownBeforeCommitlog) + .map(Stage::executor) + .collect(Collectors.toList()); + } + /** * This method shuts down all registered stages. */ @@ -158,6 +173,18 @@ public enum Stage ExecutorUtils.shutdownNow(executors()); } + public static void shutdownAndAwaitMutatingExecutors(boolean interrupt, long timeout, TimeUnit units) throws InterruptedException, TimeoutException + { + List executors = mutatingExecutors(); + ExecutorUtils.shutdown(interrupt, executors); + ExecutorUtils.awaitTermination(timeout, units, executors); + } + + public static boolean areMutationExecutorsTerminated() + { + return mutatingExecutors().stream().allMatch(ExecutorService::isTerminated); + } + @VisibleForTesting public static void shutdownAndWait(long timeout, TimeUnit units) throws InterruptedException, TimeoutException { diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 4afb1ee9e2..9340d9847d 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -18,6 +18,8 @@ package org.apache.cassandra.config; +import java.util.concurrent.TimeUnit; + import org.apache.cassandra.exceptions.ConfigurationException; /** A class that extracts system properties for the cassandra node it runs within. */ @@ -147,6 +149,11 @@ public enum CassandraRelevantProperties */ BOOTSTRAP_SCHEMA_DELAY_MS("cassandra.schema_delay_ms"), + /** + * When draining, how long to wait for mutating executors to shutdown. + */ + DRAIN_EXECUTOR_TIMEOUT_MS("cassandra.drain_executor_timeout_ms", String.valueOf(TimeUnit.MINUTES.toMillis(5))), + /** * Gossip quarantine delay is used while evaluating membership changes and should only be changed with extreme care. */ diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 00d8887ebf..421b62697b 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -125,6 +125,7 @@ import static java.util.stream.Collectors.toList; import static java.util.stream.Collectors.toMap; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK; +import static org.apache.cassandra.config.CassandraRelevantProperties.DRAIN_EXECUTOR_TIMEOUT_MS; import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACEMENT_ALLOW_EMPTY; import static org.apache.cassandra.index.SecondaryIndexManager.getIndexName; import static org.apache.cassandra.index.SecondaryIndexManager.isIndexColumnFamily; @@ -4924,13 +4925,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE protected synchronized void drain(boolean isFinalShutdown) throws IOException, InterruptedException, ExecutionException { - ExecutorService counterMutationStage = Stage.COUNTER_MUTATION.executor(); - ExecutorService viewMutationStage = Stage.VIEW_MUTATION.executor(); - ExecutorService mutationStage = Stage.MUTATION.executor(); - - if (mutationStage.isTerminated() - && counterMutationStage.isTerminated() - && viewMutationStage.isTerminated()) + if (Stage.areMutationExecutorsTerminated()) { if (!isFinalShutdown) logger.warn("Cannot drain node (did it already happen?)"); @@ -4985,14 +4980,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE if (!isFinalShutdown) setMode(Mode.DRAINING, "clearing mutation stage", false); - viewMutationStage.shutdown(); - counterMutationStage.shutdown(); - mutationStage.shutdown(); - - // FIXME? should these *really* take up to one hour? - viewMutationStage.awaitTermination(3600, TimeUnit.SECONDS); - counterMutationStage.awaitTermination(3600, TimeUnit.SECONDS); - mutationStage.awaitTermination(3600, TimeUnit.SECONDS); + Stage.shutdownAndAwaitMutatingExecutors(false, + DRAIN_EXECUTOR_TIMEOUT_MS.getInt(), TimeUnit.MILLISECONDS); StorageProxy.instance.verifyNoHintsInProgress(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/AutoBootstrapTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/AutoBootstrapTest.java index dc754ca6a2..3051639dd8 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/AutoBootstrapTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/AutoBootstrapTest.java @@ -26,11 +26,14 @@ import org.junit.Test; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.TokenSupplier; import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.distributed.test.TestBaseImpl; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.test.ring.BootstrapTest.count; +import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate; -public class AutoBootstrapTest extends BootstrapTest +public class AutoBootstrapTest extends TestBaseImpl { // Originally part of BootstrapTest. Broken out into separate test as the in-JVM dtests fail // if too many instances are created in the same JVM. Bug in the JVM is suspected.