natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Avoid pushing schema mutations when setting up distributed system keyspaces locally 

patch by Caleb Rackliffe; reviewed by Aleksey Yeschenko, Benjamin Lerer
and Brandon Williams for CASSANDRA-16387
This commit is contained in:
Caleb Rackliffe 2021-02-03 13:08:10 -06:00 committed by Benjamin Lerer
parent e538df5e50
commit 1f686fd634
5 changed files with 49 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

24
src/java/org/apache/cassandra/service/MigrationManager.java
View File

@ -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<Mutation> 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<Mutation> 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.

10
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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> 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> 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()

6
test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
View File

@ -103,7 +103,7 @@ import org.reflections.Reflections;
*/
public abstract class AbstractCluster<I extends IInstance> implements ICluster<I>, 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<I extends IInstance> implements ICluster<I
private void schemaChange(String query, boolean ignoreStoppedInstances)
{
I instance = ignoreStoppedInstances ? getFirstRunningInstance() : get(1);
schemaChange(query, ignoreStoppedInstances, instance);
}
public void schemaChange(String query, boolean ignoreStoppedInstances, I instance)
{
instance.sync(() -> {
try (SchemaChangeMonitor monitor = new SchemaChangeMonitor())
{

24
test/distributed/org/apache/cassandra/distributed/test/InternodeEncryptionEnforcementTest.java
View File

@ -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));
}
}