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

[CASSANDRA-20736] Rework CMS initialization

This commit is contained in:
Sam Tunnicliffe 2026-01-23 12:39:01 +00:00
parent 4af3e47b2b
commit f54b49c2a4
5 changed files with 70 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -362,10 +362,38 @@ public class ClusterMetadata
capLastModified(cmsMembership, epoch));
}
public ClusterMetadata initializeClusterIdentifier(int clusterIdentifier,
NodeAddresses addresses,
NodeVersion version,
Location location)
/**
* To be used only during the execute of PRE_INITIALIZE_CMS, this sets the DataPlacement for the metadata keyspace
* so that the global log can be initialized and the subsequent entry containing the INITIALIZE_CMS transformation
* can be committed.
* @param initialCMSPlacement Expected to be a singleton placement identifying the local node as the sole replica
* for the metadata keyspace
* @return ClusterMetadata instance in the correct state to constitute the result of the PRE_INITIALIZE_CMS
* transformation
*/
public ClusterMetadata forcePreInitializedState(DataPlacement initialCMSPlacement)
{
assert epoch.isEqualOrBefore(Epoch.FIRST);
// double check that all metadata elements have had the last modified epoch set correctly
ClusterMetadata initial = forceEpoch(Epoch.FIRST);
initial.cmsDataPlacement = initialCMSPlacement;
return initial;
}
/**
* Produce a ClusterMetadata suitable for use as the base state in the INITIALIZE_CMS transformation. This should
* only be used on the first CMS node when bootstrapping the CMS after upgrade or in a brand new cluster.
* @param clusterIdentifier Unique identifier for split brain detection & protection
* @param addresses The NodeAddresses of the first CMS node.
* @param version Version info for the first CMS node.
* @param location The rack & DC of the first CMS node.
* @return ClusterMetadata instance in the correct state to constitute the base state of the INITIALIZE_CMS
* transformation
*/
public ClusterMetadata forceInitializedState(int clusterIdentifier,
NodeAddresses addresses,
NodeVersion version,
Location location)
{
if (this.metadataIdentifier != EMPTY_METADATA_IDENTIFIER)
throw new IllegalStateException(String.format("Can only initialize cluster identifier once, but it was already set to %d", this.metadataIdentifier));

1
src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
View File

@ -178,6 +178,7 @@ public class ClusterMetadataService
// the distributed metadata table.
if (metadata.epoch.isEqualOrBefore(Epoch.FIRST) || ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort()))
return LOCAL;
return REMOTE;
}

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

@ -151,26 +151,22 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
*/
public static void initializeAsFirstCMSNode()
{
InetAddressAndPort addr = FBUtilities.getBroadcastAddressAndPort();
String datacenter = DatabaseDescriptor.getLocator().local().datacenter;
ClusterMetadataService cms = ClusterMetadataService.instance();
cms.log().bootstrap(addr, datacenter, cms.logBootstrapCallback());
NodeAddresses addresses = NodeAddresses.current();
Location location = DatabaseDescriptor.getLocator().local();
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(addresses.broadcastAddress.hashCode(),
ClusterMetadataService cms = ClusterMetadataService.instance();
cms.log().bootstrap(addresses.broadcastAddress, location.datacenter, cms.logBootstrapCallback());
ClusterMetadata metadata = ClusterMetadata.current().forceInitializedState(addresses.broadcastAddress.hashCode(),
addresses,
NodeVersion.CURRENT,
location));
ClusterMetadataService.instance().commit(initialize,
m -> { logger.info("INITIALIZE_CMS committed successfully"); return m;},
(code, message) -> {
logger.info("INITIALIZE_CMS commit failure: ({}) {}", code, message);
throw new CMSInitializationException();
});
ClusterMetadata initialized = ClusterMetadataService.instance().commit(initialize);
location);
assert ClusterMetadataService.state(metadata) == 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);
ClusterMetadata initialized = ClusterMetadataService.instance().commit(initialize,
m -> { logger.info("INITIALIZE_CMS committed successfully"); return m;},
(code, message) -> {
logger.info("INITIALIZE_CMS commit failure: ({}) {}", code, message);
throw new CMSInitializationException();
});
NodeId id = initialized.myNodeId();
SystemKeyspace.setLocalHostId(id.toUUID());
}

40
src/java/org/apache/cassandra/tcm/transformations/cms/PreInitialize.java
View File

