You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2020/11/09 20:25:37 UTC
[cassandra] branch trunk updated (85f69cf -> 159e021)
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git.
from 85f69cf Add a ratelimiter to snapshot creation and deletion
new a960d87 add schema coordinator
new 0845008 Wait for schema agreement when bootstrapping
new fe70155 Merge branch 'cassandra-3.0' into cassandra-3.11
new 159e021 Merge branch 'cassandra-3.11' into trunk
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 1 +
.../cassandra/schema/MigrationCoordinator.java | 516 +++++++++++++++++++++
.../apache/cassandra/schema/MigrationManager.java | 149 +-----
.../org/apache/cassandra/schema/MigrationTask.java | 111 -----
src/java/org/apache/cassandra/schema/Schema.java | 2 +-
.../cassandra/schema/SchemaMigrationEvent.java | 7 +-
.../apache/cassandra/service/StorageService.java | 63 ++-
.../cassandra/service/StorageServiceMBean.java | 7 +
.../cassandra/distributed/action/GossipHelper.java | 10 +-
.../cassandra/schema/MigrationCoordinatorTest.java | 319 +++++++++++++
10 files changed, 908 insertions(+), 277 deletions(-)
create mode 100644 src/java/org/apache/cassandra/schema/MigrationCoordinator.java
delete mode 100644 src/java/org/apache/cassandra/schema/MigrationTask.java
create mode 100644 test/unit/org/apache/cassandra/schema/MigrationCoordinatorTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch 'cassandra-3.11' into trunk
Posted by bd...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 159e021aa36a44cdc7674dd234d4a6e189478280
Merge: 85f69cf fe70155
Author: Blake Eggleston <bd...@gmail.com>
AuthorDate: Mon Nov 9 12:22:43 2020 -0800
Merge branch 'cassandra-3.11' into trunk
CHANGES.txt | 1 +
.../cassandra/schema/MigrationCoordinator.java | 516 +++++++++++++++++++++
.../apache/cassandra/schema/MigrationManager.java | 149 +-----
.../org/apache/cassandra/schema/MigrationTask.java | 111 -----
src/java/org/apache/cassandra/schema/Schema.java | 2 +-
.../cassandra/schema/SchemaMigrationEvent.java | 7 +-
.../apache/cassandra/service/StorageService.java | 63 ++-
.../cassandra/service/StorageServiceMBean.java | 7 +
.../cassandra/distributed/action/GossipHelper.java | 10 +-
.../cassandra/schema/MigrationCoordinatorTest.java | 319 +++++++++++++
10 files changed, 908 insertions(+), 277 deletions(-)
diff --cc CHANGES.txt
index f90ffd4,b81b0c8..1b08837
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,41 -1,10 +1,42 @@@
-3.11.10
+4.0-beta4
+ * Add a ratelimiter to snapshot creation and deletion (CASSANDRA-13019)
+ * Produce consistent tombstone for reads to avoid digest mistmatch (CASSANDRA-15369)
+ * Fix SSTableloader issue when restoring a table named backups (CASSANDRA-16235)
+ * Invalid serialized size for responses caused by increasing message time by 1ms which caused extra bytes in size calculation (CASSANDRA-16103)
+ * Throw BufferOverflowException from DataOutputBuffer for better visibility (CASSANDRA-16214)
+ * TLS connections to the storage port on a node without server encryption configured causes java.io.IOException accessing missing keystore (CASSANDRA-16144)
+Merged from 3.11:
Merged from 3.0:
- * Fix invalid cell value skipping when reading from disk (CASSANDRA-16223)
++ * Wait for schema agreement when bootstrapping (CASSANDRA-15158)
* Prevent invoking enable/disable gossip when not in NORMAL (CASSANDRA-16146)
-3.11.9
- * Synchronize Keyspace instance store/clear (CASSANDRA-16210)
+4.0-beta3
+ * Segregate Network and Chunk Cache BufferPools and Recirculate Partially Freed Chunks (CASSANDRA-15229)
+ * Fail truncation requests when they fail on a replica (CASSANDRA-16208)
+ * Move compact storage validation earlier in startup process (CASSANDRA-16063)
+ * Fix ByteBufferAccessor cast exceptions are thrown when trying to query a virtual table (CASSANDRA-16155)
+ * Consolidate node liveness check for forced repair (CASSANDRA-16113)
+ * Use unsigned short in ValueAccessor.sliceWithShortLength (CASSANDRA-16147)
+ * Abort repairs when getting a truncation request (CASSANDRA-15854)
+ * Remove bad assert when getting active compactions for an sstable (CASSANDRA-15457)
+ * Avoid failing compactions with very large partitions (CASSANDRA-15164)
+ * Prevent NPE in StreamMessage in type lookup (CASSANDRA-16131)
+ * Avoid invalid state transition exception during incremental repair (CASSANDRA-16067)
+ * Allow zero padding in timestamp serialization (CASSANDRA-16105)
+ * Add byte array backed cells (CASSANDRA-15393)
+ * Correctly handle pending ranges with adjacent range movements (CASSANDRA-14801)
+ * Avoid adding locahost when streaming trivial ranges (CASSANDRA-16099)
+ * Add nodetool getfullquerylog (CASSANDRA-15988)
+ * Fix yaml format and alignment in tpstats (CASSANDRA-11402)
+ * Avoid trying to keep track of RTs for endpoints we won't write to during read repair (CASSANDRA-16084)
+ * When compaction gets interrupted, the exception should include the compactionId (CASSANDRA-15954)
+ * Make Table/Keyspace Metric Names Consistent With Each Other (CASSANDRA-15909)
+ * Mutating sstable component may race with entire-sstable-streaming(ZCS) causing checksum validation failure (CASSANDRA-15861)
+ * NPE thrown while updating speculative execution time if keyspace is removed during task execution (CASSANDRA-15949)
+ * Show the progress of data streaming and index build (CASSANDRA-15406)
+ * Add flag to disable chunk cache and disable by default (CASSANDRA-16036)
+ * Upgrade to snakeyaml >= 1.26 version for CVE-2017-18640 fix (CASSANDRA-16150)
+Merged from 3.11:
* Fix ColumnFilter to avoid querying cells of unselected complex columns (CASSANDRA-15977)
* Fix memory leak in CompressedChunkReader (CASSANDRA-15880)
* Don't attempt value skipping with mixed version cluster (CASSANDRA-15833)
diff --cc src/java/org/apache/cassandra/schema/MigrationCoordinator.java
index 0000000,0000000..9fb348e
new file mode 100644
--- /dev/null
+++ b/src/java/org/apache/cassandra/schema/MigrationCoordinator.java
@@@ -1,0 -1,0 +1,516 @@@
++/*
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++
++package org.apache.cassandra.schema;
++
++import java.lang.management.ManagementFactory;
++import java.lang.management.RuntimeMXBean;
++import java.util.ArrayDeque;
++import java.util.ArrayList;
++import java.util.Collection;
++import java.util.Deque;
++import java.util.HashMap;
++import java.util.List;
++import java.util.Map;
++import java.util.Set;
++import java.util.UUID;
++import java.util.concurrent.Future;
++import java.util.concurrent.FutureTask;
++import java.util.concurrent.TimeUnit;
++import java.util.concurrent.atomic.AtomicInteger;
++import java.util.function.LongSupplier;
++
++import com.google.common.annotations.VisibleForTesting;
++import com.google.common.collect.ImmutableSet;
++import com.google.common.collect.Sets;
++import com.google.common.util.concurrent.Futures;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import org.apache.cassandra.concurrent.ScheduledExecutors;
++import org.apache.cassandra.concurrent.Stage;
++import org.apache.cassandra.db.Mutation;
++import org.apache.cassandra.exceptions.RequestFailureReason;
++import org.apache.cassandra.gms.ApplicationState;
++import org.apache.cassandra.gms.EndpointState;
++import org.apache.cassandra.gms.FailureDetector;
++import org.apache.cassandra.gms.Gossiper;
++import org.apache.cassandra.locator.InetAddressAndPort;
++import org.apache.cassandra.net.Message;
++import org.apache.cassandra.net.MessagingService;
++import org.apache.cassandra.net.NoPayload;
++import org.apache.cassandra.net.RequestCallback;
++import org.apache.cassandra.net.Verb;
++import org.apache.cassandra.utils.FBUtilities;
++import org.apache.cassandra.utils.concurrent.WaitQueue;
++
++public class MigrationCoordinator
++{
++ private static final Logger logger = LoggerFactory.getLogger(MigrationCoordinator.class);
++ private static final Future<Void> FINISHED_FUTURE = Futures.immediateFuture(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 = 60000;
++ private static final int MAX_OUTSTANDING_VERSION_REQUESTS = 3;
++
++ public static final MigrationCoordinator instance = new MigrationCoordinator();
++
++ static class VersionInfo
++ {
++ final UUID version;
++
++ final Set<InetAddressAndPort> endpoints = Sets.newConcurrentHashSet();
++ final Set<InetAddressAndPort> outstandingRequests = Sets.newConcurrentHashSet();
++ final Deque<InetAddressAndPort> requestQueue = new ArrayDeque<>();
++
++ private final WaitQueue waitQueue = new WaitQueue();
++
++ volatile boolean receivedSchema;
++
++ VersionInfo(UUID version)
++ {
++ this.version = version;
++ }
++
++ WaitQueue.Signal register()
++ {
++ return waitQueue.register();
++ }
++
++ void markReceived()
++ {
++ if (receivedSchema)
++ return;
++
++ receivedSchema = true;
++ waitQueue.signalAll();
++ }
++
++ boolean wasReceived()
++ {
++ return receivedSchema;
++ }
++ }
++
++ private final Map<UUID, VersionInfo> versionInfo = new HashMap<>();
++ private final Map<InetAddressAndPort, UUID> endpointVersions = new HashMap<>();
++ private final AtomicInteger inflightTasks = new AtomicInteger();
++
++ public void start()
++ {
++ ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(this::pullUnreceivedSchemaVersions, 1, 1, TimeUnit.MINUTES);
++ }
++
++ public synchronized void reset()
++ {
++ versionInfo.clear();
++ }
++
++ synchronized List<Future<Void>> pullUnreceivedSchemaVersions()
++ {
++ List<Future<Void>> futures = new ArrayList<>();
++ for (VersionInfo info : versionInfo.values())
++ {
++ if (info.wasReceived() || info.outstandingRequests.size() > 0)
++ continue;
++
++ Future<Void> future = maybePullSchema(info);
++ if (future != null && future != FINISHED_FUTURE)
++ futures.add(future);
++ }
++
++ return futures;
++ }
++
++ synchronized Future<Void> maybePullSchema(VersionInfo info)
++ {
++ if (info.endpoints.isEmpty() || info.wasReceived() || !shouldPullSchema(info.version))
++ return FINISHED_FUTURE;
++
++ if (info.outstandingRequests.size() >= getMaxOutstandingVersionRequests())
++ return FINISHED_FUTURE;
++
++ for (int i=0, isize=info.requestQueue.size(); i<isize; i++)
++ {
++ InetAddressAndPort endpoint = info.requestQueue.remove();
++ if (!info.endpoints.contains(endpoint))
++ continue;
++
++ if (shouldPullFromEndpoint(endpoint) && info.outstandingRequests.add(endpoint))
++ {
++ return scheduleSchemaPull(endpoint, info);
++ }
++ else
++ {
++ // return to queue
++ info.requestQueue.offer(endpoint);
++ }
++ }
++
++ // no suitable endpoints were found, check again in a minute, the periodic task will pick it up
++ return null;
++ }
++
++ public synchronized Map<UUID, Set<InetAddressAndPort>> outstandingVersions()
++ {
++ HashMap<UUID, Set<InetAddressAndPort>> map = new HashMap<>();
++ for (VersionInfo info : versionInfo.values())
++ if (!info.wasReceived())
++ map.put(info.version, ImmutableSet.copyOf(info.endpoints));
++ return map;
++ }
++
++ @VisibleForTesting
++ protected VersionInfo getVersionInfoUnsafe(UUID version)
++ {
++ return versionInfo.get(version);
++ }
++
++ @VisibleForTesting
++ protected int getMaxOutstandingVersionRequests()
++ {
++ return MAX_OUTSTANDING_VERSION_REQUESTS;
++ }
++
++ @VisibleForTesting
++ protected boolean isAlive(InetAddressAndPort endpoint)
++ {
++ return FailureDetector.instance.isAlive(endpoint);
++ }
++
++ @VisibleForTesting
++ protected boolean shouldPullSchema(UUID version)
++ {
++ if (Schema.instance.getVersion() == null)
++ {
++ logger.debug("Not pulling schema for version {}, because local schama version is not known yet", version);
++ return false;
++ }
++
++ if (Schema.instance.isSameVersion(version))
++ {
++ logger.debug("Not pulling schema for version {}, because schema versions match: " +
++ "local={}, remote={}",
++ version,
++ Schema.schemaVersionToString(Schema.instance.getVersion()),
++ Schema.schemaVersionToString(version));
++ return false;
++ }
++ return true;
++ }
++
++ // Since 3.0.14 protocol contains only a CASSANDRA-13004 bugfix, it is safe to accept schema changes
++ // from both 3.0 and 3.0.14.
++ private static boolean is30Compatible(int version)
++ {
++ return version == MessagingService.current_version || version == MessagingService.VERSION_3014;
++ }
++
++ @VisibleForTesting
++ protected boolean shouldPullFromEndpoint(InetAddressAndPort endpoint)
++ {
++ if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
++ return false;
++
++ EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
++ if (state == null)
++ return false;
++
++ final String releaseVersion = state.getApplicationState(ApplicationState.RELEASE_VERSION).value;
++ final String ourMajorVersion = FBUtilities.getReleaseVersionMajor();
++
++ if (!releaseVersion.startsWith(ourMajorVersion))
++ {
++ logger.debug("Not pulling schema from {} because release version in Gossip is not major version {}, it is {}",
++ endpoint, ourMajorVersion, releaseVersion);
++ return false;
++ }
++
++ if (!MessagingService.instance().versions.knows(endpoint))
++ {
++ logger.debug("Not pulling schema from {} because their messaging version is unknown", endpoint);
++ return false;
++ }
++
++ if (MessagingService.instance().versions.getRaw(endpoint) != MessagingService.current_version)
++ {
++ logger.debug("Not pulling schema from {} because their schema format is incompatible", endpoint);
++ return false;
++ }
++
++ if (Gossiper.instance.isGossipOnlyMember(endpoint))
++ {
++ logger.debug("Not pulling schema from {} because it's a gossip only member", endpoint);
++ return false;
++ }
++ return true;
++ }
++
++ @VisibleForTesting
++ protected boolean shouldPullImmediately(InetAddressAndPort endpoint, UUID version)
++ {
++ if (Schema.instance.isEmpty() || 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 {}, " +
++ "schema versions: local={}, remote={}",
++ endpoint,
++ Schema.schemaVersionToString(Schema.instance.getVersion()),
++ Schema.schemaVersionToString(version));
++ return true;
++ }
++ return false;
++ }
++
++ @VisibleForTesting
++ protected boolean isLocalVersion(UUID version)
++ {
++ return Schema.instance.isSameVersion(version);
++ }
++
++ /**
++ * If a previous schema update brought our version the same as the incoming schema, don't apply it
++ */
++ synchronized boolean shouldApplySchemaFor(VersionInfo info)
++ {
++ if (info.wasReceived())
++ return false;
++ return !isLocalVersion(info.version);
++ }
++
++ public synchronized Future<Void> reportEndpointVersion(InetAddressAndPort endpoint, UUID version)
++ {
++ UUID current = endpointVersions.put(endpoint, version);
++ if (current != null && current.equals(version))
++ return FINISHED_FUTURE;
++
++ VersionInfo info = versionInfo.computeIfAbsent(version, VersionInfo::new);
++ if (isLocalVersion(version))
++ info.markReceived();
++ info.endpoints.add(endpoint);
++ info.requestQueue.addFirst(endpoint);
++
++ // disassociate this endpoint from its (now) previous schema version
++ removeEndpointFromVersion(endpoint, current);
++
++ return maybePullSchema(info);
++ }
++
++ public Future<Void> reportEndpointVersion(InetAddressAndPort endpoint, EndpointState state)
++ {
++ if (state == null)
++ return FINISHED_FUTURE;
++
++ UUID version = state.getSchemaVersion();
++
++ if (version == null)
++ return FINISHED_FUTURE;
++
++ return reportEndpointVersion(endpoint, version);
++ }
++
++ private synchronized void removeEndpointFromVersion(InetAddressAndPort endpoint, UUID version)
++ {
++ if (version == null)
++ return;
++
++ VersionInfo info = versionInfo.get(version);
++
++ if (info == null)
++ return;
++
++ info.endpoints.remove(endpoint);
++ if (info.endpoints.isEmpty())
++ {
++ info.waitQueue.signalAll();
++ versionInfo.remove(version);
++ }
++ }
++
++ Future<Void> scheduleSchemaPull(InetAddressAndPort endpoint, VersionInfo info)
++ {
++ FutureTask<Void> task = new FutureTask<>(() -> pullSchema(new Callback(endpoint, info)), null);
++ if (shouldPullImmediately(endpoint, info.version))
++ {
++ Stage.MIGRATION.submit(task);
++ }
++ else
++ {
++ ScheduledExecutors.nonPeriodicTasks.schedule(() -> Stage.MIGRATION.submit(task), MIGRATION_DELAY_IN_MS, TimeUnit.MILLISECONDS);
++ }
++ return task;
++ }
++
++ @VisibleForTesting
++ protected void mergeSchemaFrom(InetAddressAndPort endpoint, Collection<Mutation> mutations)
++ {
++ Schema.instance.mergeAndAnnounceVersion(mutations);
++ }
++
++ class Callback implements RequestCallback<Collection<Mutation>>
++ {
++ final InetAddressAndPort endpoint;
++ final VersionInfo info;
++
++ public Callback(InetAddressAndPort endpoint, VersionInfo info)
++ {
++ this.endpoint = endpoint;
++ this.info = info;
++ }
++
++ public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason)
++ {
++ fail();
++ }
++
++ Future<Void> fail()
++ {
++ return pullComplete(endpoint, info, false);
++ }
++
++ public void onResponse(Message<Collection<Mutation>> message)
++ {
++ response(message.payload);
++ }
++
++ Future<Void> response(Collection<Mutation> mutations)
++ {
++ synchronized (info)
++ {
++ if (shouldApplySchemaFor(info))
++ {
++ try
++ {
++ mergeSchemaFrom(endpoint, mutations);
++ }
++ catch (Exception e)
++ {
++ logger.error(String.format("Unable to merge schema from %s", endpoint), e);
++ return fail();
++ }
++ }
++ return pullComplete(endpoint, info, true);
++ }
++ }
++
++ public boolean isLatencyForSnitch()
++ {
++ return false;
++ }
++ }
++
++ private void pullSchema(Callback callback)
++ {
++ if (!isAlive(callback.endpoint))
++ {
++ logger.warn("Can't send schema pull request: node {} is down.", callback.endpoint);
++ callback.fail();
++ return;
++ }
++
++ // There is a chance that quite some time could have passed between now and the MM#maybeScheduleSchemaPull(),
++ // potentially enough for the endpoint node to restart - which is an issue if it does restart upgraded, with
++ // a higher major.
++ if (!shouldPullFromEndpoint(callback.endpoint))
++ {
++ logger.info("Skipped sending a migration request: node {} has a higher major version now.", callback.endpoint);
++ callback.fail();
++ return;
++ }
++
++ logger.debug("Requesting schema from {}", callback.endpoint);
++ sendMigrationMessage(callback);
++ }
++
++ protected void sendMigrationMessage(Callback callback)
++ {
++ inflightTasks.getAndIncrement();
++ Message message = Message.out(Verb.SCHEMA_PULL_REQ, NoPayload.noPayload);
++ logger.info("Sending schema pull request to {}", callback.endpoint);
++ MessagingService.instance().sendWithCallback(message, callback.endpoint, callback);
++ }
++
++ private synchronized Future<Void> pullComplete(InetAddressAndPort endpoint, VersionInfo info, boolean wasSuccessful)
++ {
++ inflightTasks.decrementAndGet();
++ if (wasSuccessful)
++ info.markReceived();
++
++ info.outstandingRequests.remove(endpoint);
++ info.requestQueue.add(endpoint);
++ return maybePullSchema(info);
++ }
++
++ public int getInflightTasks()
++ {
++ return inflightTasks.get();
++ }
++
++ /**
++ * Wait until we've received schema responses for all versions we're aware of
++ * @param waitMillis
++ * @return true if response for all schemas were received, false if we timed out waiting
++ */
++ public boolean awaitSchemaRequests(long waitMillis)
++ {
++ if (!FBUtilities.getBroadcastAddressAndPort().equals(InetAddressAndPort.getLoopbackAddress()))
++ Gossiper.waitToSettle();
++
++ WaitQueue.Signal signal = null;
++ try
++ {
++ synchronized (this)
++ {
++ List<WaitQueue.Signal> signalList = new ArrayList<>(versionInfo.size());
++ for (VersionInfo version : versionInfo.values())
++ {
++ if (version.wasReceived())
++ continue;
++
++ signalList.add(version.register());
++ }
++
++ if (signalList.isEmpty())
++ return true;
++
++ WaitQueue.Signal[] signals = new WaitQueue.Signal[signalList.size()];
++ signalList.toArray(signals);
++ signal = WaitQueue.all(signals);
++ }
++
++ return signal.awaitUntil(System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(waitMillis));
++ }
++ catch (InterruptedException e)
++ {
++ throw new RuntimeException(e);
++ }
++ finally
++ {
++ if (signal != null)
++ signal.cancel();
++ }
++ }
++}
diff --cc src/java/org/apache/cassandra/schema/MigrationManager.java
index 8f627cd,0000000..c2ec511
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/schema/MigrationManager.java
+++ b/src/java/org/apache/cassandra/schema/MigrationManager.java
@@@ -1,501 -1,0 +1,360 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.schema;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.*;
+import java.lang.management.ManagementFactory;
+import java.util.function.LongSupplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.Futures;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
- import org.apache.cassandra.concurrent.ScheduledExecutors;
+import org.apache.cassandra.db.*;
+import org.apache.cassandra.exceptions.AlreadyExistsException;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.gms.*;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.concurrent.Stage.MIGRATION;
+import static org.apache.cassandra.net.Verb.SCHEMA_PUSH_REQ;
+
+public class MigrationManager
+{
+ private static final Logger logger = LoggerFactory.getLogger(MigrationManager.class);
+
+ public static final MigrationManager instance = new MigrationManager();
+
+ private static LongSupplier getUptimeFn = () -> ManagementFactory.getRuntimeMXBean().getUptime();
+
+ @VisibleForTesting
+ public static void setUptimeFn(LongSupplier supplier)
+ {
+ getUptimeFn = supplier;
+ }
+
+ private static final int MIGRATION_DELAY_IN_MS = 60000;
+
+ private static final int MIGRATION_TASK_WAIT_IN_SECONDS = Integer.parseInt(System.getProperty("cassandra.migration_task_wait_in_seconds", "1"));
+
+ private MigrationManager() {}
+
- public static void scheduleSchemaPull(InetAddressAndPort endpoint, EndpointState state)
- {
- UUID schemaVersion = state.getSchemaVersion();
- if (!endpoint.equals(FBUtilities.getBroadcastAddressAndPort()) && schemaVersion != null)
- maybeScheduleSchemaPull(schemaVersion, endpoint, state.getApplicationState(ApplicationState.RELEASE_VERSION).value);
- }
-
- /**
- * If versions differ this node sends request with local migration list to the endpoint
- * and expecting to receive a list of migrations to apply locally.
- */
- private static void maybeScheduleSchemaPull(final UUID theirVersion, final InetAddressAndPort endpoint, String releaseVersion)
- {
- String ourMajorVersion = FBUtilities.getReleaseVersionMajor();
- if (!releaseVersion.startsWith(ourMajorVersion))
- {
- logger.debug("Not pulling schema because release version in Gossip is not major version {}, it is {}", ourMajorVersion, releaseVersion);
- return;
- }
- if (Schema.instance.getVersion() == null)
- {
- logger.debug("Not pulling schema from {}, because local schema version is not known yet",
- endpoint);
- SchemaMigrationDiagnostics.unknownLocalSchemaVersion(endpoint, theirVersion);
- return;
- }
- if (Schema.instance.isSameVersion(theirVersion))
- {
- logger.debug("Not pulling schema from {}, because schema versions match ({})",
- endpoint,
- Schema.schemaVersionToString(theirVersion));
- SchemaMigrationDiagnostics.versionMatch(endpoint, theirVersion);
- return;
- }
- if (!shouldPullSchemaFrom(endpoint))
- {
- logger.debug("Not pulling schema from {}, because versions match ({}/{}), or shouldPullSchemaFrom returned false",
- endpoint, Schema.instance.getVersion(), theirVersion);
- SchemaMigrationDiagnostics.skipPull(endpoint, theirVersion);
- return;
- }
-
- if (Schema.instance.isEmpty() || 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 {}, " +
- "schema versions: local={}, remote={}",
- endpoint,
- Schema.schemaVersionToString(Schema.instance.getVersion()),
- Schema.schemaVersionToString(theirVersion));
- submitMigrationTask(endpoint);
- }
- else
- {
- // Include a delay to make sure we have a chance to apply any changes being
- // pushed out simultaneously. See CASSANDRA-5025
- Runnable runnable = () ->
- {
- // grab the latest version of the schema since it may have changed again since the initial scheduling
- UUID epSchemaVersion = Gossiper.instance.getSchemaVersion(endpoint);
- if (epSchemaVersion == null)
- {
- logger.debug("epState vanished for {}, not submitting migration task", endpoint);
- return;
- }
- if (Schema.instance.isSameVersion(epSchemaVersion))
- {
- logger.debug("Not submitting migration task for {} because our versions match ({})", endpoint, epSchemaVersion);
- return;
- }
- logger.debug("Submitting migration task for {}, schema version mismatch: local={}, remote={}",
- endpoint,
- Schema.schemaVersionToString(Schema.instance.getVersion()),
- Schema.schemaVersionToString(epSchemaVersion));
- submitMigrationTask(endpoint);
- };
- ScheduledExecutors.nonPeriodicTasks.schedule(runnable, MIGRATION_DELAY_IN_MS, TimeUnit.MILLISECONDS);
- }
- }
-
- private static Future<?> submitMigrationTask(InetAddressAndPort endpoint)
- {
- /*
- * Do not de-ref the future because that causes distributed deadlock (CASSANDRA-3832) because we are
- * running in the gossip stage.
- */
- return MIGRATION.submit(new MigrationTask(endpoint));
- }
-
- static boolean shouldPullSchemaFrom(InetAddressAndPort endpoint)
- {
- /*
- * Don't request schema from nodes with a differnt or unknonw major version (may have incompatible schema)
- * Don't request schema from fat clients
- */
- return MessagingService.instance().versions.knows(endpoint)
- && MessagingService.instance().versions.getRaw(endpoint) == MessagingService.current_version
- && !Gossiper.instance.isGossipOnlyMember(endpoint);
- }
-
+ private static boolean shouldPushSchemaTo(InetAddressAndPort endpoint)
+ {
+ // only push schema to nodes with known and equal versions
+ return !endpoint.equals(FBUtilities.getBroadcastAddressAndPort())
+ && MessagingService.instance().versions.knows(endpoint)
+ && MessagingService.instance().versions.getRaw(endpoint) == MessagingService.current_version;
+ }
+
- public static boolean isReadyForBootstrap()
- {
- return MigrationTask.getInflightTasks().isEmpty();
- }
-
- public static void waitUntilReadyForBootstrap()
- {
- CountDownLatch completionLatch;
- while ((completionLatch = MigrationTask.getInflightTasks().poll()) != null)
- {
- try
- {
- if (!completionLatch.await(MIGRATION_TASK_WAIT_IN_SECONDS, TimeUnit.SECONDS))
- logger.error("Migration task failed to complete");
- }
- catch (InterruptedException e)
- {
- Thread.currentThread().interrupt();
- logger.error("Migration task was interrupted");
- }
- }
- }
-
+ public static void announceNewKeyspace(KeyspaceMetadata ksm) throws ConfigurationException
+ {
+ announceNewKeyspace(ksm, false);
+ }
+
+ public static void announceNewKeyspace(KeyspaceMetadata ksm, boolean announceLocally) throws ConfigurationException
+ {
+ announceNewKeyspace(ksm, FBUtilities.timestampMicros(), announceLocally);
+ }
+
+ public static void announceNewKeyspace(KeyspaceMetadata ksm, long timestamp, boolean announceLocally) throws ConfigurationException
+ {
+ ksm.validate();
+
+ if (Schema.instance.getKeyspaceMetadata(ksm.name) != null)
+ throw new AlreadyExistsException(ksm.name);
+
+ logger.info("Create new Keyspace: {}", ksm);
+ announce(SchemaKeyspace.makeCreateKeyspaceMutation(ksm, timestamp), announceLocally);
+ }
+
+ public static void announceNewTable(TableMetadata cfm)
+ {
+ announceNewTable(cfm, true, FBUtilities.timestampMicros());
+ }
+
- /**
- * Announces the table even if the definition is already know locally.
- * This should generally be avoided but is used internally when we want to force the most up to date version of
- * a system table schema (Note that we don't know if the schema we force _is_ the most recent version or not, we
- * just rely on idempotency to basically ignore that announce if it's not. That's why we can't use announceTableUpdate
- * it would for instance delete new columns if this is not called with the most up-to-date version)
- *
- * Note that this is only safe for system tables where we know the id is fixed and will be the same whatever version
- * of the definition is used.
- */
- public static void forceAnnounceNewTable(TableMetadata cfm)
- {
- announceNewTable(cfm, false, 0);
- }
-
+ private static void announceNewTable(TableMetadata cfm, boolean throwOnDuplicate, long timestamp)
+ {
+ cfm.validate();
+
+ KeyspaceMetadata ksm = Schema.instance.getKeyspaceMetadata(cfm.keyspace);
+ if (ksm == null)
+ throw new ConfigurationException(String.format("Cannot add table '%s' to non existing keyspace '%s'.", cfm.name, cfm.keyspace));
+ // If we have a table or a view which has the same name, we can't add a new one
+ else if (throwOnDuplicate && ksm.getTableOrViewNullable(cfm.name) != null)
+ throw new AlreadyExistsException(cfm.keyspace, cfm.name);
+
+ logger.info("Create new table: {}", cfm);
+ announce(SchemaKeyspace.makeCreateTableMutation(ksm, cfm, timestamp), false);
+ }
+
+ static void announceKeyspaceUpdate(KeyspaceMetadata ksm)
+ {
+ ksm.validate();
+
+ KeyspaceMetadata oldKsm = Schema.instance.getKeyspaceMetadata(ksm.name);
+ if (oldKsm == null)
+ throw new ConfigurationException(String.format("Cannot update non existing keyspace '%s'.", ksm.name));
+
+ logger.info("Update Keyspace '{}' From {} To {}", ksm.name, oldKsm, ksm);
+ announce(SchemaKeyspace.makeCreateKeyspaceMutation(ksm.name, ksm.params, FBUtilities.timestampMicros()), false);
+ }
+
+ public static void announceTableUpdate(TableMetadata tm)
+ {
+ announceTableUpdate(tm, false);
+ }
+
+ public static void announceTableUpdate(TableMetadata updated, boolean announceLocally)
+ {
+ updated.validate();
+
+ TableMetadata current = Schema.instance.getTableMetadata(updated.keyspace, updated.name);
+ if (current == null)
+ throw new ConfigurationException(String.format("Cannot update non existing table '%s' in keyspace '%s'.", updated.name, updated.keyspace));
+ KeyspaceMetadata ksm = Schema.instance.getKeyspaceMetadata(current.keyspace);
+
+ updated.validateCompatibility(current);
+
+ long timestamp = FBUtilities.timestampMicros();
+
+ logger.info("Update table '{}/{}' From {} To {}", current.keyspace, current.name, current, updated);
+ Mutation.SimpleBuilder builder = SchemaKeyspace.makeUpdateTableMutation(ksm, current, updated, timestamp);
+
+ announce(builder, announceLocally);
+ }
+
+ static void announceKeyspaceDrop(String ksName)
+ {
+ KeyspaceMetadata oldKsm = Schema.instance.getKeyspaceMetadata(ksName);
+ if (oldKsm == null)
+ throw new ConfigurationException(String.format("Cannot drop non existing keyspace '%s'.", ksName));
+
+ logger.info("Drop Keyspace '{}'", oldKsm.name);
+ announce(SchemaKeyspace.makeDropKeyspaceMutation(oldKsm, FBUtilities.timestampMicros()), false);
+ }
+
+ public static void announceTableDrop(String ksName, String cfName, boolean announceLocally)
+ {
+ TableMetadata tm = Schema.instance.getTableMetadata(ksName, cfName);
+ if (tm == null)
+ throw new ConfigurationException(String.format("Cannot drop non existing table '%s' in keyspace '%s'.", cfName, ksName));
+ KeyspaceMetadata ksm = Schema.instance.getKeyspaceMetadata(ksName);
+
+ logger.info("Drop table '{}/{}'", tm.keyspace, tm.name);
+ announce(SchemaKeyspace.makeDropTableMutation(ksm, tm, FBUtilities.timestampMicros()), announceLocally);
+ }
+
+ /**
+ * actively announce a new version to active hosts via rpc
+ * @param schema The schema mutation to be applied
+ */
+ private static void announce(Mutation.SimpleBuilder schema, boolean announceLocally)
+ {
+ List<Mutation> mutations = Collections.singletonList(schema.build());
+
+ if (announceLocally)
+ Schema.instance.merge(mutations);
+ else
+ announce(mutations);
+ }
+
+ public static void announce(Mutation change)
+ {
+ announce(Collections.singleton(change));
+ }
+
+ public static void announce(Collection<Mutation> schema)
+ {
+ Future<?> f = MIGRATION.submit(() -> Schema.instance.mergeAndAnnounceVersion(schema));
+
+ Set<InetAddressAndPort> schemaDestinationEndpoints = new HashSet<>();
+ Set<InetAddressAndPort> schemaEndpointsIgnored = new HashSet<>();
+ Message<Collection<Mutation>> message = Message.out(SCHEMA_PUSH_REQ, schema);
+ for (InetAddressAndPort endpoint : Gossiper.instance.getLiveMembers())
+ {
+ if (shouldPushSchemaTo(endpoint))
+ {
+ MessagingService.instance().send(message, endpoint);
+ schemaDestinationEndpoints.add(endpoint);
+ }
+ else
+ {
+ schemaEndpointsIgnored.add(endpoint);
+ }
+ }
+
+ SchemaAnnouncementDiagnostics.schemaMutationsAnnounced(schemaDestinationEndpoints, schemaEndpointsIgnored);
+ FBUtilities.waitOnFuture(f);
+ }
+
+ public static KeyspacesDiff announce(SchemaTransformation transformation, boolean locally)
+ {
+ long now = FBUtilities.timestampMicros();
+
+ Future<Schema.TransformationResult> future =
+ MIGRATION.submit(() -> Schema.instance.transform(transformation, locally, now));
+
+ Schema.TransformationResult result = Futures.getUnchecked(future);
+ if (!result.success)
+ throw result.exception;
+
+ if (locally || result.diff.isEmpty())
+ return result.diff;
+
+ Set<InetAddressAndPort> schemaDestinationEndpoints = new HashSet<>();
+ Set<InetAddressAndPort> schemaEndpointsIgnored = new HashSet<>();
+ Message<Collection<Mutation>> message = Message.out(SCHEMA_PUSH_REQ, result.mutations);
+ for (InetAddressAndPort endpoint : Gossiper.instance.getLiveMembers())
+ {
+ if (shouldPushSchemaTo(endpoint))
+ {
+ MessagingService.instance().send(message, endpoint);
+ schemaDestinationEndpoints.add(endpoint);
+ }
+ else
+ {
+ schemaEndpointsIgnored.add(endpoint);
+ }
+ }
+
+ SchemaAnnouncementDiagnostics.schemaTransformationAnnounced(schemaDestinationEndpoints, schemaEndpointsIgnored,
+ transformation);
+
+ return result.diff;
+ }
+
+ /**
+ * Clear all locally stored schema information and reset schema to initial state.
+ * Called by user (via JMX) who wants to get rid of schema disagreement.
+ */
+ public static void resetLocalSchema()
+ {
+ logger.info("Starting local schema reset...");
+
+ logger.debug("Truncating schema tables...");
+
+ SchemaMigrationDiagnostics.resetLocalSchema();
+
+ SchemaKeyspace.truncate();
+
+ logger.debug("Clearing local schema keyspace definitions...");
+
+ Schema.instance.clear();
+
+ Set<InetAddressAndPort> liveEndpoints = Gossiper.instance.getLiveMembers();
+ liveEndpoints.remove(FBUtilities.getBroadcastAddressAndPort());
+
+ // force migration if there are nodes around
+ for (InetAddressAndPort node : liveEndpoints)
+ {
- if (shouldPullSchemaFrom(node))
- {
- logger.debug("Requesting schema from {}", node);
- FBUtilities.waitOnFuture(submitMigrationTask(node));
- break;
- }
++ EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(node);
++ Future<Void> pull = MigrationCoordinator.instance.reportEndpointVersion(node, state);
++ if (pull != null)
++ FBUtilities.waitOnFuture(pull);
+ }
+
+ logger.info("Local schema reset is complete.");
+ }
+
+ /**
+ * We have a set of non-local, distributed system keyspaces, e.g. system_traces, system_auth, etc.
+ * (see {@link SchemaConstants#REPLICATED_SYSTEM_KEYSPACE_NAMES}), that need to be created on cluster initialisation,
+ * and later evolved on major upgrades (sometimes minor too). This method compares the current known definitions
+ * of the tables (if the keyspace exists) to the expected, most modern ones expected by the running version of C*;
+ * if any changes have been detected, a schema Mutation will be created which, when applied, should make
+ * cluster's view of that keyspace aligned with the expected modern definition.
+ *
+ * @param keyspace the expected modern definition of the keyspace
+ * @param generation timestamp to use for the table changes in the schema mutation
+ *
+ * @return empty Optional if the current definition is up to date, or an Optional with the Mutation that would
+ * bring the schema in line with the expected definition.
+ */
+ public static Optional<Mutation> evolveSystemKeyspace(KeyspaceMetadata keyspace, long generation)
+ {
+ Mutation.SimpleBuilder builder = null;
+
+ KeyspaceMetadata definedKeyspace = Schema.instance.getKeyspaceMetadata(keyspace.name);
+ Tables definedTables = null == definedKeyspace ? Tables.none() : definedKeyspace.tables;
+
+ for (TableMetadata table : keyspace.tables)
+ {
+ if (table.equals(definedTables.getNullable(table.name)))
+ continue;
+
+ if (null == builder)
+ {
+ // for the keyspace definition itself (name, replication, durability) always use generation 0;
+ // this ensures that any changes made to replication by the user will never be overwritten.
+ builder = SchemaKeyspace.makeCreateKeyspaceMutation(keyspace.name, keyspace.params, 0);
+
+ // now set the timestamp to generation, so the tables have the expected timestamp
+ builder.timestamp(generation);
+ }
+
+ // for table definitions always use the provided generation; these tables, unlike their containing
+ // keyspaces, are *NOT* meant to be altered by the user; if their definitions need to change,
+ // the schema must be updated in code, and the appropriate generation must be bumped.
+ SchemaKeyspace.addTableToSchemaMutation(table, true, builder);
+ }
+
+ return builder == null ? Optional.empty() : Optional.of(builder.build());
+ }
+
+ public static class MigrationsSerializer implements IVersionedSerializer<Collection<Mutation>>
+ {
+ public static MigrationsSerializer instance = new MigrationsSerializer();
+
+ public void serialize(Collection<Mutation> schema, DataOutputPlus out, int version) throws IOException
+ {
+ out.writeInt(schema.size());
+ for (Mutation mutation : schema)
+ Mutation.serializer.serialize(mutation, out, version);
+ }
+
+ public Collection<Mutation> deserialize(DataInputPlus in, int version) throws IOException
+ {
+ int count = in.readInt();
+ Collection<Mutation> schema = new ArrayList<>(count);
+
+ for (int i = 0; i < count; i++)
+ schema.add(Mutation.serializer.deserialize(in, version));
+
+ return schema;
+ }
+
+ public long serializedSize(Collection<Mutation> schema, int version)
+ {
+ int size = TypeSizes.sizeof(schema.size());
+ for (Mutation mutation : schema)
+ size += mutation.serializedSize(version);
+ return size;
+ }
+ }
+}
diff --cc src/java/org/apache/cassandra/schema/Schema.java
index 9c0c590,0000000..c04c631
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/schema/Schema.java
+++ b/src/java/org/apache/cassandra/schema/Schema.java
@@@ -1,890 -1,0 +1,890 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.schema;
+
+import java.util.*;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.MapDifference;
+import com.google.common.collect.Sets;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.functions.*;
+import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.UserType;
+import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.gms.ApplicationState;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.locator.LocalStrategy;
+import org.apache.cassandra.schema.KeyspaceMetadata.KeyspaceDiff;
+import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.Pair;
+import org.cliffc.high_scale_lib.NonBlockingHashMap;
+
+import static java.lang.String.format;
+
+import static com.google.common.collect.Iterables.size;
+
+public final class Schema implements SchemaProvider
+{
+ public static final Schema instance = new Schema();
+
+ private volatile Keyspaces keyspaces = Keyspaces.none();
+
+ // UUID -> mutable metadata ref map. We have to update these in place every time a table changes.
+ private final Map<TableId, TableMetadataRef> metadataRefs = new NonBlockingHashMap<>();
+
+ // (keyspace name, index name) -> mutable metadata ref map. We have to update these in place every time an index changes.
+ private final Map<Pair<String, String>, TableMetadataRef> indexMetadataRefs = new NonBlockingHashMap<>();
+
+ // Keyspace objects, one per keyspace. Only one instance should ever exist for any given keyspace.
+ private final Map<String, Keyspace> keyspaceInstances = new NonBlockingHashMap<>();
+
+ private volatile UUID version;
+
+ private final List<SchemaChangeListener> changeListeners = new CopyOnWriteArrayList<>();
+
+ /**
+ * Initialize empty schema object and load the hardcoded system tables
+ */
+ private Schema()
+ {
+ if (DatabaseDescriptor.isDaemonInitialized() || DatabaseDescriptor.isToolInitialized())
+ {
+ load(SchemaKeyspace.metadata());
+ load(SystemKeyspace.metadata());
+ }
+ }
+
+ /**
+ * load keyspace (keyspace) definitions, but do not initialize the keyspace instances.
+ * Schema version may be updated as the result.
+ */
+ public void loadFromDisk()
+ {
+ loadFromDisk(true);
+ }
+
+ /**
+ * Load schema definitions from disk.
+ *
+ * @param updateVersion true if schema version needs to be updated
+ */
+ public void loadFromDisk(boolean updateVersion)
+ {
+ SchemaDiagnostics.schemataLoading(this);
+ SchemaKeyspace.fetchNonSystemKeyspaces().forEach(this::load);
+ if (updateVersion)
+ updateVersion();
+ SchemaDiagnostics.schemataLoaded(this);
+ }
+
+ /**
+ * Update (or insert) new keyspace definition
+ *
+ * @param ksm The metadata about keyspace
+ */
+ synchronized public void load(KeyspaceMetadata ksm)
+ {
+ KeyspaceMetadata previous = keyspaces.getNullable(ksm.name);
+
+ if (previous == null)
+ loadNew(ksm);
+ else
+ reload(previous, ksm);
+
+ keyspaces = keyspaces.withAddedOrUpdated(ksm);
+ }
+
+ private void loadNew(KeyspaceMetadata ksm)
+ {
+ ksm.tablesAndViews()
+ .forEach(metadata -> metadataRefs.put(metadata.id, new TableMetadataRef(metadata)));
+
+ ksm.tables
+ .indexTables()
+ .forEach((name, metadata) -> indexMetadataRefs.put(Pair.create(ksm.name, name), new TableMetadataRef(metadata)));
+
+ SchemaDiagnostics.metadataInitialized(this, ksm);
+ }
+
+ private void reload(KeyspaceMetadata previous, KeyspaceMetadata updated)
+ {
+ Keyspace keyspace = getKeyspaceInstance(updated.name);
+ if (null != keyspace)
+ keyspace.setMetadata(updated);
+
+ Tables.TablesDiff tablesDiff = Tables.diff(previous.tables, updated.tables);
+ Views.ViewsDiff viewsDiff = Views.diff(previous.views, updated.views);
+
+ MapDifference<String, TableMetadata> indexesDiff = previous.tables.indexesDiff(updated.tables);
+
+ // clean up after removed entries
+ tablesDiff.dropped.forEach(table -> metadataRefs.remove(table.id));
+ viewsDiff.dropped.forEach(view -> metadataRefs.remove(view.metadata.id));
+
+ indexesDiff.entriesOnlyOnLeft()
+ .values()
+ .forEach(indexTable -> indexMetadataRefs.remove(Pair.create(indexTable.keyspace, indexTable.indexName().get())));
+
+ // load up new entries
+ tablesDiff.created.forEach(table -> metadataRefs.put(table.id, new TableMetadataRef(table)));
+ viewsDiff.created.forEach(view -> metadataRefs.put(view.metadata.id, new TableMetadataRef(view.metadata)));
+
+ indexesDiff.entriesOnlyOnRight()
+ .values()
+ .forEach(indexTable -> indexMetadataRefs.put(Pair.create(indexTable.keyspace, indexTable.indexName().get()), new TableMetadataRef(indexTable)));
+
+ // refresh refs to updated ones
+ tablesDiff.altered.forEach(diff -> metadataRefs.get(diff.after.id).set(diff.after));
+ viewsDiff.altered.forEach(diff -> metadataRefs.get(diff.after.metadata.id).set(diff.after.metadata));
+
+ indexesDiff.entriesDiffering()
+ .values()
+ .stream()
+ .map(MapDifference.ValueDifference::rightValue)
+ .forEach(indexTable -> indexMetadataRefs.get(Pair.create(indexTable.keyspace, indexTable.indexName().get())).set(indexTable));
+
+ SchemaDiagnostics.metadataReloaded(this, previous, updated, tablesDiff, viewsDiff, indexesDiff);
+ }
+
+ public void registerListener(SchemaChangeListener listener)
+ {
+ changeListeners.add(listener);
+ }
+
+ @SuppressWarnings("unused")
+ public void unregisterListener(SchemaChangeListener listener)
+ {
+ changeListeners.remove(listener);
+ }
+
+ /**
+ * Get keyspace instance by name
+ *
+ * @param keyspaceName The name of the keyspace
+ *
+ * @return Keyspace object or null if keyspace was not found
+ */
+ @Override
+ public Keyspace getKeyspaceInstance(String keyspaceName)
+ {
+ return keyspaceInstances.get(keyspaceName);
+ }
+
+ public ColumnFamilyStore getColumnFamilyStoreInstance(TableId id)
+ {
+ TableMetadata metadata = getTableMetadata(id);
+ if (metadata == null)
+ return null;
+
+ Keyspace instance = getKeyspaceInstance(metadata.keyspace);
+ if (instance == null)
+ return null;
+
+ return instance.hasColumnFamilyStore(metadata.id)
+ ? instance.getColumnFamilyStore(metadata.id)
+ : null;
+ }
+
+ /**
+ * Store given Keyspace instance to the schema
+ *
+ * @param keyspace The Keyspace instance to store
+ *
+ * @throws IllegalArgumentException if Keyspace is already stored
+ */
+ @Override
+ public void storeKeyspaceInstance(Keyspace keyspace)
+ {
+ if (keyspaceInstances.putIfAbsent(keyspace.getName(), keyspace) != null)
+ throw new IllegalArgumentException(String.format("Keyspace %s was already initialized.", keyspace.getName()));
+ }
+
+ /**
+ * Remove keyspace from schema
+ *
+ * @param keyspaceName The name of the keyspace to remove
+ *
+ * @return removed keyspace instance or null if it wasn't found
+ */
+ public Keyspace removeKeyspaceInstance(String keyspaceName)
+ {
+ return keyspaceInstances.remove(keyspaceName);
+ }
+
+ public Keyspaces snapshot()
+ {
+ return keyspaces;
+ }
+
+ /**
+ * Remove keyspace definition from system
+ *
+ * @param ksm The keyspace definition to remove
+ */
+ synchronized void unload(KeyspaceMetadata ksm)
+ {
+ keyspaces = keyspaces.without(ksm.name);
+
+ ksm.tablesAndViews()
+ .forEach(t -> metadataRefs.remove(t.id));
+
+ ksm.tables
+ .indexTables()
+ .keySet()
+ .forEach(name -> indexMetadataRefs.remove(Pair.create(ksm.name, name)));
+
+ SchemaDiagnostics.metadataRemoved(this, ksm);
+ }
+
+ public int getNumberOfTables()
+ {
+ return keyspaces.stream().mapToInt(k -> size(k.tablesAndViews())).sum();
+ }
+
+ public ViewMetadata getView(String keyspaceName, String viewName)
+ {
+ assert keyspaceName != null;
+ KeyspaceMetadata ksm = keyspaces.getNullable(keyspaceName);
+ return (ksm == null) ? null : ksm.views.getNullable(viewName);
+ }
+
+ /**
+ * Get metadata about keyspace by its name
+ *
+ * @param keyspaceName The name of the keyspace
+ *
+ * @return The keyspace metadata or null if it wasn't found
+ */
+ @Override
+ public KeyspaceMetadata getKeyspaceMetadata(String keyspaceName)
+ {
+ assert keyspaceName != null;
+ KeyspaceMetadata keyspace = keyspaces.getNullable(keyspaceName);
+ return null != keyspace ? keyspace : VirtualKeyspaceRegistry.instance.getKeyspaceMetadataNullable(keyspaceName);
+ }
+
+ private Set<String> getNonSystemKeyspacesSet()
+ {
+ return Sets.difference(keyspaces.names(), SchemaConstants.LOCAL_SYSTEM_KEYSPACE_NAMES);
+ }
+
+ /**
+ * @return collection of the non-system keyspaces (note that this count as system only the
+ * non replicated keyspaces, so keyspace like system_traces which are replicated are actually
+ * returned. See getUserKeyspace() below if you don't want those)
+ */
+ public ImmutableList<String> getNonSystemKeyspaces()
+ {
+ return ImmutableList.copyOf(getNonSystemKeyspacesSet());
+ }
+
+ /**
+ * @return a collection of keyspaces that do not use LocalStrategy for replication
+ */
+ public List<String> getNonLocalStrategyKeyspaces()
+ {
+ return keyspaces.stream()
+ .filter(keyspace -> keyspace.params.replication.klass != LocalStrategy.class)
+ .map(keyspace -> keyspace.name)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * @return collection of the user defined keyspaces
+ */
+ public List<String> getUserKeyspaces()
+ {
+ return ImmutableList.copyOf(Sets.difference(getNonSystemKeyspacesSet(), SchemaConstants.REPLICATED_SYSTEM_KEYSPACE_NAMES));
+ }
+
+ /**
+ * Get metadata about keyspace inner ColumnFamilies
+ *
+ * @param keyspaceName The name of the keyspace
+ *
+ * @return metadata about ColumnFamilies the belong to the given keyspace
+ */
+ public Iterable<TableMetadata> getTablesAndViews(String keyspaceName)
+ {
+ assert keyspaceName != null;
+ KeyspaceMetadata ksm = keyspaces.getNullable(keyspaceName);
+ assert ksm != null;
+ return ksm.tablesAndViews();
+ }
+
+ /**
+ * @return collection of the all keyspace names registered in the system (system and non-system)
+ */
+ public Set<String> getKeyspaces()
+ {
+ return keyspaces.names();
+ }
+
+ /* TableMetadata/Ref query/control methods */
+
+ /**
+ * Given a keyspace name and table/view name, get the table metadata
+ * reference. If the keyspace name or table/view name is not present
+ * this method returns null.
+ *
+ * @return TableMetadataRef object or null if it wasn't found
+ */
+ @Override
+ public TableMetadataRef getTableMetadataRef(String keyspace, String table)
+ {
+ TableMetadata tm = getTableMetadata(keyspace, table);
+ return tm == null
+ ? null
+ : metadataRefs.get(tm.id);
+ }
+
+ public TableMetadataRef getIndexTableMetadataRef(String keyspace, String index)
+ {
+ return indexMetadataRefs.get(Pair.create(keyspace, index));
+ }
+
+ Map<Pair<String, String>, TableMetadataRef> getIndexTableMetadataRefs()
+ {
+ return indexMetadataRefs;
+ }
+
+ /**
+ * Get Table metadata by its identifier
+ *
+ * @param id table or view identifier
+ *
+ * @return metadata about Table or View
+ */
+ @Override
+ public TableMetadataRef getTableMetadataRef(TableId id)
+ {
+ return metadataRefs.get(id);
+ }
+
+ @Override
+ public TableMetadataRef getTableMetadataRef(Descriptor descriptor)
+ {
+ return getTableMetadataRef(descriptor.ksname, descriptor.cfname);
+ }
+
+ Map<TableId, TableMetadataRef> getTableMetadataRefs()
+ {
+ return metadataRefs;
+ }
+
+ /**
+ * Given a keyspace name and table name, get the table
+ * meta data. If the keyspace name or table name is not valid
+ * this function returns null.
+ *
+ * @param keyspace The keyspace name
+ * @param table The table name
+ *
+ * @return TableMetadata object or null if it wasn't found
+ */
+ public TableMetadata getTableMetadata(String keyspace, String table)
+ {
+ assert keyspace != null;
+ assert table != null;
+
+ KeyspaceMetadata ksm = getKeyspaceMetadata(keyspace);
+ return ksm == null
+ ? null
+ : ksm.getTableOrViewNullable(table);
+ }
+
+ @Override
+ public TableMetadata getTableMetadata(TableId id)
+ {
+ TableMetadata table = keyspaces.getTableOrViewNullable(id);
+ return null != table ? table : VirtualKeyspaceRegistry.instance.getTableMetadataNullable(id);
+ }
+
+ public TableMetadata validateTable(String keyspaceName, String tableName)
+ {
+ if (tableName.isEmpty())
+ throw new InvalidRequestException("non-empty table is required");
+
+ KeyspaceMetadata keyspace = getKeyspaceMetadata(keyspaceName);
+ if (keyspace == null)
+ throw new KeyspaceNotDefinedException(format("keyspace %s does not exist", keyspaceName));
+
+ TableMetadata metadata = keyspace.getTableOrViewNullable(tableName);
+ if (metadata == null)
+ throw new InvalidRequestException(format("table %s does not exist", tableName));
+
+ return metadata;
+ }
+
+ public TableMetadata getTableMetadata(Descriptor descriptor)
+ {
+ return getTableMetadata(descriptor.ksname, descriptor.cfname);
+ }
+
+ /* Function helpers */
+
+ /**
+ * Get all function overloads with the specified name
+ *
+ * @param name fully qualified function name
+ * @return an empty list if the keyspace or the function name are not found;
+ * a non-empty collection of {@link Function} otherwise
+ */
+ public Collection<Function> getFunctions(FunctionName name)
+ {
+ if (!name.hasKeyspace())
+ throw new IllegalArgumentException(String.format("Function name must be fully qualified: got %s", name));
+
+ KeyspaceMetadata ksm = getKeyspaceMetadata(name.keyspace);
+ return ksm == null
+ ? Collections.emptyList()
+ : ksm.functions.get(name);
+ }
+
+ /**
+ * Find the function with the specified name
+ *
+ * @param name fully qualified function name
+ * @param argTypes function argument types
+ * @return an empty {@link Optional} if the keyspace or the function name are not found;
+ * a non-empty optional of {@link Function} otherwise
+ */
+ public Optional<Function> findFunction(FunctionName name, List<AbstractType<?>> argTypes)
+ {
+ if (!name.hasKeyspace())
+ throw new IllegalArgumentException(String.format("Function name must be fully quallified: got %s", name));
+
+ KeyspaceMetadata ksm = getKeyspaceMetadata(name.keyspace);
+ return ksm == null
+ ? Optional.empty()
+ : ksm.functions.find(name, argTypes);
+ }
+
+ /* Version control */
+
+ /**
+ * @return current schema version
+ */
+ public UUID getVersion()
+ {
+ return version;
+ }
+
+ /**
+ * Checks whether the given schema version is the same as the current local schema.
+ */
+ public boolean isSameVersion(UUID schemaVersion)
+ {
+ return schemaVersion != null && schemaVersion.equals(version);
+ }
+
+ /**
+ * Checks whether the current schema is empty.
+ */
+ public boolean isEmpty()
+ {
+ return SchemaConstants.emptyVersion.equals(version);
+ }
+
+ /**
+ * Read schema from system keyspace and calculate MD5 digest of every row, resulting digest
+ * will be converted into UUID which would act as content-based version of the schema.
+ */
+ public void updateVersion()
+ {
+ version = SchemaKeyspace.calculateSchemaDigest();
+ SystemKeyspace.updateSchemaVersion(version);
+ SchemaDiagnostics.versionUpdated(this);
+ }
+
+ /*
+ * Like updateVersion, but also announces via gossip
+ */
+ public void updateVersionAndAnnounce()
+ {
+ updateVersion();
+ passiveAnnounceVersion();
+ }
+
+ /**
+ * Announce my version passively over gossip.
+ * Used to notify nodes as they arrive in the cluster.
+ */
+ private void passiveAnnounceVersion()
+ {
+ Gossiper.instance.addLocalApplicationState(ApplicationState.SCHEMA, StorageService.instance.valueFactory.schema(version));
+ SchemaDiagnostics.versionAnnounced(this);
+ }
+
+ /**
+ * Clear all KS/CF metadata and reset version.
+ */
+ public synchronized void clear()
+ {
+ getNonSystemKeyspaces().forEach(k -> unload(getKeyspaceMetadata(k)));
+ updateVersionAndAnnounce();
+ SchemaDiagnostics.schemataCleared(this);
+ }
+
+ /*
+ * Reload schema from local disk. Useful if a user made changes to schema tables by hand, or has suspicion that
+ * in-memory representation got out of sync somehow with what's on disk.
+ */
+ public synchronized void reloadSchemaAndAnnounceVersion()
+ {
+ Keyspaces before = keyspaces.filter(k -> !SchemaConstants.isLocalSystemKeyspace(k.name));
+ Keyspaces after = SchemaKeyspace.fetchNonSystemKeyspaces();
+ merge(Keyspaces.diff(before, after));
+ updateVersionAndAnnounce();
+ }
+
+ /**
+ * Merge remote schema in form of mutations with local and mutate ks/cf metadata objects
+ * (which also involves fs operations on add/drop ks/cf)
+ *
+ * @param mutations the schema changes to apply
+ *
+ * @throws ConfigurationException If one of metadata attributes has invalid value
+ */
- synchronized void mergeAndAnnounceVersion(Collection<Mutation> mutations)
++ public synchronized void mergeAndAnnounceVersion(Collection<Mutation> mutations)
+ {
+ merge(mutations);
+ updateVersionAndAnnounce();
+ }
+
+ public synchronized TransformationResult transform(SchemaTransformation transformation, boolean locally, long now)
+ {
+ KeyspacesDiff diff;
+ try
+ {
+ Keyspaces before = keyspaces;
+ Keyspaces after = transformation.apply(before);
+ diff = Keyspaces.diff(before, after);
+ }
+ catch (RuntimeException e)
+ {
+ return new TransformationResult(e);
+ }
+
+ if (diff.isEmpty())
+ return new TransformationResult(diff, Collections.emptyList());
+
+ Collection<Mutation> mutations = SchemaKeyspace.convertSchemaDiffToMutations(diff, now);
+ SchemaKeyspace.applyChanges(mutations);
+
+ merge(diff);
+ updateVersion();
+ if (!locally)
+ passiveAnnounceVersion();
+
+ return new TransformationResult(diff, mutations);
+ }
+
+ public static final class TransformationResult
+ {
+ public final boolean success;
+ public final RuntimeException exception;
+ public final KeyspacesDiff diff;
+ public final Collection<Mutation> mutations;
+
+ private TransformationResult(boolean success, RuntimeException exception, KeyspacesDiff diff, Collection<Mutation> mutations)
+ {
+ this.success = success;
+ this.exception = exception;
+ this.diff = diff;
+ this.mutations = mutations;
+ }
+
+ TransformationResult(RuntimeException exception)
+ {
+ this(false, exception, null, null);
+ }
+
+ TransformationResult(KeyspacesDiff diff, Collection<Mutation> mutations)
+ {
+ this(true, null, diff, mutations);
+ }
+ }
+
+ synchronized void merge(Collection<Mutation> mutations)
+ {
+ // only compare the keyspaces affected by this set of schema mutations
+ Set<String> affectedKeyspaces = SchemaKeyspace.affectedKeyspaces(mutations);
+
+ // fetch the current state of schema for the affected keyspaces only
+ Keyspaces before = keyspaces.filter(k -> affectedKeyspaces.contains(k.name));
+
+ // apply the schema mutations
+ SchemaKeyspace.applyChanges(mutations);
+
+ // apply the schema mutations and fetch the new versions of the altered keyspaces
+ Keyspaces after = SchemaKeyspace.fetchKeyspaces(affectedKeyspaces);
+
+ merge(Keyspaces.diff(before, after));
+ }
+
+ private void merge(KeyspacesDiff diff)
+ {
+ diff.dropped.forEach(this::dropKeyspace);
+ diff.created.forEach(this::createKeyspace);
+ diff.altered.forEach(this::alterKeyspace);
+ }
+
+ private void alterKeyspace(KeyspaceDiff delta)
+ {
+ SchemaDiagnostics.keyspaceAltering(this, delta);
+
+ // drop tables and views
+ delta.views.dropped.forEach(this::dropView);
+ delta.tables.dropped.forEach(this::dropTable);
+
+ load(delta.after);
+
+ // add tables and views
+ delta.tables.created.forEach(this::createTable);
+ delta.views.created.forEach(this::createView);
+
+ // update tables and views
+ delta.tables.altered.forEach(diff -> alterTable(diff.after));
+ delta.views.altered.forEach(diff -> alterView(diff.after));
+
+ // deal with all added, and altered views
+ Keyspace.open(delta.after.name).viewManager.reload(true);
+
+ // notify on everything dropped
+ delta.udas.dropped.forEach(uda -> notifyDropAggregate((UDAggregate) uda));
+ delta.udfs.dropped.forEach(udf -> notifyDropFunction((UDFunction) udf));
+ delta.views.dropped.forEach(this::notifyDropView);
+ delta.tables.dropped.forEach(this::notifyDropTable);
+ delta.types.dropped.forEach(this::notifyDropType);
+
+ // notify on everything created
+ delta.types.created.forEach(this::notifyCreateType);
+ delta.tables.created.forEach(this::notifyCreateTable);
+ delta.views.created.forEach(this::notifyCreateView);
+ delta.udfs.created.forEach(udf -> notifyCreateFunction((UDFunction) udf));
+ delta.udas.created.forEach(uda -> notifyCreateAggregate((UDAggregate) uda));
+
+ // notify on everything altered
+ if (!delta.before.params.equals(delta.after.params))
+ notifyAlterKeyspace(delta.before, delta.after);
+ delta.types.altered.forEach(diff -> notifyAlterType(diff.before, diff.after));
+ delta.tables.altered.forEach(diff -> notifyAlterTable(diff.before, diff.after));
+ delta.views.altered.forEach(diff -> notifyAlterView(diff.before, diff.after));
+ delta.udfs.altered.forEach(diff -> notifyAlterFunction(diff.before, diff.after));
+ delta.udas.altered.forEach(diff -> notifyAlterAggregate(diff.before, diff.after));
+ SchemaDiagnostics.keyspaceAltered(this, delta);
+ }
+
+ private void createKeyspace(KeyspaceMetadata keyspace)
+ {
+ SchemaDiagnostics.keyspaceCreating(this, keyspace);
+ load(keyspace);
+ Keyspace.open(keyspace.name);
+
+ notifyCreateKeyspace(keyspace);
+ keyspace.types.forEach(this::notifyCreateType);
+ keyspace.tables.forEach(this::notifyCreateTable);
+ keyspace.views.forEach(this::notifyCreateView);
+ keyspace.functions.udfs().forEach(this::notifyCreateFunction);
+ keyspace.functions.udas().forEach(this::notifyCreateAggregate);
+ SchemaDiagnostics.keyspaceCreated(this, keyspace);
+ }
+
+ private void dropKeyspace(KeyspaceMetadata keyspace)
+ {
+ SchemaDiagnostics.keyspaceDroping(this, keyspace);
+ keyspace.views.forEach(this::dropView);
+ keyspace.tables.forEach(this::dropTable);
+
+ // remove the keyspace from the static instances.
+ Keyspace.clear(keyspace.name);
+ unload(keyspace);
+ Keyspace.writeOrder.awaitNewBarrier();
+
+ keyspace.functions.udas().forEach(this::notifyDropAggregate);
+ keyspace.functions.udfs().forEach(this::notifyDropFunction);
+ keyspace.views.forEach(this::notifyDropView);
+ keyspace.tables.forEach(this::notifyDropTable);
+ keyspace.types.forEach(this::notifyDropType);
+ notifyDropKeyspace(keyspace);
+ SchemaDiagnostics.keyspaceDroped(this, keyspace);
+ }
+
+ private void dropView(ViewMetadata metadata)
+ {
+ Keyspace.open(metadata.keyspace()).viewManager.dropView(metadata.name());
+ dropTable(metadata.metadata);
+ }
+
+ private void dropTable(TableMetadata metadata)
+ {
+ SchemaDiagnostics.tableDropping(this, metadata);
+ ColumnFamilyStore cfs = Keyspace.open(metadata.keyspace).getColumnFamilyStore(metadata.name);
+ assert cfs != null;
+ // make sure all the indexes are dropped, or else.
+ cfs.indexManager.markAllIndexesRemoved();
+ CompactionManager.instance.interruptCompactionFor(Collections.singleton(metadata), (sstable) -> true, true);
+ if (DatabaseDescriptor.isAutoSnapshot())
+ cfs.snapshot(Keyspace.getTimestampedSnapshotNameWithPrefix(cfs.name, ColumnFamilyStore.SNAPSHOT_DROP_PREFIX));
+ CommitLog.instance.forceRecycleAllSegments(Collections.singleton(metadata.id));
+ Keyspace.open(metadata.keyspace).dropCf(metadata.id);
+ SchemaDiagnostics.tableDropped(this, metadata);
+ }
+
+ private void createTable(TableMetadata table)
+ {
+ SchemaDiagnostics.tableCreating(this, table);
+ Keyspace.open(table.keyspace).initCf(metadataRefs.get(table.id), true);
+ SchemaDiagnostics.tableCreated(this, table);
+ }
+
+ private void createView(ViewMetadata view)
+ {
+ Keyspace.open(view.keyspace()).initCf(metadataRefs.get(view.metadata.id), true);
+ }
+
+ private void alterTable(TableMetadata updated)
+ {
+ SchemaDiagnostics.tableAltering(this, updated);
+ Keyspace.open(updated.keyspace).getColumnFamilyStore(updated.name).reload();
+ SchemaDiagnostics.tableAltered(this, updated);
+ }
+
+ private void alterView(ViewMetadata updated)
+ {
+ Keyspace.open(updated.keyspace()).getColumnFamilyStore(updated.name()).reload();
+ }
+
+ private void notifyCreateKeyspace(KeyspaceMetadata ksm)
+ {
+ changeListeners.forEach(l -> l.onCreateKeyspace(ksm.name));
+ }
+
+ private void notifyCreateTable(TableMetadata metadata)
+ {
+ changeListeners.forEach(l -> l.onCreateTable(metadata.keyspace, metadata.name));
+ }
+
+ private void notifyCreateView(ViewMetadata view)
+ {
+ changeListeners.forEach(l -> l.onCreateView(view.keyspace(), view.name()));
+ }
+
+ private void notifyCreateType(UserType ut)
+ {
+ changeListeners.forEach(l -> l.onCreateType(ut.keyspace, ut.getNameAsString()));
+ }
+
+ private void notifyCreateFunction(UDFunction udf)
+ {
+ changeListeners.forEach(l -> l.onCreateFunction(udf.name().keyspace, udf.name().name, udf.argTypes()));
+ }
+
+ private void notifyCreateAggregate(UDAggregate udf)
+ {
+ changeListeners.forEach(l -> l.onCreateAggregate(udf.name().keyspace, udf.name().name, udf.argTypes()));
+ }
+
+ private void notifyAlterKeyspace(KeyspaceMetadata before, KeyspaceMetadata after)
+ {
+ changeListeners.forEach(l -> l.onAlterKeyspace(after.name));
+ }
+
+ private void notifyAlterTable(TableMetadata before, TableMetadata after)
+ {
+ boolean changeAffectedPreparedStatements = before.changeAffectsPreparedStatements(after);
+ changeListeners.forEach(l -> l.onAlterTable(after.keyspace, after.name, changeAffectedPreparedStatements));
+ }
+
+ private void notifyAlterView(ViewMetadata before, ViewMetadata after)
+ {
+ boolean changeAffectedPreparedStatements = before.metadata.changeAffectsPreparedStatements(after.metadata);
+ changeListeners.forEach(l ->l.onAlterView(after.keyspace(), after.name(), changeAffectedPreparedStatements));
+ }
+
+ private void notifyAlterType(UserType before, UserType after)
+ {
+ changeListeners.forEach(l -> l.onAlterType(after.keyspace, after.getNameAsString()));
+ }
+
+ private void notifyAlterFunction(UDFunction before, UDFunction after)
+ {
+ changeListeners.forEach(l -> l.onAlterFunction(after.name().keyspace, after.name().name, after.argTypes()));
+ }
+
+ private void notifyAlterAggregate(UDAggregate before, UDAggregate after)
+ {
+ changeListeners.forEach(l -> l.onAlterAggregate(after.name().keyspace, after.name().name, after.argTypes()));
+ }
+
+ private void notifyDropKeyspace(KeyspaceMetadata ksm)
+ {
+ changeListeners.forEach(l -> l.onDropKeyspace(ksm.name));
+ }
+
+ private void notifyDropTable(TableMetadata metadata)
+ {
+ changeListeners.forEach(l -> l.onDropTable(metadata.keyspace, metadata.name));
+ }
+
+ private void notifyDropView(ViewMetadata view)
+ {
+ changeListeners.forEach(l -> l.onDropView(view.keyspace(), view.name()));
+ }
+
+ private void notifyDropType(UserType ut)
+ {
+ changeListeners.forEach(l -> l.onDropType(ut.keyspace, ut.getNameAsString()));
+ }
+
+ private void notifyDropFunction(UDFunction udf)
+ {
+ changeListeners.forEach(l -> l.onDropFunction(udf.name().keyspace, udf.name().name, udf.argTypes()));
+ }
+
+ private void notifyDropAggregate(UDAggregate udf)
+ {
+ changeListeners.forEach(l -> l.onDropAggregate(udf.name().keyspace, udf.name().name, udf.argTypes()));
+ }
+
+
+ /**
+ * Converts the given schema version to a string. Returns {@code unknown}, if {@code version} is {@code null}
+ * or {@code "(empty)"}, if {@code version} refers to an {@link SchemaConstants#emptyVersion empty) schema.
+ */
+ public static String schemaVersionToString(UUID version)
+ {
+ return version == null
+ ? "unknown"
+ : SchemaConstants.emptyVersion.equals(version)
+ ? "(empty)"
+ : version.toString();
+ }
+}
diff --cc src/java/org/apache/cassandra/schema/SchemaMigrationEvent.java
index 0e6b7a3,0000000..a984804
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/schema/SchemaMigrationEvent.java
+++ b/src/java/org/apache/cassandra/schema/SchemaMigrationEvent.java
@@@ -1,114 -1,0 +1,111 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.schema;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.diag.DiagnosticEvent;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.MessagingService;
+
+/**
+ * Internal events emitted by {@link MigrationManager}.
+ */
+final class SchemaMigrationEvent extends DiagnosticEvent
+{
+ private final MigrationManagerEventType type;
+ @Nullable
+ private final InetAddressAndPort endpoint;
+ @Nullable
+ private final UUID endpointSchemaVersion;
+ private final UUID localSchemaVersion;
+ private final Integer localMessagingVersion;
+ private final SystemKeyspace.BootstrapState bootstrapState;
- @Nullable
- private Integer inflightTaskCount;
++ private final Integer inflightTaskCount;
+ @Nullable
+ private Integer endpointMessagingVersion;
+ @Nullable
+ private Boolean endpointGossipOnlyMember;
+ @Nullable
+ private Boolean isAlive;
+
+ enum MigrationManagerEventType
+ {
+ UNKNOWN_LOCAL_SCHEMA_VERSION,
+ VERSION_MATCH,
+ SKIP_PULL,
+ RESET_LOCAL_SCHEMA,
+ TASK_CREATED,
+ TASK_SEND_ABORTED,
+ TASK_REQUEST_SEND
+ }
+
+ SchemaMigrationEvent(MigrationManagerEventType type,
+ @Nullable InetAddressAndPort endpoint, @Nullable UUID endpointSchemaVersion)
+ {
+ this.type = type;
+ this.endpoint = endpoint;
+ this.endpointSchemaVersion = endpointSchemaVersion;
+
+ localSchemaVersion = Schema.instance.getVersion();
+ localMessagingVersion = MessagingService.current_version;
+
- Queue<CountDownLatch> inflightTasks = MigrationTask.getInflightTasks();
- if (inflightTasks != null)
- inflightTaskCount = inflightTasks.size();
++ inflightTaskCount = MigrationCoordinator.instance.getInflightTasks();
+
+ this.bootstrapState = SystemKeyspace.getBootstrapState();
+
+ if (endpoint == null) return;
+
+ if (MessagingService.instance().versions.knows(endpoint))
+ endpointMessagingVersion = MessagingService.instance().versions.getRaw(endpoint);
+
+ endpointGossipOnlyMember = Gossiper.instance.isGossipOnlyMember(endpoint);
+ this.isAlive = FailureDetector.instance.isAlive(endpoint);
+ }
+
+ public Enum<?> getType()
+ {
+ return type;
+ }
+
+ public Map<String, Serializable> toMap()
+ {
+ HashMap<String, Serializable> ret = new HashMap<>();
+ if (endpoint != null) ret.put("endpoint", endpoint.getHostAddressAndPort());
+ ret.put("endpointSchemaVersion", Schema.schemaVersionToString(endpointSchemaVersion));
+ ret.put("localSchemaVersion", Schema.schemaVersionToString(localSchemaVersion));
+ if (endpointMessagingVersion != null) ret.put("endpointMessagingVersion", endpointMessagingVersion);
+ if (localMessagingVersion != null) ret.put("localMessagingVersion", localMessagingVersion);
+ if (endpointGossipOnlyMember != null) ret.put("endpointGossipOnlyMember", endpointGossipOnlyMember);
+ if (isAlive != null) ret.put("endpointIsAlive", isAlive);
+ if (bootstrapState != null) ret.put("bootstrapState", bootstrapState.name());
+ if (inflightTaskCount != null) ret.put("inflightTaskCount", inflightTaskCount);
+ return ret;
+ }
+}
diff --cc src/java/org/apache/cassandra/service/StorageService.java
index ca2bff2,734f176..4a3477c
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@@ -30,13 -29,10 +30,14 @@@ import java.util.concurrent.atomic.Atom
import java.util.regex.MatchResult;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import java.util.stream.StreamSupport;
++import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.management.*;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
import javax.management.openmbean.TabularDataSupport;
@@@ -91,17 -82,16 +92,18 @@@ import org.apache.cassandra.repair.*
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
import org.apache.cassandra.schema.KeyspaceMetadata;
-import org.apache.cassandra.schema.SchemaKeyspace;
-import org.apache.cassandra.service.paxos.CommitVerbHandler;
-import org.apache.cassandra.service.paxos.PrepareVerbHandler;
-import org.apache.cassandra.service.paxos.ProposeVerbHandler;
++import org.apache.cassandra.schema.MigrationCoordinator;
+import org.apache.cassandra.schema.MigrationManager;
+import org.apache.cassandra.schema.ReplicationParams;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaConstants;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.TableMetadataRef;
+import org.apache.cassandra.schema.ViewMetadata;
import org.apache.cassandra.streaming.*;
-import org.apache.cassandra.thrift.EndpointDetails;
-import org.apache.cassandra.thrift.TokenRange;
-import org.apache.cassandra.thrift.cassandraConstants;
import org.apache.cassandra.tracing.TraceKeyspace;
import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.transport.Server;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.logging.LoggingSupportFactory;
import org.apache.cassandra.utils.progress.ProgressEvent;
@@@ -135,8 -116,10 +137,11 @@@ public class StorageService extends Not
{
private static final Logger logger = LoggerFactory.getLogger(StorageService.class);
+ public static final int INDEFINITE = -1;
public static final int RING_DELAY = getRingDelay(); // delay after which we assume ring has stablized
+ public static final int SCHEMA_DELAY = getRingDelay(); // delay after which we assume ring has stablized
+
+ private static final boolean REQUIRE_SCHEMAS = !Boolean.getBoolean("cassandra.skip_schema_check");
private final JMXProgressSupport progressSupport = new JMXProgressSupport(this);
@@@ -149,11 -141,23 +154,25 @@@
return Integer.parseInt(newdelay);
}
else
+ {
return 30 * 1000;
+ }
}
+ private static int getSchemaDelay()
+ {
+ String newdelay = System.getProperty("cassandra.schema_delay_ms");
+ if (newdelay != null)
+ {
+ logger.info("Overriding SCHEMA_DELAY to {}ms", newdelay);
+ return Integer.parseInt(newdelay);
+ }
+ else
+ {
+ return 30 * 1000;
+ }
+ }
+
/* This abstraction maintains the token/endpoint metadata information */
private TokenMetadata tokenMetadata = new TokenMetadata();
@@@ -774,11 -824,13 +793,12 @@@
public static boolean isSeed()
{
- return DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddress());
+ return DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddressAndPort());
}
- @VisibleForTesting
- public void prepareToJoin() throws ConfigurationException
+ private void prepareToJoin() throws ConfigurationException
{
+ MigrationCoordinator.instance.start();
if (!joined)
{
Map<ApplicationState, VersionedValue> appStates = new EnumMap<>(ApplicationState.class);
@@@ -880,26 -932,23 +900,31 @@@
}
Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
}
- // if our schema hasn't matched yet, wait until it has
- // we do this by waiting for all in-flight migration requests and responses to complete
- // (post CASSANDRA-1391 we don't expect this to be necessary very often, but it doesn't hurt to be careful)
- if (!MigrationManager.isReadyForBootstrap())
- {
- setMode(Mode.JOINING, "waiting for schema information to complete", true);
- MigrationManager.waitUntilReadyForBootstrap();
- }
+
+ boolean schemasReceived = MigrationCoordinator.instance.awaitSchemaRequests(TimeUnit.SECONDS.toMillis(SCHEMA_DELAY));
+
+ if (schemasReceived)
+ return;
+
+ logger.warn(String.format("There are nodes in the cluster with a different schema version than us we did not merged schemas from, " +
+ "our version : (%s), outstanding versions -> endpoints : %s",
+ Schema.instance.getVersion(),
+ MigrationCoordinator.instance.outstandingVersions()));
+
+ if (REQUIRE_SCHEMAS)
+ throw new RuntimeException("Didn't receive schemas for all known versions within the timeout");
}
+ private void joinTokenRing(long schemaTimeoutMillis) throws ConfigurationException
+ {
+ joinTokenRing(!isSurveyMode, shouldBootstrap(), schemaTimeoutMillis, INDEFINITE);
+ }
+
@VisibleForTesting
- private void joinTokenRing(int delay) throws ConfigurationException
+ public void joinTokenRing(boolean finishJoiningRing,
+ boolean shouldBootstrap,
+ long schemaTimeoutMillis,
+ long bootstrapTimeoutMillis) throws ConfigurationException
{
joined = true;
@@@ -2233,11 -2104,11 +2258,11 @@@
}
break;
case SCHEMA:
- SystemKeyspace.updatePeerInfo(endpoint, "schema_version", UUID.fromString(value.value), executor);
+ SystemKeyspace.updatePeerInfo(endpoint, "schema_version", UUID.fromString(value.value));
- MigrationManager.instance.scheduleSchemaPull(endpoint, epState);
+ MigrationCoordinator.instance.reportEndpointVersion(endpoint, UUID.fromString(value.value));
break;
case HOST_ID:
- SystemKeyspace.updatePeerInfo(endpoint, "host_id", UUID.fromString(value.value), executor);
+ SystemKeyspace.updatePeerInfo(endpoint, "host_id", UUID.fromString(value.value));
break;
case RPC_READY:
notifyRpcChange(endpoint, epState.isRpcReady());
@@@ -3126,13 -2866,11 +3151,11 @@@
{
onChange(endpoint, entry.getKey(), entry.getValue());
}
- MigrationManager.instance.scheduleSchemaPull(endpoint, epState);
+ MigrationCoordinator.instance.reportEndpointVersion(endpoint, epState);
}
- public void onAlive(InetAddress endpoint, EndpointState state)
+ public void onAlive(InetAddressAndPort endpoint, EndpointState state)
{
- MigrationManager.instance.scheduleSchemaPull(endpoint, state);
-
if (tokenMetadata.isMember(endpoint))
notifyUp(endpoint);
}
@@@ -5683,42 -5381,9 +5706,58 @@@
}
@Override
+ public void enableFullQueryLogger(String path, String rollCycle, Boolean blocking, int maxQueueWeight, long maxLogSize, String archiveCommand, int maxArchiveRetries)
+ {
+ FullQueryLoggerOptions fqlOptions = DatabaseDescriptor.getFullQueryLogOptions();
+ path = path != null ? path : fqlOptions.log_dir;
+ rollCycle = rollCycle != null ? rollCycle : fqlOptions.roll_cycle;
+ blocking = blocking != null ? blocking : fqlOptions.block;
+ maxQueueWeight = maxQueueWeight != Integer.MIN_VALUE ? maxQueueWeight : fqlOptions.max_queue_weight;
+ maxLogSize = maxLogSize != Long.MIN_VALUE ? maxLogSize : fqlOptions.max_log_size;
+ archiveCommand = archiveCommand != null ? archiveCommand : fqlOptions.archive_command;
+ maxArchiveRetries = maxArchiveRetries != Integer.MIN_VALUE ? maxArchiveRetries : fqlOptions.max_archive_retries;
+
+ Preconditions.checkNotNull(path, "cassandra.yaml did not set log_dir and not set as parameter");
+ FullQueryLogger.instance.enable(Paths.get(path), rollCycle, blocking, maxQueueWeight, maxLogSize, archiveCommand, maxArchiveRetries);
+ }
+
+ @Override
+ public void resetFullQueryLogger()
+ {
+ FullQueryLogger.instance.reset(DatabaseDescriptor.getFullQueryLogOptions().log_dir);
+ }
+
+ @Override
+ public void stopFullQueryLogger()
+ {
+ FullQueryLogger.instance.stop();
+ }
+
+ @Override
+ public boolean isFullQueryLogEnabled()
+ {
+ return FullQueryLogger.instance.isEnabled();
+ }
+
+ @Override
+ public CompositeData getFullQueryLoggerOptions()
+ {
+ return FullQueryLoggerOptionsCompositeData.toCompositeData(FullQueryLogger.instance.getFullQueryLoggerOptions());
+ }
++
++ @Override
+ public Map<String, Set<InetAddress>> getOutstandingSchemaVersions()
+ {
- Map<UUID, Set<InetAddress>> outstanding = MigrationCoordinator.instance.outstandingVersions();
- return outstanding.entrySet().stream().collect(Collectors.toMap(e -> e.getKey().toString(), Entry::getValue));
++ Map<UUID, Set<InetAddressAndPort>> outstanding = MigrationCoordinator.instance.outstandingVersions();
++ return outstanding.entrySet().stream().collect(Collectors.toMap(e -> e.getKey().toString(),
++ e -> e.getValue().stream().map(i -> i.address).collect(Collectors.toSet())));
++ }
++
++ @Override
++ public Map<String, Set<String>> getOutstandingSchemaVersionsWithPort()
++ {
++ Map<UUID, Set<InetAddressAndPort>> outstanding = MigrationCoordinator.instance.outstandingVersions();
++ return outstanding.entrySet().stream().collect(Collectors.toMap(e -> e.getKey().toString(),
++ e -> e.getValue().stream().map(InetAddressAndPort::toString).collect(Collectors.toSet())));
+ }
}
diff --cc src/java/org/apache/cassandra/service/StorageServiceMBean.java
index 618a37e,218cdd9..0c4b5b0
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@@ -22,19 -21,16 +22,21 @@@ import java.io.Serializable
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
-import java.util.Collection;
import java.util.List;
import java.util.Map;
+ import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
-
+import javax.annotation.Nullable;
import javax.management.NotificationEmitter;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
+import org.apache.cassandra.db.ColumnFamilyStoreMBean;
+import org.apache.cassandra.exceptions.ConfigurationException;
++import org.apache.cassandra.locator.InetAddressAndPort;
+
public interface StorageServiceMBean extends NotificationEmitter
{
/**
@@@ -750,73 -707,9 +752,78 @@@
*/
public boolean resumeBootstrap();
- /** Returns the max version that this node will negotiate for native protocol connections */
- public int getMaxNativeProtocolVersion();
+ /** Gets the concurrency settings for processing stages*/
+ static class StageConcurrency implements Serializable
+ {
+ public final int corePoolSize;
+ public final int maximumPoolSize;
+
+ public StageConcurrency(int corePoolSize, int maximumPoolSize)
+ {
+ this.corePoolSize = corePoolSize;
+ this.maximumPoolSize = maximumPoolSize;
+ }
+
+ }
+ public Map<String, List<Integer>> getConcurrency(List<String> stageNames);
+
+ /** Sets the concurrency setting for processing stages */
+ public void setConcurrency(String threadPoolName, int newCorePoolSize, int newMaximumPoolSize);
+
+ /** Clears the history of clients that have connected in the past **/
+ void clearConnectionHistory();
+ public void disableAuditLog();
+ public void enableAuditLog(String loggerName, Map<String, String> parameters, String includedKeyspaces, String excludedKeyspaces, String includedCategories, String excludedCategories, String includedUsers, String excludedUsers) throws ConfigurationException;
+ public void enableAuditLog(String loggerName, String includedKeyspaces, String excludedKeyspaces, String includedCategories, String excludedCategories, String includedUsers, String excludedUsers) throws ConfigurationException;
+ public boolean isAuditLogEnabled();
+ public String getCorruptedTombstoneStrategy();
+ public void setCorruptedTombstoneStrategy(String strategy);
+
+ /**
+ * Start the fully query logger.
+ * @param path Path where the full query log will be stored. If null cassandra.yaml value is used.
+ * @param rollCycle How often to create a new file for query data (MINUTELY, DAILY, HOURLY)
+ * @param blocking Whether threads submitting queries to the query log should block if they can't be drained to the filesystem or alternatively drops samples and log
+ * @param maxQueueWeight How many bytes of query data to queue before blocking or dropping samples
+ * @param maxLogSize How many bytes of log data to store before dropping segments. Might not be respected if a log file hasn't rolled so it can be deleted.
+ * @param archiveCommand executable archiving the rolled log files,
+ * @param maxArchiveRetries max number of times to retry a failing archive command
+ *
+ */
+ public void enableFullQueryLogger(String path, String rollCycle, Boolean blocking, int maxQueueWeight, long maxLogSize, @Nullable String archiveCommand, int maxArchiveRetries);
+
+ /**
+ * Disable the full query logger if it is enabled.
+ * Also delete any generated files in the last used full query log path as well as the one configure in cassandra.yaml
+ */
+ public void resetFullQueryLogger();
+
+ /**
+ * Stop logging queries but leave any generated files on disk.
+ */
+ public void stopFullQueryLogger();
+
+ public boolean isFullQueryLogEnabled();
+
+ /**
+ * Returns the current state of FQL.
+ */
+ CompositeData getFullQueryLoggerOptions();
+
+ /** Sets the initial allocation size of backing arrays for new RangeTombstoneList objects */
+ public void setInitialRangeTombstoneListAllocationSize(int size);
+
+ /** Returns the initial allocation size of backing arrays for new RangeTombstoneList objects */
+ public int getInitialRangeTombstoneListAllocationSize();
+
+ /** Sets the resize factor to use when growing/resizing a RangeTombstoneList */
+ public void setRangeTombstoneListResizeGrowthFactor(double growthFactor);
+
+ /** Returns the resize factor to use when growing/resizing a RangeTombstoneList */
+ public double getRangeTombstoneResizeListGrowthFactor();
+
+ /** Returns a map of schema version -> list of endpoints reporting that version that we need schema updates for */
++ @Deprecated
+ public Map<String, Set<InetAddress>> getOutstandingSchemaVersions();
++ public Map<String, Set<String>> getOutstandingSchemaVersionsWithPort();
}
diff --cc test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java
index 70ef503,0000000..6dda98e
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java
+++ b/test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java
@@@ -1,457 -1,0 +1,461 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.action;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.shared.VersionedApplicationState;
+import org.apache.cassandra.gms.ApplicationState;
+import org.apache.cassandra.gms.EndpointState;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.gms.VersionedValue;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.MessagingService;
++import org.apache.cassandra.schema.MigrationCoordinator;
+import org.apache.cassandra.schema.MigrationManager;
+import org.apache.cassandra.service.PendingRangeCalculatorService;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.distributed.impl.DistributedTestSnitch.toCassandraInetAddressAndPort;
+
+public class GossipHelper
+{
+ public static InstanceAction statusToBootstrap(IInvokableInstance newNode)
+ {
+ return (instance) ->
+ {
+ changeGossipState(instance,
+ newNode,
+ Arrays.asList(tokens(newNode),
+ statusBootstrapping(newNode),
+ statusWithPortBootstrapping(newNode)));
+ };
+ }
+
+ public static InstanceAction statusToNormal(IInvokableInstance peer)
+ {
+ return (target) ->
+ {
+ changeGossipState(target,
+ peer,
+ Arrays.asList(tokens(peer),
+ statusNormal(peer),
+ releaseVersion(peer),
+ netVersion(peer),
+ statusWithPortNormal(peer)));
+ };
+ }
+
+ /**
+ * This method is unsafe and should be used _only_ when gossip is not used or available: it creates versioned values on the
+ * target instance, which means Gossip versioning gets out of sync. Use a safe couterpart at all times when performing _any_
+ * ring movement operations _or_ if Gossip is used.
+ */
+ public static void unsafeStatusToNormal(IInvokableInstance target, IInstance peer)
+ {
+ int messagingVersion = peer.getMessagingVersion();
+ changeGossipState(target,
+ peer,
+ Arrays.asList(unsafeVersionedValue(target,
+ ApplicationState.TOKENS,
+ (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).tokens(tokens),
+ peer.config().getString("partitioner"),
+ peer.config().getString("initial_token")),
+ unsafeVersionedValue(target,
+ ApplicationState.STATUS,
+ (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).normal(tokens),
+ peer.config().getString("partitioner"),
+ peer.config().getString("initial_token")),
+ unsafeVersionedValue(target,
+ ApplicationState.STATUS_WITH_PORT,
+ (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).normal(tokens),
+ peer.config().getString("partitioner"),
+ peer.config().getString("initial_token")),
+ unsafeVersionedValue(target,
+ ApplicationState.NET_VERSION,
+ (partitioner) -> new VersionedValue.VersionedValueFactory(partitioner).networkVersion(messagingVersion),
+ peer.config().getString("partitioner")),
+ unsafeReleaseVersion(target,
+ peer.config().getString("partitioner"),
+ peer.getReleaseVersionString())));
+ }
+
+ public static InstanceAction statusToLeaving(IInvokableInstance newNode)
+ {
+ return (instance) -> {
+ changeGossipState(instance,
+ newNode,
+ Arrays.asList(tokens(newNode),
+ statusLeaving(newNode),
+ statusWithPortLeaving(newNode)));
+ };
+ }
+
+ public static InstanceAction bootstrap()
+ {
+ return new BootstrapAction();
+ }
+
+ public static InstanceAction bootstrap(boolean joinRing, Duration waitForBootstrap, Duration waitForSchema)
+ {
+ return new BootstrapAction(joinRing, waitForBootstrap, waitForSchema);
+ }
+
+ public static InstanceAction disseminateGossipState(IInvokableInstance newNode)
+ {
+ return new DisseminateGossipState(newNode);
+ }
+
+ public static InstanceAction pullSchemaFrom(IInvokableInstance pullFrom)
+ {
+ return new PullSchemaFrom(pullFrom);
+ }
+
+ private static InstanceAction disableBinary()
+ {
+ return (instance) -> instance.nodetoolResult("disablebinary").asserts().success();
+ }
+
+ private static class DisseminateGossipState implements InstanceAction
+ {
+ final Map<InetSocketAddress, byte[]> gossipState;
+
+ public DisseminateGossipState(IInvokableInstance... from)
+ {
+ gossipState = new HashMap<>();
+ for (IInvokableInstance node : from)
+ {
+ byte[] epBytes = node.callsOnInstance(() -> {
+ EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(FBUtilities.getBroadcastAddressAndPort());
+ return toBytes(epState);
+ }).call();
+ gossipState.put(node.broadcastAddress(), epBytes);
+ }
+ }
+
+ public void accept(IInvokableInstance instance)
+ {
+ instance.appliesOnInstance((IIsolatedExecutor.SerializableFunction<Map<InetSocketAddress, byte[]>, Void>)
+ (map) -> {
+ Map<InetAddressAndPort, EndpointState> newState = new HashMap<>();
+ for (Map.Entry<InetSocketAddress, byte[]> e : map.entrySet())
+ newState.put(toCassandraInetAddressAndPort(e.getKey()), fromBytes(e.getValue()));
+
+ Gossiper.runInGossipStageBlocking(() -> {
+ Gossiper.instance.applyStateLocally(newState);
+ });
+ return null;
+ }).apply(gossipState);
+ }
+ }
+
+ private static byte[] toBytes(EndpointState epState)
+ {
+ try (DataOutputBuffer out = new DataOutputBuffer(1024))
+ {
+ EndpointState.serializer.serialize(epState, out, MessagingService.current_version);
+ return out.toByteArray();
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static EndpointState fromBytes(byte[] bytes)
+ {
+ try (DataInputBuffer in = new DataInputBuffer(bytes))
+ {
+ return EndpointState.serializer.deserialize(in, MessagingService.current_version);
+ }
+ catch (Throwable t)
+ {
+ throw new RuntimeException(t);
+ }
+ }
+
+ private static class PullSchemaFrom implements InstanceAction
+ {
+ final InetSocketAddress pullFrom;
+
+ public PullSchemaFrom(IInvokableInstance pullFrom)
+ {
+ this.pullFrom = pullFrom.broadcastAddress();;
+ }
+
+ public void accept(IInvokableInstance pullTo)
+ {
+ pullTo.acceptsOnInstance((InetSocketAddress pullFrom) -> {
+ InetAddressAndPort endpoint = toCassandraInetAddressAndPort(pullFrom);
+ EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
- MigrationManager.scheduleSchemaPull(endpoint, state);
- MigrationManager.waitUntilReadyForBootstrap();
++ MigrationCoordinator.instance.reportEndpointVersion(endpoint, state);
++ MigrationCoordinator.instance.awaitSchemaRequests(TimeUnit.SECONDS.toMillis(10));
+ }).accept(pullFrom);
+ }
+ }
+
+ private static class BootstrapAction implements InstanceAction, Serializable
+ {
+ private final boolean joinRing;
+ private final Duration waitForBootstrap;
+ private final Duration waitForSchema;
+
+ public BootstrapAction()
+ {
+ this(true, Duration.ofMinutes(10), Duration.ofSeconds(10));
+ }
+
+ public BootstrapAction(boolean joinRing, Duration waitForBootstrap, Duration waitForSchema)
+ {
+ this.joinRing = joinRing;
+ this.waitForBootstrap = waitForBootstrap;
+ this.waitForSchema = waitForSchema;
+ }
+
+ public void accept(IInvokableInstance instance)
+ {
+ instance.appliesOnInstance((String partitionerString, String tokenString) -> {
+ IPartitioner partitioner = FBUtilities.newPartitioner(partitionerString);
+ List<Token> tokens = Collections.singletonList(partitioner.getTokenFactory().fromString(tokenString));
+ try
+ {
+ Collection<InetAddressAndPort> collisions = StorageService.instance.prepareForBootstrap(waitForSchema.toMillis());
+ assert collisions.size() == 0 : String.format("Didn't expect any replacements but got %s", collisions);
+ boolean isBootstrapSuccessful = StorageService.instance.bootstrap(tokens, waitForBootstrap.toMillis());
+ assert isBootstrapSuccessful : "Bootstrap did not complete successfully";
+ StorageService.instance.setUpDistributedSystemKeyspaces();
+ if (joinRing)
+ StorageService.instance.finishJoiningRing(true, tokens);
+ }
+ catch (Throwable t)
+ {
+ throw new RuntimeException(t);
+ }
+
+ return null;
+ }).apply(instance.config().getString("partitioner"), instance.config().getString("initial_token"));
+ }
+ }
+
+ public static InstanceAction decomission()
+ {
+ return (target) -> target.nodetoolResult("decommission").asserts().success();
+ }
+
+
+ public static VersionedApplicationState tokens(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.TOKENS, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).tokens(tokens));
+ }
+
+ public static VersionedApplicationState netVersion(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.NET_VERSION, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).networkVersion());
+ }
+
+ private static VersionedApplicationState unsafeReleaseVersion(IInvokableInstance instance, String partitionerStr, String releaseVersionStr)
+ {
+ return unsafeVersionedValue(instance, ApplicationState.RELEASE_VERSION, (partitioner) -> new VersionedValue.VersionedValueFactory(partitioner).releaseVersion(releaseVersionStr), partitionerStr);
+ }
+
+ public static VersionedApplicationState releaseVersion(IInvokableInstance instance)
+ {
+ return unsafeReleaseVersion(instance, instance.config().getString("partitioner"), instance.getReleaseVersionString());
+ }
+
+ public static VersionedApplicationState statusNormal(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).normal(tokens));
+ }
+
+ public static VersionedApplicationState statusWithPortNormal(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS_WITH_PORT, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).normal(tokens));
+ }
+
+ public static VersionedApplicationState statusBootstrapping(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).bootstrapping(tokens));
+ }
+
+ public static VersionedApplicationState statusWithPortBootstrapping(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS_WITH_PORT, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).bootstrapping(tokens));
+ }
+
+ public static VersionedApplicationState statusLeaving(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).leaving(tokens));
+ }
+
+ public static VersionedApplicationState statusLeft(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS, (partitioner, tokens) -> {
+ return new VersionedValue.VersionedValueFactory(partitioner).left(tokens, System.currentTimeMillis() + Gossiper.aVeryLongTime);
+ });
+ }
+
+ public static VersionedApplicationState statusWithPortLeft(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS_WITH_PORT, (partitioner, tokens) -> {
+ return new VersionedValue.VersionedValueFactory(partitioner).left(tokens, System.currentTimeMillis() + Gossiper.aVeryLongTime);
+
+ });
+ }
+
+ public static VersionedApplicationState statusWithPortLeaving(IInvokableInstance instance)
+ {
+ return versionedToken(instance, ApplicationState.STATUS_WITH_PORT, (partitioner, tokens) -> new VersionedValue.VersionedValueFactory(partitioner).leaving(tokens));
+ }
+
+ public static VersionedValue toVersionedValue(VersionedApplicationState vv)
+ {
+ return VersionedValue.unsafeMakeVersionedValue(vv.value, vv.version);
+ }
+
+ public static ApplicationState toApplicationState(VersionedApplicationState vv)
+ {
+ return ApplicationState.values()[vv.applicationState];
+ }
+
+ private static VersionedApplicationState unsafeVersionedValue(IInvokableInstance instance,
+ ApplicationState applicationState,
+ IIsolatedExecutor.SerializableBiFunction<IPartitioner, Collection<Token>, VersionedValue> supplier,
+ String partitionerStr, String initialTokenStr)
+ {
+ return instance.appliesOnInstance((String partitionerString, String tokenString) -> {
+ IPartitioner partitioner = FBUtilities.newPartitioner(partitionerString);
+ Token token = partitioner.getTokenFactory().fromString(tokenString);
+
+ VersionedValue versionedValue = supplier.apply(partitioner, Collections.singleton(token));
+ return new VersionedApplicationState(applicationState.ordinal(), versionedValue.value, versionedValue.version);
+ }).apply(partitionerStr, initialTokenStr);
+ }
+
+ private static VersionedApplicationState unsafeVersionedValue(IInvokableInstance instance,
+ ApplicationState applicationState,
+ IIsolatedExecutor.SerializableFunction<IPartitioner, VersionedValue> supplier,
+ String partitionerStr)
+ {
+ return instance.appliesOnInstance((String partitionerString) -> {
+ IPartitioner partitioner = FBUtilities.newPartitioner(partitionerString);
+ VersionedValue versionedValue = supplier.apply(partitioner);
+ return new VersionedApplicationState(applicationState.ordinal(), versionedValue.value, versionedValue.version);
+ }).apply(partitionerStr);
+ }
+
+ public static VersionedApplicationState versionedToken(IInvokableInstance instance, ApplicationState applicationState, IIsolatedExecutor.SerializableBiFunction<IPartitioner, Collection<Token>, VersionedValue> supplier)
+ {
+ return unsafeVersionedValue(instance, applicationState, supplier, instance.config().getString("partitioner"), instance.config().getString("initial_token"));
+ }
+
+ public static InstanceAction removeFromRing(IInvokableInstance peer)
+ {
+ return (target) -> {
+ InetAddressAndPort endpoint = toCassandraInetAddressAndPort(peer.broadcastAddress());
+ VersionedApplicationState newState = statusLeft(peer);
+
+ target.runOnInstance(() -> {
+ // state to 'left'
+ EndpointState currentState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
+ ApplicationState as = toApplicationState(newState);
+ VersionedValue vv = toVersionedValue(newState);
+ currentState.addApplicationState(as, vv);
+ StorageService.instance.onChange(endpoint, as, vv);
+
+ // remove from gossip
+ Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.unsafeAnulEndpoint(endpoint));
+ SystemKeyspace.removeEndpoint(endpoint);
+ PendingRangeCalculatorService.instance.update();
+ PendingRangeCalculatorService.instance.blockUntilFinished();
+ });
+ };
+ }
+
+ /**
+ * Changes gossip state of the `peer` on `target`
+ */
+ public static void changeGossipState(IInvokableInstance target, IInstance peer, List<VersionedApplicationState> newState)
+ {
+ InetSocketAddress addr = peer.broadcastAddress();
+ UUID hostId = peer.config().hostId();
+ int netVersion = peer.getMessagingVersion();
+ target.runOnInstance(() -> {
+ InetAddressAndPort endpoint = toCassandraInetAddressAndPort(addr);
+ StorageService storageService = StorageService.instance;
+
+ Gossiper.runInGossipStageBlocking(() -> {
+ EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
+ if (state == null)
+ {
+ Gossiper.instance.initializeNodeUnsafe(endpoint, hostId, netVersion, 1);
+ state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
+ if (state.isAlive() && !Gossiper.instance.isDeadState(state))
+ Gossiper.instance.realMarkAlive(endpoint, state);
+ }
+
+ for (VersionedApplicationState value : newState)
+ {
+ ApplicationState as = toApplicationState(value);
+ VersionedValue vv = toVersionedValue(value);
+ state.addApplicationState(as, vv);
+ storageService.onChange(endpoint, as, vv);
+ }
+ });
+ });
+ }
+
+ public static void withProperty(String prop, boolean value, Runnable r)
+ {
+ String before = System.getProperty(prop);
+ try
+ {
+ System.setProperty(prop, Boolean.toString(value));
+ r.run();
+ }
+ finally
+ {
- System.setProperty(prop, before == null ? "true" : before);
++ if (before == null)
++ System.clearProperty(prop);
++ else
++ System.setProperty(prop, before);
+ }
+ }
+}
diff --cc test/unit/org/apache/cassandra/schema/MigrationCoordinatorTest.java
index 0000000,0000000..daba0f1
new file mode 100644
--- /dev/null
+++ b/test/unit/org/apache/cassandra/schema/MigrationCoordinatorTest.java
@@@ -1,0 -1,0 +1,319 @@@
++/*
++ * Licensed to the Apache Software Foundation (ASF) under one
++ * or more contributor license agreements. See the NOTICE file
++ * distributed with this work for additional information
++ * regarding copyright ownership. The ASF licenses this file
++ * to you under the Apache License, Version 2.0 (the
++ * "License"); you may not use this file except in compliance
++ * with the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++
++package org.apache.cassandra.schema;
++
++import java.net.UnknownHostException;
++import java.util.Collection;
++import java.util.Collections;
++import java.util.HashSet;
++import java.util.LinkedList;
++import java.util.List;
++import java.util.Queue;
++import java.util.Set;
++import java.util.UUID;
++import java.util.concurrent.Future;
++
++import com.google.common.collect.Iterables;
++import com.google.common.collect.Sets;
++import com.google.common.util.concurrent.Futures;
++import org.junit.Assert;
++import org.junit.Test;
++
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
++
++import org.apache.cassandra.config.DatabaseDescriptor;
++import org.apache.cassandra.db.Mutation;
++import org.apache.cassandra.locator.InetAddressAndPort;
++import org.apache.cassandra.utils.concurrent.WaitQueue;
++
++import static com.google.common.util.concurrent.Futures.getUnchecked;
++
++public class MigrationCoordinatorTest
++{
++ private static final Logger logger = LoggerFactory.getLogger(MigrationCoordinatorTest.class);
++
++ private static final InetAddressAndPort EP1;
++ private static final InetAddressAndPort EP2;
++ private static final InetAddressAndPort EP3;
++
++ private static final UUID LOCAL_VERSION = UUID.randomUUID();
++ private static final UUID V1 = UUID.randomUUID();
++ private static final UUID V2 = UUID.randomUUID();
++ private static final UUID V3 = UUID.randomUUID();
++
++ static
++ {
++ try
++ {
++ EP1 = InetAddressAndPort.getByName("10.0.0.1");
++ EP2 = InetAddressAndPort.getByName("10.0.0.2");
++ EP3 = InetAddressAndPort.getByName("10.0.0.3");
++ }
++ catch (UnknownHostException e)
++ {
++ throw new AssertionError(e);
++ }
++
++ DatabaseDescriptor.daemonInitialization();
++ }
++
++ private static class InstrumentedCoordinator extends MigrationCoordinator
++ {
++
++ Queue<Callback> requests = new LinkedList<>();
++ @Override
++ protected void sendMigrationMessage(MigrationCoordinator.Callback callback)
++ {
++ requests.add(callback);
++ }
++
++ boolean shouldPullSchema = true;
++ @Override
++ protected boolean shouldPullSchema(UUID version)
++ {
++ return shouldPullSchema;
++ }
++
++ boolean shouldPullFromEndpoint = true;
++ @Override
++ protected boolean shouldPullFromEndpoint(InetAddressAndPort endpoint)
++ {
++ return shouldPullFromEndpoint;
++ }
++
++ boolean shouldPullImmediately = true;
++ @Override
++ protected boolean shouldPullImmediately(InetAddressAndPort endpoint, UUID version)
++ {
++ return shouldPullImmediately;
++ }
++
++ Set<InetAddressAndPort> deadNodes = new HashSet<>();
++ protected boolean isAlive(InetAddressAndPort endpoint)
++ {
++ return !deadNodes.contains(endpoint);
++ }
++
++ UUID localVersion = LOCAL_VERSION;
++ @Override
++ protected boolean isLocalVersion(UUID version)
++ {
++ return localVersion.equals(version);
++ }
++
++ int maxOutstandingRequests = 3;
++ @Override
++ protected int getMaxOutstandingVersionRequests()
++ {
++ return maxOutstandingRequests;
++ }
++
++ Set<InetAddressAndPort> mergedSchemasFrom = new HashSet<>();
++ @Override
++ protected void mergeSchemaFrom(InetAddressAndPort endpoint, Collection<Mutation> mutations)
++ {
++ mergedSchemasFrom.add(endpoint);
++ }
++ }
++
++ @Test
++ public void requestResponseCycle() throws InterruptedException
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
++ coordinator.maxOutstandingRequests = 1;
++
++ Assert.assertTrue(coordinator.requests.isEmpty());
++
++ // first schema report should send a migration request
++ getUnchecked(coordinator.reportEndpointVersion(EP1, V1));
++ Assert.assertEquals(1, coordinator.requests.size());
++ Assert.assertFalse(coordinator.awaitSchemaRequests(1));
++
++ // second should not
++ getUnchecked(coordinator.reportEndpointVersion(EP2, V1));
++ Assert.assertEquals(1, coordinator.requests.size());
++ Assert.assertFalse(coordinator.awaitSchemaRequests(1));
++
++ // until the first request fails, then the second endpoint should be contacted
++ MigrationCoordinator.Callback request1 = coordinator.requests.poll();
++ Assert.assertEquals(EP1, request1.endpoint);
++ getUnchecked(request1.fail());
++ Assert.assertTrue(coordinator.mergedSchemasFrom.isEmpty());
++ Assert.assertFalse(coordinator.awaitSchemaRequests(1));
++
++ // ... then the second endpoint should be contacted
++ Assert.assertEquals(1, coordinator.requests.size());
++ MigrationCoordinator.Callback request2 = coordinator.requests.poll();
++ Assert.assertEquals(EP2, request2.endpoint);
++ Assert.assertFalse(coordinator.awaitSchemaRequests(1));
++ getUnchecked(request2.response(Collections.emptyList()));
++ Assert.assertEquals(EP2, Iterables.getOnlyElement(coordinator.mergedSchemasFrom));
++ Assert.assertTrue(coordinator.awaitSchemaRequests(1));
++
++ // and migration tasks should not be sent out for subsequent version reports
++ getUnchecked(coordinator.reportEndpointVersion(EP3, V1));
++ Assert.assertTrue(coordinator.requests.isEmpty());
++
++ }
++
++ /**
++ * If we don't send a request for a version, and endpoints associated with
++ * it all change versions, we should signal anyone waiting on that version
++ */
++ @Test
++ public void versionsAreSignaledWhenDeleted()
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
++
++ coordinator.reportEndpointVersion(EP1, V1);
++ WaitQueue.Signal signal = coordinator.getVersionInfoUnsafe(V1).register();
++ Assert.assertFalse(signal.isSignalled());
++
++ coordinator.reportEndpointVersion(EP1, V2);
++ Assert.assertNull(coordinator.getVersionInfoUnsafe(V1));
++
++ Assert.assertTrue(signal.isSignalled());
++ }
++
++ private static void assertNoContact(InstrumentedCoordinator coordinator, InetAddressAndPort endpoint, UUID version, boolean startupShouldBeUnblocked)
++ {
++ Assert.assertTrue(coordinator.requests.isEmpty());
++ Future<Void> future = coordinator.reportEndpointVersion(EP1, V1);
++ if (future != null)
++ getUnchecked(future);
++ Assert.assertTrue(coordinator.requests.isEmpty());
++
++ Assert.assertEquals(startupShouldBeUnblocked, coordinator.awaitSchemaRequests(1));
++ }
++
++ private static void assertNoContact(InstrumentedCoordinator coordinator, boolean startupShouldBeUnblocked)
++ {
++ assertNoContact(coordinator, EP1, V1, startupShouldBeUnblocked);
++ }
++
++ @Test
++ public void dontContactNodesWithSameSchema()
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
++
++ coordinator.localVersion = V1;
++ assertNoContact(coordinator, true);
++ }
++
++ @Test
++ public void dontContactIncompatibleNodes()
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
++
++ coordinator.shouldPullFromEndpoint = false;
++ assertNoContact(coordinator, false);
++ }
++
++ @Test
++ public void dontContactDeadNodes()
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
++
++ coordinator.deadNodes.add(EP1);
++ assertNoContact(coordinator, EP1, V1, false);
++ }
++
++ /**
++ * If a node has become incompativle between when the task was scheduled and when it
++ * was run, we should detect that and fail the task
++ */
++ @Test
++ public void testGossipRace()
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator() {
++ protected boolean shouldPullImmediately(InetAddressAndPort endpoint, UUID version)
++ {
++ // this is the last thing that gets called before scheduling the pull, so set this flag here
++ shouldPullFromEndpoint = false;
++ return super.shouldPullImmediately(endpoint, version);
++ }
++ };
++
++ Assert.assertTrue(coordinator.shouldPullFromEndpoint(EP1));
++ assertNoContact(coordinator, EP1, V1, false);
++ }
++
++ @Test
++ public void testWeKeepSendingRequests()
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
++
++ getUnchecked(coordinator.reportEndpointVersion(EP3, V2));
++ coordinator.requests.remove().response(Collections.emptyList());
++
++ getUnchecked(coordinator.reportEndpointVersion(EP1, V1));
++ getUnchecked(coordinator.reportEndpointVersion(EP2, V1));
++
++ MigrationCoordinator.Callback prev = null;
++ Set<InetAddressAndPort> EPs = Sets.newHashSet(EP1, EP2);
++ int ep1requests = 0;
++ int ep2requests = 0;
++
++ for (int i=0; i<10; i++)
++ {
++ Assert.assertEquals(String.format("%s", i), 2, coordinator.requests.size());
++
++ MigrationCoordinator.Callback next = coordinator.requests.remove();
++
++ // we should be contacting endpoints in a round robin fashion
++ Assert.assertTrue(EPs.contains(next.endpoint));
++ if (prev != null && prev.endpoint.equals(next.endpoint))
++ Assert.fail(String.format("Not expecting prev %s to be equal to next %s", prev.endpoint, next.endpoint));
++
++ // should send a new request
++ next.fail();
++ prev = next;
++ Assert.assertFalse(coordinator.awaitSchemaRequests(1));
++
++ Assert.assertEquals(2, coordinator.requests.size());
++ }
++ logger.info("{} -> {}", EP1, ep1requests);
++ logger.info("{} -> {}", EP2, ep2requests);
++
++ // a single success should unblock startup though
++ coordinator.requests.remove().response(Collections.emptyList());
++ Assert.assertTrue(coordinator.awaitSchemaRequests(1));
++
++ }
++
++ /**
++ * Pull unreceived schemas should detect and send requests out for any
++ * schemas that are marked unreceived and have no outstanding requests
++ */
++ @Test
++ public void pullUnreceived()
++ {
++ InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
++
++ coordinator.shouldPullFromEndpoint = false;
++ assertNoContact(coordinator, false);
++
++ coordinator.shouldPullFromEndpoint = true;
++ Assert.assertEquals(0, coordinator.requests.size());
++ List<Future<Void>> futures = coordinator.pullUnreceivedSchemaVersions();
++ futures.forEach(Futures::getUnchecked);
++ Assert.assertEquals(1, coordinator.requests.size());
++ }
++}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org