diff --git a/CHANGES.txt b/CHANGES.txt index 8a52c9b32e..d450f81726 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -133,6 +133,7 @@ * GossiperTest.testHasVersion3Nodes didn't take into account trunk version changes, fixed to rely on latest version (CASSANDRA-16651) * Update JNA library to 5.9.0 and snappy-java to version 1.1.8.4 (CASSANDRA-17040) Merged from 4.0: + * Clean up schema migration coordinator and tests (CASSANDRA-17533) * Shut repair task executor down without interruption to avoid compromising shared channel proxies (CASSANDRA-17466) * Generate valid KEYSPACE / MATERIALIZED VIEW for CQL for views (CASSANDRA-17266) * Fix timestamp tz parsing (CASSANDRA-17467) diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index b6bcebb7bf..3b63cf8d03 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -158,6 +158,10 @@ public enum CassandraRelevantProperties GOSSIPER_SKIP_WAITING_TO_SETTLE("cassandra.skip_wait_for_gossip_to_settle", "-1"), + IGNORED_SCHEMA_CHECK_VERSIONS("cassandra.skip_schema_check_for_versions"), + + IGNORED_SCHEMA_CHECK_ENDPOINTS("cassandra.skip_schema_check_for_endpoints"), + SHUTDOWN_ANNOUNCE_DELAY_IN_MS("cassandra.shutdown_announce_in_ms", "2000"), /** diff --git a/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java b/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java index 437f48918d..1ccecc60fc 100644 --- a/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java +++ b/src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java @@ -116,7 +116,8 @@ public class DefaultSchemaUpdateHandler implements SchemaUpdateHandler, IEndpoin Schema.instance.getVersion(), migrationCoordinator.outstandingVersions(), CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK.getKey(), - MigrationCoordinator.IGNORED_ENDPOINTS_PROP, MigrationCoordinator.IGNORED_VERSIONS_PROP); + CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_ENDPOINTS.getKey(), + CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_VERSIONS.getKey()); if (requireSchemas) { diff --git a/src/java/org/apache/cassandra/schema/MigrationCoordinator.java b/src/java/org/apache/cassandra/schema/MigrationCoordinator.java index 54da2ed783..e3153e1e6f 100644 --- a/src/java/org/apache/cassandra/schema/MigrationCoordinator.java +++ b/src/java/org/apache/cassandra/schema/MigrationCoordinator.java @@ -35,9 +35,11 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiConsumer; +import java.util.function.LongSupplier; import java.util.function.Supplier; import com.google.common.annotations.VisibleForTesting; @@ -71,6 +73,8 @@ import org.apache.cassandra.utils.concurrent.Future; import org.apache.cassandra.utils.concurrent.ImmediateFuture; import org.apache.cassandra.utils.concurrent.WaitQueue; +import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_ENDPOINTS; +import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_VERSIONS; import static org.apache.cassandra.net.Verb.SCHEMA_PUSH_REQ; import static org.apache.cassandra.utils.Clock.Global.nanoTime; import static org.apache.cassandra.utils.Simulate.With.MONITORS; @@ -90,15 +94,20 @@ public class MigrationCoordinator private static final Logger logger = LoggerFactory.getLogger(MigrationCoordinator.class); private static final Future FINISHED_FUTURE = ImmediateFuture.success(null); + private static LongSupplier getUptimeFn = () -> ManagementFactory.getRuntimeMXBean().getUptime(); + + @VisibleForTesting + public static void setUptimeFn(LongSupplier supplier) + { + getUptimeFn = supplier; + } + private static final int MIGRATION_DELAY_IN_MS = CassandraRelevantProperties.MIGRATION_DELAY.getInt(); public static final int MAX_OUTSTANDING_VERSION_REQUESTS = 3; - public static final String IGNORED_VERSIONS_PROP = "cassandra.skip_schema_check_for_versions"; - public static final String IGNORED_ENDPOINTS_PROP = "cassandra.skip_schema_check_for_endpoints"; - private static ImmutableSet getIgnoredVersions() { - String s = System.getProperty(IGNORED_VERSIONS_PROP); + String s = IGNORED_SCHEMA_CHECK_VERSIONS.getString(); if (s == null || s.isEmpty()) return ImmutableSet.of(); @@ -117,7 +126,7 @@ public class MigrationCoordinator { Set endpoints = new HashSet<>(); - String s = System.getProperty(IGNORED_ENDPOINTS_PROP); + String s = IGNORED_SCHEMA_CHECK_ENDPOINTS.getString(); if (s == null || s.isEmpty()) return endpoints; @@ -335,7 +344,7 @@ public class MigrationCoordinator private boolean shouldPullImmediately(InetAddressAndPort endpoint, UUID version) { UUID localSchemaVersion = schemaVersion.get(); - if (SchemaConstants.emptyVersion.equals(localSchemaVersion) || ManagementFactory.getRuntimeMXBean().getUptime() < MIGRATION_DELAY_IN_MS) + if (SchemaConstants.emptyVersion.equals(localSchemaVersion) || getUptimeFn.getAsLong() < MIGRATION_DELAY_IN_MS) { // If we think we may be bootstrapping or have recently started, submit MigrationTask immediately logger.debug("Immediately submitting migration task for {}, " + @@ -467,16 +476,29 @@ public class MigrationCoordinator SchemaDiagnostics.versionAnnounced(Schema.instance); } - private Future submitToMigrationIfNotShutdown(Runnable task) + private Future submitToMigrationIfNotShutdown(Runnable task) { - if (executor.isShutdown() || executor.isTerminated()) + boolean skipped = false; + try { - logger.info("Skipped scheduled pulling schema from other nodes: the MIGRATION executor service has been shutdown."); + if (executor.isShutdown() || executor.isTerminated()) + { + skipped = true; + return ImmediateFuture.success(null); + } + return executor.submit(task); + } + catch (RejectedExecutionException ex) + { + skipped = true; return ImmediateFuture.success(null); } - else + finally { - return executor.submit(task, null); + if (skipped) + { + logger.info("Skipped scheduled pulling schema from other nodes: the MIGRATION executor service has been shutdown."); + } } } diff --git a/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java b/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java index fc377a65c0..d6aee7cc23 100644 --- a/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java +++ b/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java @@ -51,6 +51,7 @@ import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.distributed.impl.DistributedTestSnitch.toCassandraInetAddressAndPort; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; +import static org.junit.Assert.assertTrue; public class GossipHelper { @@ -228,7 +229,7 @@ public class GossipHelper InetAddressAndPort endpoint = toCassandraInetAddressAndPort(pullFrom); EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint); Gossiper.instance.doOnChangeNotifications(endpoint, ApplicationState.SCHEMA, state.getApplicationState(ApplicationState.SCHEMA)); - Schema.instance.waitUntilReady(Duration.ofSeconds(10)); + assertTrue("schema is ready", Schema.instance.waitUntilReady(Duration.ofSeconds(10))); }).accept(pullFrom); } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 90bf1fb535..e283bb6476 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -35,7 +35,9 @@ import java.util.UUID; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Stream; import javax.management.ListenerNotFoundException; import javax.management.Notification; @@ -104,6 +106,7 @@ import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.NoPayload; import org.apache.cassandra.net.Verb; +import org.apache.cassandra.schema.MigrationCoordinator; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.ActiveRepairService; @@ -159,7 +162,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance { public final IInstanceConfig config; private volatile boolean initialized = false; - private final long startedAt; + private final AtomicLong startedAt = new AtomicLong(); @Deprecated Instance(IInstanceConfig config, ClassLoader classLoader) @@ -181,11 +184,6 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance Object clusterId = Objects.requireNonNull(config.get(Constants.KEY_DTEST_API_CLUSTER_ID), "cluster_id is not defined"); ClusterIDDefiner.setId("cluster-" + clusterId); InstanceIDDefiner.setInstanceId(config.num()); - // Defer initialisation of Clock.Global until cluster/instance identifiers are set. - // Otherwise, the instance classloader's logging classes are setup ahead of time and - // the patterns/file paths are not set correctly. This will be addressed in a subsequent - // commit to extend the functionality of the @Shared annotation to app classes. - startedAt = nanoTime(); FBUtilities.setBroadcastInetAddressAndPort(InetAddressAndPort.getByAddressOverrideDefaults(config.broadcastAddress().getAddress(), config.broadcastAddress().getPort())); @@ -538,6 +536,12 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance @Override public void startup(ICluster cluster) { + // Defer initialisation of Clock.Global until cluster/instance identifiers are set. + // Otherwise, the instance classloader's logging classes are setup ahead of time and + // the patterns/file paths are not set correctly. This will be addressed in a subsequent + // commit to extend the functionality of the @Shared annotation to app classes. + assert startedAt.compareAndSet(0L, System.nanoTime()) : "startedAt uninitialized"; + sync(() -> { try { @@ -631,6 +635,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance StorageService.instance.registerDaemon(CassandraDaemon.getInstanceForTesting()); if (config.has(GOSSIP)) { + MigrationCoordinator.setUptimeFn(() -> TimeUnit.NANOSECONDS.toMillis(nanoTime() - startedAt.get())); try { StorageService.instance.initServer(); @@ -812,6 +817,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance finally { super.shutdown(); + startedAt.set(0L); } }); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/MigrationCoordinatorTest.java b/test/distributed/org/apache/cassandra/distributed/test/MigrationCoordinatorTest.java index cf895ecdec..ca89b43d0c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/MigrationCoordinatorTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/MigrationCoordinatorTest.java @@ -28,9 +28,10 @@ import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.api.TokenSupplier; import org.apache.cassandra.distributed.shared.NetworkTopology; -import org.apache.cassandra.schema.MigrationCoordinator; import org.apache.cassandra.schema.Schema; +import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_ENDPOINTS; +import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_VERSIONS; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; @@ -42,8 +43,9 @@ public class MigrationCoordinatorTest extends TestBaseImpl { System.clearProperty("cassandra.replace_address"); System.clearProperty("cassandra.consistent.rangemovement"); - System.clearProperty(MigrationCoordinator.IGNORED_ENDPOINTS_PROP); - System.clearProperty(MigrationCoordinator.IGNORED_VERSIONS_PROP); + + System.clearProperty(IGNORED_SCHEMA_CHECK_VERSIONS.getKey()); + System.clearProperty(IGNORED_SCHEMA_CHECK_VERSIONS.getKey()); } /** * We shouldn't wait on versions only available from a node being replaced @@ -86,7 +88,7 @@ public class MigrationCoordinatorTest extends TestBaseImpl IInstanceConfig config = cluster.newInstanceConfig(); config.set("auto_bootstrap", true); - System.setProperty(MigrationCoordinator.IGNORED_ENDPOINTS_PROP, ignoredEndpoint.getHostAddress()); + IGNORED_SCHEMA_CHECK_ENDPOINTS.setString(ignoredEndpoint.getHostAddress()); System.setProperty("cassandra.consistent.rangemovement", "false"); cluster.bootstrap(config).startup(); } @@ -113,7 +115,7 @@ public class MigrationCoordinatorTest extends TestBaseImpl IInstanceConfig config = cluster.newInstanceConfig(); config.set("auto_bootstrap", true); - System.setProperty(MigrationCoordinator.IGNORED_VERSIONS_PROP, initialVersion.toString() + ',' + oldVersion.toString()); + IGNORED_SCHEMA_CHECK_VERSIONS.setString(initialVersion.toString() + ',' + oldVersion); System.setProperty("cassandra.consistent.rangemovement", "false"); cluster.bootstrap(config).startup(); } diff --git a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceSyncSchemaForBootstrap.java b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceSyncSchemaForBootstrap.java index bf10db2d0a..c8bc9f741e 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceSyncSchemaForBootstrap.java +++ b/test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceSyncSchemaForBootstrap.java @@ -25,12 +25,13 @@ import org.apache.cassandra.simulator.systems.SimulatedActionTask; import static org.apache.cassandra.simulator.Action.Modifier.DISPLAY_ORIGIN; import static org.apache.cassandra.simulator.Action.Modifiers.RELIABLE_NO_TIMEOUTS; +import static org.junit.Assert.assertTrue; class OnInstanceSyncSchemaForBootstrap extends SimulatedActionTask { public OnInstanceSyncSchemaForBootstrap(ClusterActions actions, int node) { super("Sync Schema on " + node, RELIABLE_NO_TIMEOUTS.with(DISPLAY_ORIGIN), RELIABLE_NO_TIMEOUTS, actions, actions.cluster.get(node), - () -> Schema.instance.waitUntilReady(Duration.ofMinutes(10))); + () -> assertTrue("schema is ready", Schema.instance.waitUntilReady(Duration.ofMinutes(10)))); } }