diff --git a/CHANGES.txt b/CHANGES.txt index da8ceaa57e..376f4c8291 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.25: + * Avoid pushing schema mutations when setting up distributed system keyspaces locally (CASSANDRA-16387) Merged from 2.2: * Make TokenMetadata's ring version increments atomic (CASSANDRA-16286) diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java index 2b0c83481b..fba3a221fd 100644 --- a/src/java/org/apache/cassandra/service/MigrationManager.java +++ b/src/java/org/apache/cassandra/service/MigrationManager.java @@ -414,11 +414,6 @@ public class MigrationManager announce(SchemaKeyspace.makeDropAggregateMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); } - static void announceGlobally(Mutation schema) - { - announce(Collections.singletonList(schema), false); - } - /** * actively announce a new version to active hosts via rpc * @param schema The schema mutation to be applied @@ -447,13 +442,7 @@ public class MigrationManager // Returns a future on the local application of the schema private static Future announce(final Collection schema) { - Future f = StageManager.getStage(Stage.MIGRATION).submit(new WrappedRunnable() - { - protected void runMayThrow() throws ConfigurationException - { - SchemaKeyspace.mergeSchemaAndAnnounceVersion(schema); - } - }); + Future f = announceWithoutPush(schema); for (InetAddress endpoint : Gossiper.instance.getLiveMembers()) { @@ -467,6 +456,17 @@ public class MigrationManager return f; } + public static Future announceWithoutPush(Collection schema) + { + return StageManager.getStage(Stage.MIGRATION).submit(new WrappedRunnable() + { + protected void runMayThrow() throws ConfigurationException + { + SchemaKeyspace.mergeSchemaAndAnnounceVersion(schema); + } + }); + } + /** * Announce my version passively over gossip. * Used to notify nodes as they arrive in the cluster. diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index a72b530c8e..31db74f06d 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -1077,7 +1077,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE @VisibleForTesting public void ensureTraceKeyspace() { - evolveSystemKeyspace(TraceKeyspace.metadata(), TraceKeyspace.GENERATION).ifPresent(MigrationManager::announceGlobally); + Optional mutation = evolveSystemKeyspace(TraceKeyspace.metadata(), TraceKeyspace.GENERATION); + mutation.ifPresent(value -> FBUtilities.waitOnFuture(MigrationManager.announceWithoutPush(Collections.singleton(value)))); } public static boolean isReplacingSameAddress() @@ -1146,7 +1147,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE if (!authSetupCalled.getAndSet(true)) { if (setUpSchema) - evolveSystemKeyspace(AuthKeyspace.metadata(), AuthKeyspace.GENERATION).ifPresent(MigrationManager::announceGlobally); + { + Optional mutation = evolveSystemKeyspace(AuthKeyspace.metadata(), AuthKeyspace.GENERATION); + mutation.ifPresent(value -> FBUtilities.waitOnFuture(MigrationManager.announceWithoutPush(Collections.singleton(value)))); + } DatabaseDescriptor.getRoleManager().setup(); DatabaseDescriptor.getAuthenticator().setup(); @@ -1165,7 +1169,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE evolveSystemKeyspace( AuthKeyspace.metadata(), AuthKeyspace.GENERATION).ifPresent(changes::add); if (!changes.isEmpty()) - MigrationManager.announce(changes, false); + FBUtilities.waitOnFuture(MigrationManager.announceWithoutPush(changes)); } public boolean isJoined() diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index afa9574a13..ede075f41a 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -103,7 +103,7 @@ import org.reflections.Reflections; */ public abstract class AbstractCluster implements ICluster, AutoCloseable { - public static Versions.Version CURRENT_VERSION = new Versions.Version(FBUtilities.getReleaseVersionString(), Versions.getClassPath());; + public static Versions.Version CURRENT_VERSION = new Versions.Version(FBUtilities.getReleaseVersionString(), Versions.getClassPath()); // WARNING: we have this logger not (necessarily) for logging, but // to ensure we have instantiated the main classloader's LoggerFactory (and any LogbackStatusListener) @@ -492,7 +492,11 @@ public abstract class AbstractCluster implements ICluster { try (SchemaChangeMonitor monitor = new SchemaChangeMonitor()) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java b/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java index 5c0e3b3f93..c51bf7b3f9 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java @@ -31,7 +31,10 @@ import org.apache.cassandra.net.MessagingService; import static com.google.common.collect.Iterables.getOnlyElement; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.junit.matchers.JUnitMatchers.containsString; public final class InternodeEncryptionEnforcementTest extends TestBaseImpl { @@ -61,6 +64,16 @@ public final class InternodeEncryptionEnforcementTest extends TestBaseImpl try (Cluster cluster = builder.start()) { + try + { + openConnections(cluster); + fail("Instances should not be able to connect, much less complete a schema change."); + } + catch (RuntimeException ise) + { + assertThat(ise.getMessage(), containsString("agreement not reached")); + } + /* * instance (1) won't connect to (2), since (2) won't have a TLS listener; * instance (2) won't connect to (1), since inbound check will reject @@ -113,6 +126,8 @@ public final class InternodeEncryptionEnforcementTest extends TestBaseImpl try (Cluster cluster = builder.start()) { + openConnections(cluster); + /* * instance (1) should connect to instance (2) without any issues; * instance (2) should connect to instance (1) without any issues. @@ -134,4 +149,13 @@ public final class InternodeEncryptionEnforcementTest extends TestBaseImpl cluster.get(2).runOnInstance(runnable); } } + + private void openConnections(Cluster cluster) + { + cluster.schemaChange("CREATE KEYSPACE test_connections_from_1 " + + "WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};", false, cluster.get(1)); + + cluster.schemaChange("CREATE KEYSPACE test_connections_from_2 " + + "WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};", false, cluster.get(2)); + } }