diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java index 423d2c82af..34849584da 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -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)); diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java index c1eb9ac847..2c15bc5e9b 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadataService.java @@ -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; } diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java index 4c0b625d38..22bab09a09 100644 --- a/src/java/org/apache/cassandra/tcm/Startup.java +++ b/src/java/org/apache/cassandra/tcm/Startup.java @@ -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()); } diff --git a/src/java/org/apache/cassandra/tcm/transformations/cms/PreInitialize.java b/src/java/org/apache/cassandra/tcm/transformations/cms/PreInitialize.java index 7286743f91..2725fffebf 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/cms/PreInitialize.java +++ b/src/java/org/apache/cassandra/tcm/transformations/cms/PreInitialize.java @@ -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 diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDelayedInitializeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDelayedInitializeTest.java index e1fffc74b7..b406ae914c 100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDelayedInitializeTest.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeDelayedInitializeTest.java @@ -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 zuper) + public static ClusterMetadata forceInitializedState(@SuperCall Callable 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)