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

Improve CMS initialization 

* Better handling of DOWN unupgraded nodes
* Test that aborted CMS initialization cleans up state properly
* Clean up orphaned PreInitialize entries in the log on bounce
* Unconditionally reset initiator during abort
* Ensure that metadata log entries aren't exchanged before CMS
  initialization is complete
* Precalculate common serialization version, excluding non-upgraded and
  LEFT nodes
* Decide if metadata-impacting upgrade is in progress using min common
  version
* Add metadata identifier to nodetool cms output

Patch by Sam Tunnicliffe and Marcus Eriksson; reviewed by Sam
Tunnicliffe and Marcus Eriksson for CASSANDRA-21036

Co-authored-by: Marcus Eriksson <marcuse@apache.org>
Co-authored-by: Sam Tunnicliffe <samt@apache.org>
This commit is contained in:
Sam Tunnicliffe 2025-11-20 15:52:34 +00:00
parent b11633be4f
commit f05b27502f
21 changed files with 716 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.1
* Improve CMS initialization (CASSANDRA-21036)
* Introducing comments and security labels for schema elements (CASSANDRA-20943)
* Extend nodetool tablestats for dictionary memory usage (CASSANDRA-20940)
* Introduce separate GCInspector thresholds for concurrent GC events (CASSANDRA-20980)

2
src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java
View File

