From 76be530a364b376c1d69d8447230ad5cf023be7f Mon Sep 17 00:00:00 2001 From: Jacek Lewandowski Date: Thu, 8 Sep 2022 16:02:50 +0200 Subject: [PATCH] Fix schema reset functionality MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Patch by Jacek Lewandowski, reviewed by Andrés de la Peña and Ekaterina Dimitrova for CASSANDRA-17819 --- CHANGES.txt | 1 + .../config/CassandraRelevantProperties.java | 12 +- .../schema/MigrationCoordinator.java | 148 ++++++++++++++++-- .../cassandra/schema/MigrationManager.java | 4 + .../cassandra/service/StorageService.java | 2 +- .../distributed/test/SchemaTest.java | 67 ++++++++ 6 files changed, 220 insertions(+), 14 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 27d89bbd3a..c78d97d61c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -7,6 +7,7 @@ Merged from 3.11: * Fix potential IndexOutOfBoundsException in PagingState in mixed mode clusters (CASSANDRA-17840) Merged from 3.0: * Fix scrubber falling into infinite loop when the last partition is broken (CASSANDRA-17862) + * Fix resetting schema (CASSANDRA-17819) 4.0.6 * Fix race condition on updating cdc size and advancing to next segment (CASSANDRA-17792) diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 9340d9847d..b725d7f9b8 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -194,7 +194,17 @@ public enum CassandraRelevantProperties IS_DISABLED_MBEAN_REGISTRATION("org.apache.cassandra.disable_mbean_registration"), /** what class to use for mbean registeration */ - MBEAN_REGISTRATION_CLASS("org.apache.cassandra.mbean_registration_class"); + MBEAN_REGISTRATION_CLASS("org.apache.cassandra.mbean_registration_class"), + + MIGRATION_DELAY("cassandra.migration_delay_ms", "60000"), + /** Defines how often schema definitions are pulled from the other nodes */ + SCHEMA_PULL_INTERVAL_MS("cassandra.schema_pull_interval_ms", "60000"), + /** + * Minimum delay after a failed pull request before it is reattempted. It prevents reattempting failed requests + * immediately as it is high chance they will fail anyway. It is better to wait a bit instead of flooding logs + * and wasting resources. + */ + SCHEMA_PULL_BACKOFF_DELAY_MS("cassandra.schema_pull_backoff_delay_ms", "3000"); CassandraRelevantProperties(String key, String defaultVal) { diff --git a/src/java/org/apache/cassandra/schema/MigrationCoordinator.java b/src/java/org/apache/cassandra/schema/MigrationCoordinator.java index e04f828574..b2f4316ec3 100644 --- a/src/java/org/apache/cassandra/schema/MigrationCoordinator.java +++ b/src/java/org/apache/cassandra/schema/MigrationCoordinator.java @@ -26,10 +26,12 @@ import java.util.Collection; import java.util.Deque; import java.util.HashMap; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.WeakHashMap; import java.util.concurrent.Future; import java.util.concurrent.FutureTask; import java.util.concurrent.RejectedExecutionException; @@ -47,6 +49,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.exceptions.RequestFailureReason; import org.apache.cassandra.gms.ApplicationState; @@ -66,6 +69,15 @@ 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; +/** + * Migration coordinator is responsible for tracking schema versions on various nodes and, if needed, synchronize the + * schema. It performs periodic checks and if there is a schema version mismatch between the current node and the other + * node, it pulls the schema and applies the changes locally through the callback. + * + * In particular the Migration Coordinator keeps track of all schema versions reported from each node in the cluster. + * As long as a certain version is advertised by some node, it is being tracked. As long as a version is tracked, + * the migration coordinator tries to fetch it by its periodic job. + */ public class MigrationCoordinator { private static final Logger logger = LoggerFactory.getLogger(MigrationCoordinator.class); @@ -80,11 +92,21 @@ public class MigrationCoordinator } - private static final int MIGRATION_DELAY_IN_MS = 60000; + private static final int MIGRATION_DELAY_IN_MS = CassandraRelevantProperties.MIGRATION_DELAY.getInt(); private static final int MAX_OUTSTANDING_VERSION_REQUESTS = 3; public static final MigrationCoordinator instance = new MigrationCoordinator(); + /** + * @see CassandraRelevantProperties#SCHEMA_PULL_BACKOFF_DELAY_MS + */ + private static final long BACKOFF_DELAY_MS = CassandraRelevantProperties.SCHEMA_PULL_BACKOFF_DELAY_MS.getInt(); + + /** + * Holds the timestamps in ms for last pull request attempts. + */ + private final WeakHashMap lastPullAttemptTimestamps = new WeakHashMap<>(); + private static ImmutableSet getIgnoredVersions() { String s = IGNORED_SCHEMA_CHECK_VERSIONS.getString(); @@ -129,12 +151,27 @@ public class MigrationCoordinator { final UUID version; + /** + * The set of endpoints containing this schema version + */ final Set endpoints = Sets.newConcurrentHashSet(); + /** + * The set of endpoints from which we are already fetching the schema + */ final Set outstandingRequests = Sets.newConcurrentHashSet(); + /** + * The queue of endpoints from which we are going to fetch the schema + */ final Deque requestQueue = new ArrayDeque<>(); + /** + * Threads waiting for schema synchronization are waiting until this object is signalled + */ private final WaitQueue waitQueue = new WaitQueue(); + /** + * Whether this schema version have been received + */ volatile boolean receivedSchema; VersionInfo(UUID version) @@ -160,6 +197,18 @@ public class MigrationCoordinator { return receivedSchema; } + + @Override + public String toString() + { + return "VersionInfo{" + + "version=" + version + + ", outstandingRequests=" + outstandingRequests + + ", requestQueue=" + requestQueue + + ", waitQueue.waiting=" + waitQueue.getWaiting() + + ", receivedSchema=" + receivedSchema + + '}'; + } } private final Map versionInfo = new HashMap<>(); @@ -169,12 +218,20 @@ public class MigrationCoordinator public void start() { - ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(this::pullUnreceivedSchemaVersions, 1, 1, TimeUnit.MINUTES); + int interval = CassandraRelevantProperties.SCHEMA_PULL_INTERVAL_MS.getInt(); + ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(this::pullUnreceivedSchemaVersions, interval, interval, TimeUnit.MILLISECONDS); } + /** + * Resets the migration coordinator by notifying all waiting threads and removing all the existing version info. + */ public synchronized void reset() { - versionInfo.clear(); + logger.info("Resetting migration coordinator..."); + + // clear all the managed information + this.endpointVersions.clear(); + clearVersionsInfo(); } synchronized List> pullUnreceivedSchemaVersions() @@ -183,7 +240,10 @@ public class MigrationCoordinator for (VersionInfo info : versionInfo.values()) { if (info.wasReceived() || info.outstandingRequests.size() > 0) + { + logger.trace("Skipping pull of schema {} because it has been already recevied, or it is being received ({})", info.version, info); continue; + } Future future = maybePullSchema(info); if (future != null && future != FINISHED_FUTURE) @@ -196,16 +256,25 @@ public class MigrationCoordinator synchronized Future maybePullSchema(VersionInfo info) { if (info.endpoints.isEmpty() || info.wasReceived() || !shouldPullSchema(info.version)) + { + logger.trace("Not pulling schema {} because it was received, there is no endpoint to provide it, or we should not pull it ({})", info.version, info); return FINISHED_FUTURE; + } if (info.outstandingRequests.size() >= getMaxOutstandingVersionRequests()) + { + logger.trace("Not pulling schema {} because the number of outstanding requests has been exceeded ({} >= {})", info.version, info.outstandingRequests.size(), getMaxOutstandingVersionRequests()); return FINISHED_FUTURE; + } for (int i=0, isize=info.requestQueue.size(); i reportEndpointVersion(InetAddressAndPort endpoint, UUID version) { + logger.debug("Reported schema {} at endpoint {}", version, endpoint); if (ignoredEndpoints.contains(endpoint) || IGNORED_VERSIONS.contains(version)) { endpointVersions.remove(endpoint); removeEndpointFromVersion(endpoint, null); + logger.debug("Discarding endpoint {} or schema {} because either endpoint or schema version were marked as ignored", endpoint, version); return FINISHED_FUTURE; } UUID current = endpointVersions.put(endpoint, version); if (current != null && current.equals(version)) + { + logger.trace("Skipping report of schema {} from {} because we already know that", version, endpoint); return FINISHED_FUTURE; + } VersionInfo info = versionInfo.computeIfAbsent(version, VersionInfo::new); if (isLocalVersion(version)) + { info.markReceived(); + logger.trace("Schema {} from {} has been marked as recevied because it is equal the local schema", version, endpoint); + } + else + { + info.requestQueue.addFirst(endpoint); + } info.endpoints.add(endpoint); - info.requestQueue.addFirst(endpoint); + logger.trace("Added endpoint {} to schema {}: {}", endpoint, info.version, info); // disassociate this endpoint from its (now) previous schema version removeEndpointFromVersion(endpoint, current); @@ -401,16 +486,36 @@ public class MigrationCoordinator return; info.endpoints.remove(endpoint); + logger.trace("Removed endpoint {} from schema {}: {}", endpoint, version, info); if (info.endpoints.isEmpty()) { info.waitQueue.signalAll(); versionInfo.remove(version); + logger.trace("Removed schema info: {}", info); + } + } + + /** + * Remove all version info and signal all the waiting entities. + */ + private synchronized void clearVersionsInfo() + { + Iterator> it = versionInfo.entrySet().iterator(); + while (it.hasNext()) + { + Map.Entry entry = it.next(); + it.remove(); + entry.getValue().waitQueue.signalAll(); } } public synchronized void removeAndIgnoreEndpoint(InetAddressAndPort endpoint) { + logger.debug("Removing and ignoring endpoint {}", endpoint); Preconditions.checkArgument(endpoint != null); + // TODO The endpoint address is now ignored but when a node with the same address is added again later, + // there will be no way to include it in schema synchronization other than restarting each other node + // see https://issues.apache.org/jira/browse/CASSANDRA-17883 for details ignoredEndpoints.add(endpoint); Set versions = ImmutableSet.copyOf(versionInfo.keySet()); for (UUID version : versions) @@ -424,12 +529,26 @@ public class MigrationCoordinator FutureTask task = new FutureTask<>(() -> pullSchema(new Callback(endpoint, info)), null); if (shouldPullImmediately(endpoint, info.version)) { - submitToMigrationIfNotShutdown(task); + long nextAttempt = lastPullAttemptTimestamps.getOrDefault(endpoint, 0L) + BACKOFF_DELAY_MS; + long now = System.currentTimeMillis(); + if (nextAttempt <= now) + { + logger.debug("Pulling {} immediately from {}", info, endpoint); + submitToMigrationIfNotShutdown(task); + } + else + { + long delay = nextAttempt - now; + logger.debug("Previous pull of {} from {} failed. Postponing next attempt for {}ms", info, endpoint, delay); + ScheduledExecutors.nonPeriodicTasks.schedule(() -> submitToMigrationIfNotShutdown(task), delay, TimeUnit.MILLISECONDS); + } } else { - ScheduledExecutors.nonPeriodicTasks.schedule(()->submitToMigrationIfNotShutdown(task), MIGRATION_DELAY_IN_MS, TimeUnit.MILLISECONDS); + logger.debug("Postponing pull of {} from {} for {}ms", info, endpoint, MIGRATION_DELAY_IN_MS); + ScheduledExecutors.nonPeriodicTasks.schedule(() -> submitToMigrationIfNotShutdown(task), MIGRATION_DELAY_IN_MS, TimeUnit.MILLISECONDS); } + return task; } @@ -519,6 +638,8 @@ public class MigrationCoordinator private void pullSchema(Callback callback) { + lastPullAttemptTimestamps.put(callback.endpoint, System.currentTimeMillis()); + if (!isAlive(callback.endpoint)) { logger.warn("Can't send schema pull request: node {} is down.", callback.endpoint); @@ -552,7 +673,10 @@ public class MigrationCoordinator { inflightTasks.decrementAndGet(); if (wasSuccessful) + { info.markReceived(); + lastPullAttemptTimestamps.remove(endpoint); + } info.outstandingRequests.remove(endpoint); info.requestQueue.add(endpoint); diff --git a/src/java/org/apache/cassandra/schema/MigrationManager.java b/src/java/org/apache/cassandra/schema/MigrationManager.java index efe7c33532..600de9a47f 100644 --- a/src/java/org/apache/cassandra/schema/MigrationManager.java +++ b/src/java/org/apache/cassandra/schema/MigrationManager.java @@ -260,6 +260,10 @@ public class MigrationManager Schema.instance.clear(); + // clean the all version information from the MigrationCoordinator + MigrationCoordinator.instance.reset(); + + // now report again the versions we are aware of Set liveEndpoints = Gossiper.instance.getLiveMembers(); liveEndpoints.remove(FBUtilities.getBroadcastAddressAndPort()); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 4721113ad3..7e70e4ce20 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -5636,7 +5636,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE ColumnFamilyStore.rebuildSecondaryIndex(ksName, cfName, indices); } - public void resetLocalSchema() throws IOException + public void resetLocalSchema() { MigrationManager.resetLocalSchema(); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/SchemaTest.java b/test/distributed/org/apache/cassandra/distributed/test/SchemaTest.java index a2ce32f6f3..bc5ccbe203 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/SchemaTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/SchemaTest.java @@ -18,11 +18,23 @@ package org.apache.cassandra.distributed.test; +import java.util.concurrent.TimeUnit; + +import com.google.common.util.concurrent.Uninterruptibles; import org.junit.Test; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.FBUtilities; +import org.awaitility.Awaitility; +import static java.time.Duration.ofMillis; +import static java.time.Duration.ofSeconds; import static org.junit.Assert.assertTrue; public class SchemaTest extends TestBaseImpl @@ -86,4 +98,59 @@ public class SchemaTest extends TestBaseImpl assertTrue(causeIsUnknownColumn); } } + + @Test + public void schemaReset() throws Throwable + { + int delayUnit = 1000; + CassandraRelevantProperties.MIGRATION_DELAY.setInt(5 * delayUnit); + CassandraRelevantProperties.SCHEMA_PULL_INTERVAL_MS.setInt(5 * delayUnit); + CassandraRelevantProperties.SCHEMA_PULL_BACKOFF_DELAY_MS.setInt(delayUnit); + + try (Cluster cluster = init(Cluster.build(2).withConfig(cfg -> cfg.with(Feature.GOSSIP, Feature.NETWORK)).start())) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk INT PRIMARY KEY, v TEXT)"); + + assertTrue(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null)); + assertTrue(cluster.get(2).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null)); + + cluster.get(2).shutdown().get(); + + Awaitility.await() + .atMost(ofSeconds(30)) + .until(() -> cluster + .get(1) + .callOnInstance(() -> Gossiper.instance + .getLiveMembers() + .stream() + .allMatch(addr -> addr.equals(FBUtilities.getBroadcastAddressAndPort())))); + + // when schema is removed and there is no other node to fetch it from, node 1 should be left with clean schema + //noinspection Convert2MethodRef + cluster.get(1).runOnInstance(() -> StorageService.instance.resetLocalSchema()); + assertTrue(cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") == null)); + + // sleep slightly longer than the schema pull interval + Uninterruptibles.sleepUninterruptibly(6 * delayUnit, TimeUnit.MILLISECONDS); + + // when the other node is started, schema should be back in sync - node 2 should send schema mutations to node 1 + cluster.get(2).startup(); + + // sleep slightly longer than the schema pull interval + Uninterruptibles.sleepUninterruptibly(6 * delayUnit, TimeUnit.MILLISECONDS); + + Awaitility.waitAtMost(ofMillis(6 * delayUnit)) + .pollDelay(ofSeconds(1)) + .until(() -> cluster.get(1).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null)); + + // when schema is removed and there is a node to fetch it from, node 1 should immediately restore the schema + //noinspection Convert2MethodRef + cluster.get(2).runOnInstance(() -> StorageService.instance.resetLocalSchema()); + + Awaitility.waitAtMost(ofMillis(6 * delayUnit)) + .pollDelay(ofSeconds(1)) + .until(() -> cluster.get(2).callOnInstance(() -> Schema.instance.getTableMetadata(KEYSPACE, "tbl") != null)); + } + } + }