@ -20,6 +20,8 @@ package org.apache.cassandra.tcm.transformations.cms;
import java.io.IOException;
import com.google.common.collect.ImmutableSet;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -29,12 +31,10 @@ import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.DistributedMetadataLogKeyspace;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.Transformation;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.DataPlacements;
import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
@ -72,7 +72,6 @@ public class PreInitialize implements Transformation
assert metadata.epoch.isBefore(Epoch.FIRST);
ClusterMetadata.Transformer transformer = metadata.transformer();
// This null check is a leftover from previous implementations. In earlier versions the address and datacenter
// were not be included in the serialized form of this transform and so were not written to the local or
// distributed logs nor included in the log entries sent over the wire between peers.
@ -84,30 +83,33 @@ public class PreInitialize implements Transformation
// PRE_INITIALIZE_CMS becomes irrelevant.
if (addr != null)
{
DataPlacement.Builder dataPlacementBuilder = DataPlacement.builder();
Replica replica = new Replica(addr,
MetaStrategy.partitioner.getMinimumToken(),
MetaStrategy.partitioner.getMinimumToken(),
true);
dataPlacementBuilder.reads.withReplica(Epoch.FIRST, replica);
dataPlacementBuilder.writes.withReplica(Epoch.FIRST, replica);
DataPlacements initialPlacement = metadata.placements.unbuild()
.with(ReplicationParams.simpleMeta(1, datacenter),
dataPlacementBuilder.build()).build();
transformer.with(initialPlacement);
// create the distributed metadata keyspace in schema with replication settings based on the DC of the
// initial CMS node
Keyspaces updated = metadata.schema.getKeyspaces()
.withAddedOrReplaced(DistributedMetadataLogKeyspace.initialMetadata(datacenter));
transformer.with(new DistributedSchema(updated, Epoch.FIRST));
ClusterMetadata.Transformer.Transformed transformed = transformer.build();
// This is required to bootstrap the initialization process. Because committing to the metadata log uses the
// previous ClusterMetadata to identify CMS members who form the consensus group, the first CMS member must
// be routable at that point. After the INITIALIZE_CMS is committed, and on instances other than the first
// CMS member, this placement is derived from the CMS membership list of node ids.
DataPlacement.Builder initialPlacement = DataPlacement.builder();
Replica replica = new Replica(addr,
MetaStrategy.partitioner.getMinimumToken(),
MetaStrategy.partitioner.getMinimumToken(),
true);
initialPlacement.reads.withReplica(Epoch.FIRST, replica);
initialPlacement.writes.withReplica(Epoch.FIRST, replica);
metadata = transformed.metadata.forcePreInitializedState(initialPlacement.build());
}
else
{
metadata = metadata.forceEpoch(Epoch.FIRST);
}
ClusterMetadata.Transformer.Transformed transformed = transformer.build();
metadata = transformed.metadata.forceEpoch(Epoch.FIRST);
assert metadata.epoch.is(Epoch.FIRST) : metadata.epoch;
return new Success(metadata, LockedRanges.AffectedRanges.EMPTY, transformed.modifiedKeys);
return new Success(metadata, LockedRanges.AffectedRanges.EMPTY, ImmutableSet.of());
}
@Override

8
test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDelayedInitializeTest.java
View File

@ -181,7 +181,7 @@ public class ClusterMetadataUpgradeDelayedInitializeTest extends UpgradeTestBase
try
{
new ByteBuddy().rebase(ClusterMetadata.class)
.method(named("initializeClusterIdentifier"))
.method(named("forceInitializedState"))
.intercept(MethodDelegation.to(ClusterMetadataUpgradeDelayedInitializeTest.BBInterceptor.class))
.make()
.load(classLoader, ClassLoadingStrategy.Default.INJECTION);
@ -201,13 +201,13 @@ public class ClusterMetadataUpgradeDelayedInitializeTest extends UpgradeTestBase
public static class BBInterceptor
{
@SuppressWarnings("unused")
public static ClusterMetadata initializeClusterIdentifier(@SuperCall Callable<ClusterMetadata> zuper)
public static ClusterMetadata forceInitializedState(@SuperCall Callable<ClusterMetadata> zuper)
{
try
{
logger.info("initializeClusterIdentifier waiting...");
logger.info("forceInitializedState waiting...");
BBState.latch.await(60, TimeUnit.SECONDS);
logger.info("initializeClusterIdentifier continuing...");
logger.info("forceInitializedState continuing...");
return zuper.call();
}
catch (Throwable e)