@ -56,7 +56,7 @@ public abstract class AbstractMutationVerbHandler<T extends IMutation> implement
protected void processMessage(Message<T> message, InetAddressAndPort respondTo)
{
if (message.epoch().isAfter(Epoch.EMPTY))
if (message.epoch().isAfter(Epoch.FIRST))
{
ClusterMetadata metadata = ClusterMetadata.current();
metadata = checkTokenOwnership(metadata, message, respondTo);

4
src/java/org/apache/cassandra/gms/GossipVerbHandler.java
View File

@ -21,12 +21,14 @@ package org.apache.cassandra.gms;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
public class GossipVerbHandler<T> implements IVerbHandler<T>
{
public void doVerb(Message<T> message)
{
Gossiper.instance.setLastProcessedMessageAt(message.creationTimeMillis());
ClusterMetadataService.instance().fetchLogFromPeerAsync(message.from(), message.epoch());
if (message.epoch().isAfter(Epoch.FIRST))
ClusterMetadataService.instance().fetchLogFromPeerAsync(message.from(), message.epoch());
}
}

4
src/java/org/apache/cassandra/net/ResponseVerbHandler.java
View File

@ -28,6 +28,7 @@ import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
@ -92,6 +93,9 @@ class ResponseVerbHandler implements IVerbHandler
private void maybeFetchLogs(Message<?> message)
{
ClusterMetadata metadata = ClusterMetadata.current();
if (!message.epoch().isAfter(Epoch.FIRST))
return;
if (!message.epoch().isAfter(metadata.epoch))
return;

2
src/java/org/apache/cassandra/tcm/CMSOperations.java
View File

@ -63,6 +63,7 @@ public class CMSOperations implements CMSOperationsMBean
public static final String LOCAL_PENDING = "LOCAL_PENDING";
public static final String COMMITS_PAUSED = "COMMITS_PAUSED";
public static final String REPLICATION_FACTOR = "REPLICATION_FACTOR";
public static final String CMS_ID = "CMS_ID";
private static final Logger logger = LoggerFactory.getLogger(ClusterMetadataService.class);
public static CMSOperations instance = new CMSOperations(ClusterMetadataService.instance());
@ -161,6 +162,7 @@ public class CMSOperations implements CMSOperationsMBean
info.put(LOCAL_PENDING, Integer.toString(cms.log().pendingBufferSize()));
info.put(COMMITS_PAUSED, Boolean.toString(cms.commitsPaused()));
info.put(REPLICATION_FACTOR, ReplicationParams.meta(metadata).toString());
info.put(CMS_ID, Integer.toString(metadata.metadataIdentifier));
return info;
}

2
src/java/org/apache/cassandra/tcm/ClusterMetadata.java
View File

@ -1004,7 +1004,7 @@ public class ClusterMetadata
public boolean metadataSerializationUpgradeInProgress()
{
return !directory.clusterMaxVersion.serializationVersion().equals(directory.clusterMinVersion.serializationVersion());
return !directory.clusterMaxVersion.serializationVersion().equals(directory.commonSerializationVersion);
}
public static class Serializer implements MetadataSerializer<ClusterMetadata>

5
src/java/org/apache/cassandra/tcm/Startup.java
View File

@ -58,8 +58,8 @@ import org.apache.cassandra.tcm.log.LogStorage;
import org.apache.cassandra.tcm.log.SystemKeyspaceStorage;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.migration.Election;
import org.apache.cassandra.tcm.migration.CMSInitializationRequest;
import org.apache.cassandra.tcm.migration.Election;
import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
import org.apache.cassandra.tcm.sequences.InProgressSequences;
import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
@ -77,7 +77,7 @@ import static org.apache.cassandra.tcm.membership.NodeState.JOINED;
import static org.apache.cassandra.tcm.membership.NodeState.LEFT;
import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
/**
/**
* Initialize
*/
public class Startup
@ -141,7 +141,6 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
ClusterMetadataService.instance().log().bootstrap(addr, datacenter);
ClusterMetadata metadata = ClusterMetadata.current();
assert ClusterMetadataService.state() == LOCAL : String.format("Can't initialize as node hasn't transitioned to CMS state. State: %s.\n%s", ClusterMetadataService.state(), metadata);
Initialize initialize = new Initialize(metadata.initializeClusterIdentifier(addr.hashCode()));
ClusterMetadataService.instance().commit(initialize);
}

19
src/java/org/apache/cassandra/tcm/log/SystemKeyspaceStorage.java
View File

@ -87,12 +87,25 @@ public class SystemKeyspaceStorage implements LogStorage
public synchronized static boolean hasAnyEpoch()
{
String query = String.format("SELECT epoch FROM %s.%s LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, NAME);
String query = String.format("SELECT epoch, kind FROM %s.%s LIMIT 2", SchemaConstants.SYSTEM_KEYSPACE_NAME, NAME);
int count = 0;
long preInitializeEpoch = -1;
for (UntypedResultSet.Row row : executeInternal(query))
return true;
{
count++;
if (Transformation.Kind.fromId(row.getInt("kind")) == Transformation.Kind.PRE_INITIALIZE_CMS)
preInitializeEpoch = row.getLong("epoch");
}
if (count == 1 && preInitializeEpoch != -1)
{
logger.warn("Cleaning up orphaned PreInitialize at epoch {} - restarting in gossip mode", preInitializeEpoch);
String cleanupQuery = String.format("DELETE from %s.%s where epoch = %d", SchemaConstants.SYSTEM_KEYSPACE_NAME, NAME, preInitializeEpoch);
executeInternal(cleanupQuery);
return false;
}
return false;
return count > 0;
}
@Override

20
src/java/org/apache/cassandra/tcm/membership/Directory.java
View File

@ -73,6 +73,7 @@ public class Directory implements MetadataValue<Directory>
private final BTreeMap<String, Multimap<String, InetAddressAndPort>> racksByDC;
public final NodeVersion clusterMinVersion;
public final NodeVersion clusterMaxVersion;
public final Version commonSerializationVersion;
public Directory()
{
@ -115,6 +116,7 @@ public class Directory implements MetadataValue<Directory>
Pair<NodeVersion, NodeVersion> minMaxVer = minMaxVersions(states, versions);
clusterMinVersion = minMaxVer.left;
clusterMaxVersion = minMaxVer.right;
commonSerializationVersion = minCommonSerializationVersion(states, versions);
}
@Override
@ -131,6 +133,9 @@ public class Directory implements MetadataValue<Directory>
", hostIds=" + hostIds +
", endpointsByDC=" + endpointsByDC +
", racksByDC=" + racksByDC +
", clusterMinVersion=" + clusterMinVersion +
", clusterMaxVersion=" + clusterMaxVersion +
", commonSerializationVersion=" + commonSerializationVersion +
'}';
}
@ -778,6 +783,21 @@ public class Directory implements MetadataValue<Directory>
return Pair.create(minVersion, maxVersion);
}
public static Version minCommonSerializationVersion(BTreeMap<NodeId, NodeState> states, BTreeMap<NodeId, NodeVersion> versions)
{
int commonVersion = Integer.MAX_VALUE;
for (Map.Entry<NodeId, NodeState> entry : states.entrySet())
{
if (entry.getValue() != NodeState.LEFT)
{
NodeVersion ver = versions.get(entry.getKey());
if (ver.serializationVersion > Version.OLD.asInt() && ver.serializationVersion < commonVersion)
commonVersion = ver.serializationVersion;
}
}
return commonVersion == Integer.MAX_VALUE ? NodeVersion.CURRENT_METADATA_VERSION : Version.fromInt(commonVersion);
}
@Override
public int hashCode()
{

17
src/java/org/apache/cassandra/tcm/membership/NodeVersion.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.tcm.membership;
import java.io.IOException;
import java.util.Objects;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
@ -39,6 +40,8 @@ public class NodeVersion implements Comparable<NodeVersion>
private static final CassandraVersion SINCE_VERSION = CassandraVersion.CASSANDRA_5_1;
public final CassandraVersion cassandraVersion;
// this must be kept as an int, otherwise we can't deserialize future NodeVersions
public final int serializationVersion;
public NodeVersion(CassandraVersion cassandraVersion, Version serializationVersion)
@ -113,10 +116,9 @@ public class NodeVersion implements Comparable<NodeVersion>
@Override
public void serialize(NodeVersion t, DataOutputPlus out, Version version) throws IOException
{
out.writeUTF(t.cassandraVersion.toString());
if (t.serializationVersion == Version.UNKNOWN.asInt())
throw new IllegalStateException("Should not serialize UNKNOWN version");
out.writeUnsignedVInt32(t.serializationVersion);
serializeHelper(out, t.cassandraVersion.toString(), t.serializationVersion);
}
@Override
@ -133,5 +135,16 @@ public class NodeVersion implements Comparable<NodeVersion>
return sizeof(t.cassandraVersion.toString()) +
sizeofUnsignedVInt(t.serializationVersion);
}
/**
* Used to be able to generate a NodeVersion with a future serialization version, for test!
*/
@VisibleForTesting
static void serializeHelper(DataOutputPlus out, String cassandraVersion, int serializationVersion) throws IOException
{
out.writeUTF(cassandraVersion);
out.writeUnsignedVInt32(serializationVersion);
}
}
}

23
src/java/org/apache/cassandra/tcm/migration/Election.java
View File

@ -89,25 +89,24 @@ public class Election
Set<InetAddressAndPort> sendTo = new HashSet<>(candidates);
sendTo.removeAll(ignoredEndpoints);
sendTo.remove(FBUtilities.getBroadcastAddressAndPort());
CMSInitializationRequest initializationRequest = new CMSInitializationRequest(FBUtilities.getBroadcastAddressAndPort(), UUID.randomUUID(), metadata);
if (!updateInitiator(null, initializationRequest.initiator))
throw new IllegalStateException("Migration already initiated by " + initiator.get());
try
{
initiate(sendTo, metadata, verifyAllPeersMetadata);
initiate(initializationRequest, sendTo, metadata, verifyAllPeersMetadata);
finish(sendTo);
}
catch (Exception e)
catch (Throwable e)
{
abort(sendTo);
logger.error("Got error nominating self", e);
abort(initializationRequest.initiator, sendTo);
throw e;
}
}
private void initiate(Set<InetAddressAndPort> sendTo, ClusterMetadata metadata, boolean verifyAllPeersMetadata)
private void initiate(CMSInitializationRequest initializationRequest, Set<InetAddressAndPort> sendTo, ClusterMetadata metadata, boolean verifyAllPeersMetadata)
{
CMSInitializationRequest initializationRequest = new CMSInitializationRequest(FBUtilities.getBroadcastAddressAndPort(), UUID.randomUUID(), metadata);
if (!updateInitiator(null, initializationRequest.initiator))
throw new IllegalStateException("Migration already initiated by " + initiator.get());
logger.info("No previous migration detected, initiating");
Collection<Pair<InetAddressAndPort, CMSInitializationResponse>> metadatas = MessageDelivery.fanoutAndWait(messaging, sendTo, Verb.TCM_INIT_MIG_REQ, initializationRequest);
if (metadatas.size() != sendTo.size())
@ -149,9 +148,11 @@ public class Election
}
}
private void abort(Set<InetAddressAndPort> sendTo)
private void abort(CMSInitializationRequest.Initiator init, Set<InetAddressAndPort> sendTo)
{
CMSInitializationRequest.Initiator init = initiator.getAndSet(null);
logger.info("Aborting migration");
CMSInitializationRequest.Initiator previous = initiator.getAndSet(null);
logger.info("Reset local initiator state (was {}), sending abort message to peers", previous);
for (InetAddressAndPort ep : sendTo)
messaging.send(Message.out(Verb.TCM_ABORT_MIG, init), ep);
}

26
src/java/org/apache/cassandra/tcm/serialization/MessageSerializers.java
View File

@ -35,47 +35,47 @@ import org.apache.cassandra.tcm.migration.CMSInitializationRequest;
* NOTE: Serialization version here is used for convenience of serializing the message
* on the outgoing path. Since receiving node may have a different view of
* min serialization version, we _always_ have to either use a {@link VerboseMetadataSerializer}
* (like {@link LogState}/ {@link Replication} or explicitly serialize the version (like {@link Commit}).
* (like {@link LogState}) or explicitly serialize the version (like {@link Commit}).
*/
public class MessageSerializers
{
public static IVersionedSerializer<LogState> logStateSerializer()
{
ClusterMetadata metadata = ClusterMetadata.currentNullable();
if (metadata == null || metadata.directory.clusterMinVersion.serializationVersion == NodeVersion.CURRENT.serializationVersion)
if (metadata == null || metadata.directory.commonSerializationVersion.asInt() == NodeVersion.CURRENT.serializationVersion)
return LogState.defaultMessageSerializer;
assert !metadata.directory.clusterMinVersion.serializationVersion().equals(NodeVersion.CURRENT.serializationVersion());
return LogState.messageSerializer(metadata.directory.clusterMinVersion.serializationVersion());
assert !metadata.directory.commonSerializationVersion.equals(NodeVersion.CURRENT.serializationVersion());
return LogState.messageSerializer(metadata.directory.commonSerializationVersion);
}
public static IVersionedSerializer<Commit.Result> commitResultSerializer()
{
ClusterMetadata metadata = ClusterMetadata.currentNullable();
if (metadata == null || metadata.directory.clusterMinVersion.serializationVersion == NodeVersion.CURRENT.serializationVersion)
if (metadata == null || metadata.directory.commonSerializationVersion.asInt() == NodeVersion.CURRENT.serializationVersion)
return Commit.Result.defaultMessageSerializer;
assert !metadata.directory.clusterMinVersion.serializationVersion().equals(NodeVersion.CURRENT.serializationVersion());
return Commit.Result.messageSerializer(metadata.directory.clusterMinVersion.serializationVersion());
assert !metadata.directory.commonSerializationVersion.equals(NodeVersion.CURRENT.serializationVersion());
return Commit.Result.messageSerializer(metadata.directory.commonSerializationVersion);
}
public static IVersionedSerializer<Commit> commitSerializer()
{
ClusterMetadata metadata = ClusterMetadata.currentNullable();
if (metadata == null || metadata.directory.clusterMinVersion.serializationVersion == NodeVersion.CURRENT.serializationVersion)
if (metadata == null || metadata.directory.commonSerializationVersion.asInt() == NodeVersion.CURRENT.serializationVersion)
return Commit.defaultMessageSerializer;
assert !metadata.directory.clusterMinVersion.serializationVersion().equals(NodeVersion.CURRENT.serializationVersion());
return Commit.messageSerializer(metadata.directory.clusterMinVersion.serializationVersion());
assert !metadata.directory.commonSerializationVersion.equals(NodeVersion.CURRENT.serializationVersion());
return Commit.messageSerializer(metadata.directory.commonSerializationVersion);
}
public static IVersionedSerializer<CMSInitializationRequest> initRequestSerializer()
{
ClusterMetadata metadata = ClusterMetadata.currentNullable();
if (metadata == null || metadata.directory.clusterMinVersion.serializationVersion == NodeVersion.CURRENT.serializationVersion)
if (metadata == null || metadata.directory.commonSerializationVersion.asInt() == NodeVersion.CURRENT.serializationVersion)
return CMSInitializationRequest.defaultMessageSerializer;
assert !metadata.directory.clusterMinVersion.serializationVersion().equals(NodeVersion.CURRENT.serializationVersion());
return CMSInitializationRequest.messageSerializer(metadata.directory.clusterMinVersion.serializationVersion());
assert !metadata.directory.commonSerializationVersion.equals(NodeVersion.CURRENT.serializationVersion());
return CMSInitializationRequest.messageSerializer(metadata.directory.commonSerializationVersion);
}
}

24
src/java/org/apache/cassandra/tcm/serialization/Version.java
View File

@ -29,7 +29,13 @@ import org.apache.cassandra.tcm.membership.NodeVersion;
public enum Version
{
/**
* Used for nodes on older, pre-CEP-21 release versions
*/
OLD(-1),
/**
* Initial version
*/
V0(0),
/**
* - Moved Partitioner in ClusterMetadata serializer to be the first field
@ -81,6 +87,7 @@ public enum Version
public static final Version MIN_ACCORD_VERSION = V7;
private static Map<Integer, Version> values = new HashMap<>();
static
{
for (Version v : values())
@ -99,10 +106,9 @@ public enum Version
public static Version minCommonSerializationVersion()
{
ClusterMetadata metadata = ClusterMetadata.currentNullable();
if (metadata != null)
return metadata.directory.clusterMinVersion.serializationVersion();
return NodeVersion.CURRENT.serializationVersion();
return metadata != null
? metadata.directory.commonSerializationVersion
: NodeVersion.CURRENT.serializationVersion();
}
public int asInt()
@ -120,11 +126,21 @@ public enum Version
return version >= other.version;
}
public boolean isEqualOrBefore(Version version)
{
return this.version <= version.version;
}
public boolean isBefore(Version other)
{
return version < other.version;
}
public boolean isAfter(Version other)
{
return version > other.version;
}
public static Version fromInt(int i)
{
Version v = values.get(i);

6
src/java/org/apache/cassandra/tcm/transformations/cms/Initialize.java
View File

@ -61,6 +61,12 @@ public class Initialize extends ForceSnapshot
super(baseState);
}
@Override
public boolean allowDuringUpgrades()
{
return true;
}
public Kind kind()
{
return Kind.INITIALIZE_CMS;

2
src/java/org/apache/cassandra/tools/nodetool/CMSAdmin.java
View File

@ -34,6 +34,7 @@ import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
import picocli.CommandLine.Parameters;
import static org.apache.cassandra.tcm.CMSOperations.CMS_ID;
import static org.apache.cassandra.tcm.CMSOperations.COMMITS_PAUSED;
import static org.apache.cassandra.tcm.CMSOperations.EPOCH;
import static org.apache.cassandra.tcm.CMSOperations.IS_MEMBER;
@ -73,6 +74,7 @@ public class CMSAdmin extends AbstractCommand
{
Map<String, String> info = probe.getCMSOperationsProxy().describeCMS();
output.out.printf("Cluster Metadata Service:%n");
output.out.printf("CMS Identifier: %s%n", info.get(CMS_ID));
output.out.printf("Members: %s%n", info.get(MEMBERS));
output.out.printf("Needs reconfiguration: %s%n", info.get(NEEDS_RECONFIGURATION));
output.out.printf("Is Member: %s%n", info.get(IS_MEMBER));

44
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeIgnoreHostTest.java → test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeCleanupPreInitializeTest.java
View File

@ -22,11 +22,16 @@ import org.junit.Test;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.tcm.Transformation;
public class ClusterMetadataUpgradeIgnoreHostTest extends UpgradeTestBase
import static org.junit.Assert.assertTrue;
public class ClusterMetadataUpgradeCleanupPreInitializeTest extends UpgradeTestBase
{
@Test
public void upgradeIgnoreHostsTest() throws Throwable
public void cleanupPreInitializeTest() throws Throwable
{
new TestCase()
.nodes(3)
@ -39,12 +44,35 @@ public class ClusterMetadataUpgradeIgnoreHostTest extends UpgradeTestBase
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
})
.runAfterClusterUpgrade((cluster) -> {
// todo; isolate node 3 - actually shutting it down makes us throw exceptions when test finishes
cluster.filters().allVerbs().to(3).drop();
cluster.filters().allVerbs().from(3).drop();
cluster.get(1).nodetoolResult("cms", "initialize").asserts().failure(); // node3 unreachable
cluster.get(1).nodetoolResult("cms", "initialize", "--ignore", "127.0.0.1").asserts().failure(); // can't ignore localhost
cluster.get(1).nodetoolResult("cms", "initialize", "--ignore", "127.0.0.3").asserts().success();
cluster.get(1).executeInternal("INSERT INTO system.local_metadata_log (epoch, kind) VALUES (1, 0)");
cluster.get(1).flush("system");
cluster.get(1).shutdown().get();
cluster.get(1).startup();
cluster.get(1).logs().watchFor("Cleaning up orphaned PreInitialize at epoch 1");
cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
cluster.get(1).shutdown().get();
cluster.get(1).startup();
boolean seenPreInit = false;
boolean seenInit = false;
boolean seenSnapshot = false;
for (Object [] row : cluster.get(1).executeInternal("SELECT epoch, kind FROM system.local_metadata_log"))
{
switch (Transformation.Kind.fromId((Integer)row[1]))
{
case PRE_INITIALIZE_CMS:
seenPreInit = true;
break;
case INITIALIZE_CMS:
seenInit = true;
break;
case TRIGGER_SNAPSHOT:
seenSnapshot = true;
break;
}
}
assertTrue(seenPreInit);
assertTrue(seenInit);
assertTrue(seenSnapshot);
}).run();
}
}

219
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDelayedInitializeTest.java Normal file
View File

@ -0,0 +1,219 @@
/*
* 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.upgrade;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IUpgradeableInstance;
import org.apache.cassandra.distributed.api.NodeToolResult;
import org.apache.cassandra.distributed.shared.Versions;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Shared;
import org.awaitility.Awaitility;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.junit.Assert.assertEquals;
public class ClusterMetadataUpgradeDelayedInitializeTest extends UpgradeTestBase
{
private static final Logger logger = LoggerFactory.getLogger(ClusterMetadataUpgradeDelayedInitializeTest.class);
@Test
public void delayedInitializationTest() throws Throwable
{
// In this test we force the initiator to pause between committing the PreInitialize and Initialize, then verify
// that read/write/gossip messages exchanged whilst in this state do not cause the other nodes to fetch the
// PreInitialize log entry. We inject that pause using ByteBuddy to add a wait on a CountdownLatch controlled
// from the main test code.
final int nodeCount = 3;
Versions dtestVersions = Versions.find();
Versions.Version earliestVersion = dtestVersions.getLatest(v41);
Consumer<IInstanceConfig> configUpdater = config -> config.with(Feature.NETWORK, Feature.GOSSIP);
Consumer<UpgradeableCluster.Builder> builderUpdater = builder -> builder.withInstanceInitializer(ClusterMetadataUpgradeDelayedInitializeTest.BBInstaller::installUpgradeVersionBB);
try (UpgradeableCluster cluster = UpgradeableCluster.create(nodeCount, earliestVersion, configUpdater, builderUpdater))
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':3}"));
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (k int PRIMARY KEY)");
// Upgrade all instances
Versions.Version upgradeVersion = dtestVersions.getLatest(CURRENT);
for (IUpgradeableInstance i : cluster)
upgradeInstance(i, upgradeVersion);
logger.info(String.format("Upgrade complete. Current version: %s", upgradeVersion));
((IInvokableInstance)cluster.get(2)).runOnInstance(() -> {
KeyspaceMetadata ksm = ClusterMetadata.current().schema.getKeyspaceMetadata("distributed_test_keyspace");
TableMetadata tm = ksm.tables.getNullable("tbl");
System.out.println(tm.epoch);
});
// Start the CMS initialization, which should pause after committing the PreInitialize transform
assertEquals(1, BBState.latch.getCount());
ExecutorService initExecutor = Executors.newSingleThreadExecutor();
IUpgradeableInstance initiator = cluster.get(1);
Future<NodeToolResult> result = initExecutor.submit(() -> initiator.nodetoolResult("cms", "initialize"));
logger.info("Waiting for initiator to pause");
Awaitility.waitAtMost(30, TimeUnit.SECONDS).until(() -> {
// wait until the intiator has committed the PreInitialize entry
NodeToolResult res = initiator.nodetoolResult("cms");
return res.getStdout().contains("Epoch: 1");
});
logger.info("Initiator paused");
// internode messages for reads/writes should not trigger catchup between node1 and its peers
for (int i = 1; i <= 3; i++)
{
cluster.coordinator(i).execute(withKeyspace("INSERT INTO %s.tbl (k) VALUES (0)"), ConsistencyLevel.ALL);
cluster.coordinator(i).execute(withKeyspace("SELECT * FROM %s.tbl WHERE k=0"), ConsistencyLevel.ALL);
}
// nor should gossip messages
awaitGossipUpdateFromPeer(cluster.get(2), initiator);
awaitGossipUpdateFromPeer(cluster.get(3), initiator);
// Assert that nodes 2 & 3 did not catch up from node 1
cluster.get(2).nodetoolResult("cms").asserts().success().stdoutContains("Epoch: -9223372036854775807");
cluster.get(3).nodetoolResult("cms").asserts().success().stdoutContains("Epoch: -9223372036854775807");
logger.info("Allowing initiator to continue");
BBState.latch.countDown();
NodeToolResult initRes = result.get(30, TimeUnit.SECONDS);
initRes.asserts().success();
// Verify that the initialization completes
Awaitility.waitAtMost(30, TimeUnit.SECONDS)
.until(() -> cluster.get(1).nodetoolResult("cms").getStdout().contains("Epoch: 3") &&
cluster.get(2).nodetoolResult("cms").getStdout().contains("Epoch: 3") &&
cluster.get(3).nodetoolResult("cms").getStdout().contains("Epoch: 3"));
}
}
/**
* Inspects gossip state on one peer and waits for the heartbeat state of another peer to increment
* (which is not an absolute guarantee of direct communication between the two nodes, but should be good enough).
*
* @param waiter the node doing the checking & waiting to observe an update
* @param waitingFor the node who's heartbeat being observed
*/
private void awaitGossipUpdateFromPeer(IUpgradeableInstance waiter, IUpgradeableInstance waitingFor)
{
String targetEndpoint = waitingFor.config().getString("broadcast_address");
((IInvokableInstance) waiter).runOnInstance(() -> {
InetAddressAndPort endpoint = InetAddressAndPort.getByNameUnchecked(targetEndpoint);
final int before = Gossiper.instance.getEndpointStateForEndpoint(endpoint)
.getHeartBeatState()
.getHeartBeatVersion();
long deadline = System.nanoTime() + TimeUnit.NANOSECONDS.convert(30, TimeUnit.SECONDS);
while (Gossiper.instance.getEndpointStateForEndpoint(endpoint)
.getHeartBeatState()
.getHeartBeatVersion() < before + 1)
{
FBUtilities.sleepQuietly(100);
if (System.nanoTime() > deadline)
throw new RuntimeException("Timed out waiting for gossip state to update");
}
});
}
private void upgradeInstance(IUpgradeableInstance instance, Versions.Version upgradeTo) throws ExecutionException, InterruptedException, TimeoutException
{
int instanceId = instance.config().num();
logger.info("Shutting down instance {} to upgrade to {}", instanceId, upgradeTo.version);
instance.shutdown(true).get(60, TimeUnit.SECONDS);
logger.info("Starting instanceId {} on version {}", instanceId, upgradeTo.version);
instance.setVersion(upgradeTo);
instance.startup();
logger.info("Started instanceId {}", instanceId);
}
@Shared
public static class BBState
{
public static CountDownLatch latch = new CountDownLatch(1);
}
public static class BBInstaller
{
public static void installUpgradeVersionBB(ClassLoader classLoader, Integer num)
{
try
{
new ByteBuddy().rebase(ClusterMetadata.class)
.method(named("initializeClusterIdentifier"))
.intercept(MethodDelegation.to(ClusterMetadataUpgradeDelayedInitializeTest.BBInterceptor.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
}
catch (NoClassDefFoundError noClassDefFoundError)
{
logger.info("... but no class def", noClassDefFoundError);
}
catch (Throwable tr)
{
logger.info("Unable to intercept upgradeFromVersion method", tr);
throw tr;
}
}
}
public static class BBInterceptor
{
@SuppressWarnings("unused")
public static ClusterMetadata initializeClusterIdentifier(@SuperCall Callable<ClusterMetadata> zuper)
{
try
{
logger.info("initializeClusterIdentifier waiting...");
BBState.latch.await(60, TimeUnit.SECONDS);
logger.info("initializeClusterIdentifier continuing...");
return zuper.call();
}
catch (Throwable e)
{
logger.info("error:", e);
throw new RuntimeException(e);
}
}
}
}

107
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeIgnoreHostsTest.java Normal file
View File

@ -0,0 +1,107 @@
/*
* 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.upgrade;
import java.util.HashSet;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.junit.Test;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.shared.Versions;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
public class ClusterMetadataUpgradeIgnoreHostsTest extends UpgradeTestBase
{
@Test
public void upgradeIgnoreHostsTest() throws Throwable
{
new TestCase()
.nodes(3)
.nodesToUpgrade(1, 2, 3)
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.setup((cluster) -> {
cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':2}"));
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
})
.runAfterClusterUpgrade((cluster) -> {
// todo; isolate node 3 - actually shutting it down makes us throw exceptions when test finishes
cluster.filters().allVerbs().to(3).drop();
cluster.filters().allVerbs().from(3).drop();
cluster.get(1).nodetoolResult("cms", "initialize").asserts().failure(); // node3 unreachable
cluster.get(1).nodetoolResult("cms", "initialize", "--ignore", "127.0.0.1").asserts().failure(); // can't ignore localhost
cluster.get(1).nodetoolResult("cms", "initialize", "--ignore", "127.0.0.3").asserts().success();
cluster.get(1).nodetoolResult("cms", "reconfigure", "2").asserts().success();
}).run();
}
@Test
public void upgradeIgnoreHostsNonUpgradedTest() throws Throwable
{
new TestCase()
.nodes(3)
.nodesToUpgrade(1, 2)
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.setup((cluster) -> {
cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':2}"));
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
})
.runAfterClusterUpgrade((cluster) -> {
cluster.filters().allVerbs().to(3).drop();
cluster.filters().allVerbs().from(3).drop();
cluster.get(1).nodetoolResult("cms", "initialize", "--ignore", "127.0.0.3").asserts().success();
cluster.get(3).shutdown().get();
cluster.filters().reset();
cluster.get(3).setVersion(Versions.find().getLatest(CURRENT));
cluster.get(3).startup();
cluster.schemaChange(withKeyspace("ALTER TABLE " + KEYSPACE + ".tbl with comment = 'test'"));
((IInvokableInstance)cluster.get(3)).runOnInstance(() -> {
Epoch current = ClusterMetadata.current().epoch;
if (current.isBefore(Epoch.FIRST))
throw new AssertionError("Epoch was not incremented as expected, still at " + current);
});
// Verify that CMS identifier has propagated across the nodes as this asserts that the DOWN node
// did not affect the common serialization version for metadata.
Set<Long> identifiers = new HashSet<>();
Pattern p = Pattern.compile(".*CMS Identifier\\: ([\\d]*).*", Pattern.DOTALL);
cluster.forEach(i -> {
Matcher m = p.matcher(i.nodetoolResult("cms").getStdout());
assertTrue(m.matches());
identifiers.add(Long.parseLong(m.group(1)));
});
assertEquals(1, identifiers.size());
assertNotEquals(ClusterMetadata.EMPTY_METADATA_IDENTIFIER, (long) identifiers.iterator().next());
cluster.get(1).nodetoolResult("cms", "reconfigure", "2").asserts().success();
}).run();
}
}

110
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeUnexpectedFailureTest.java Normal file
View File

@ -0,0 +1,110 @@
/*
* 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.upgrade;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import org.junit.Test;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.migration.Election;
import static net.bytebuddy.matcher.ElementMatchers.named;
public class ClusterMetadataUpgradeUnexpectedFailureTest extends UpgradeTestBase
{
@Test
public void upgradeFailsUnexpectedlyAfterInitiation() throws Throwable
{
Consumer<UpgradeableCluster.Builder > builderUpdater = builder -> builder.withInstanceInitializer(BBInstaller::installUpgradeVersionBB);
new TestCase()
.nodes(3)
.nodesToUpgrade(1, 2, 3)
.withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP)
.set(Constants.KEY_DTEST_FULL_STARTUP, true))
.upgradesToCurrentFrom(v41)
.withBuilder(builderUpdater)
.setup((cluster) -> {
cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor':2}"));
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
})
.runAfterClusterUpgrade((cluster) -> {
// we injected a BB helper to trigger an unexpected failure on the first attempt at initialization.
// i.e. an exception that must be caught as opposed to a mismatch in metadata or down peer.
cluster.get(1).nodetoolResult("cms", "initialize").asserts().failure().errorContains("Something unexpected went wrong");
// handling the failure should have included cleaning up any state so that another attempt can be
// made, which this time should succeed.
cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
}).run();
}
public static class BBInstaller
{
public static void installUpgradeVersionBB(ClassLoader classLoader, Integer num)
{
try
{
new ByteBuddy().rebase(Election.class)
.method(named("finish"))
.intercept(MethodDelegation.to(BBInterceptor.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
}
catch (NoClassDefFoundError noClassDefFoundError)
{
throw noClassDefFoundError;
}
catch (Throwable tr)
{
throw tr;
}
}
}
public static class BBInterceptor
{
private static final AtomicBoolean firstAttempt = new AtomicBoolean(true);
@SuppressWarnings("unused")
public static void finish(Set<InetAddressAndPort> sendTo, @SuperCall Callable<Void> zuper)
{
if (firstAttempt.getAndSet(false))
throw new IllegalStateException("Something unexpected went wrong");
try
{
zuper.call();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
}
}

77
test/unit/org/apache/cassandra/tcm/membership/DirectoryTest.java
View File

@ -20,6 +20,9 @@ package org.apache.cassandra.tcm.membership;
import org.junit.Test;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.CassandraVersion;
import static org.apache.cassandra.tcm.membership.MembershipUtils.endpoint;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@ -111,6 +114,80 @@ public class DirectoryTest
assertTrue(dir.allDatacenterRacks().isEmpty());
}
@Test
public void commonSerializationVersionTest()
{
int cnt = 1;
Location DC1_R1 = new Location("datacenter1", "rack1");
Directory dir = new Directory();
assertTrue(dir.isEmpty());
assertEquals(NodeVersion.CURRENT_METADATA_VERSION, dir.commonSerializationVersion);
dir = dir.with(addresses(cnt++), DC1_R1, NodeVersion.CURRENT);
assertEquals(NodeVersion.CURRENT.serializationVersion, dir.commonSerializationVersion.asInt());
dir = dir.with(addresses(cnt++), DC1_R1, withSerializationVersion(Version.V4));
assertEquals(Version.V4, dir.commonSerializationVersion);
dir = dir.with(addresses(cnt++), DC1_R1, withSerializationVersion(Version.V3));
assertEquals(Version.V3, dir.commonSerializationVersion);
dir = dir.with(addresses(cnt++), DC1_R1, withSerializationVersion(Version.V2));
assertEquals(Version.V2, dir.commonSerializationVersion);
dir = dir.with(addresses(cnt++), DC1_R1, withSerializationVersion(Version.V1));
assertEquals(Version.V1, dir.commonSerializationVersion);
// int v
dir = dir.with(addresses(cnt++), DC1_R1, withSerializationVersion(Version.V0));
assertEquals(Version.V0, dir.commonSerializationVersion);
// Adding another, higher version doesn't affect the floor
dir = dir.with(addresses(cnt++), DC1_R1, NodeVersion.CURRENT);
assertEquals(Version.V0, dir.commonSerializationVersion);
// Version.OLD for pre-CEP-21 nodes doesn't affect the floor
dir = dir.with(addresses(cnt), DC1_R1, withSerializationVersion(Version.OLD));
assertEquals(Version.V0, dir.commonSerializationVersion);
// remove all peers except the first, common version should revert to CURRENT
while (cnt > 1)
dir = dir.without(dir.lastModified().nextEpoch(), new NodeId(cnt--));
assertEquals(NodeVersion.CURRENT.serializationVersion, dir.commonSerializationVersion.asInt());
}
@Test
public void commonSerializationVersionLEFTNodesTest()
{
int cnt = 0;
Location DC1_R1 = new Location("datacenter1", "rack1");
Directory dir = new Directory();
assertTrue(dir.isEmpty());
dir = dir.with(addresses(cnt++), DC1_R1, NodeVersion.CURRENT);
dir = dir.with(addresses(cnt++), DC1_R1, NodeVersion.CURRENT);
dir = dir.with(addresses(cnt++), DC1_R1, NodeVersion.CURRENT);
NodeAddresses leavingEndpoint = addresses(cnt++);
dir = dir.with(leavingEndpoint, DC1_R1, withSerializationVersion(Version.V4));
assertEquals(Version.V4, dir.commonSerializationVersion);
NodeId leavingNode = dir.peerId(leavingEndpoint.broadcastAddress);
dir = dir.withNodeState(leavingNode, NodeState.LEFT);
assertEquals(NodeVersion.CURRENT_METADATA_VERSION, dir.commonSerializationVersion);
}
private static NodeAddresses addresses(int i)
{
return new NodeAddresses(endpoint(i));
}
private static NodeVersion withSerializationVersion(Version version)
{
return new NodeVersion(CassandraVersion.NULL_VERSION, version);
}
private void assertInvalidLocationUpdate(Directory dir, NodeId nodeId, Location loc, String message)
{
try

49
test/unit/org/apache/cassandra/tcm/membership/NodeVersionTest.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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.tcm.membership;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.tcm.serialization.Version;
import org.junit.Test;
import java.io.IOException;
import java.nio.ByteBuffer;
import static org.junit.Assert.assertEquals;
public class NodeVersionTest
{
@Test
public void futureNodeVersionTest() throws IOException
{
ByteBuffer bb;
try (DataOutputBuffer dob = new DataOutputBuffer())
{
NodeVersion.Serializer.serializeHelper(dob, "8.0.0", NodeVersion.CURRENT.serializationVersion + 1);
bb = dob.asNewBuffer();
}
try (DataInputBuffer in = new DataInputBuffer(bb, false))
{
NodeVersion n = NodeVersion.serializer.deserialize(in, NodeVersion.CURRENT.serializationVersion());
assertEquals(n.serializationVersion().asInt(), Version.UNKNOWN.asInt());
}
}
}