From b3396b5f9e066dd3cc1552301a03d0bf6c197b4e Mon Sep 17 00:00:00 2001 From: Sam Tunnicliffe Date: Fri, 13 Jun 2025 18:41:46 +0100 Subject: [PATCH] [CASSANDRA-20736] Update CancelCMSReconfiguration --- .../apache/cassandra/tcm/CMSMembership.java | 18 +++++ .../apache/cassandra/tcm/ClusterMetadata.java | 24 +++---- .../sequences/CancelCMSReconfiguration.java | 68 ++++++++----------- 3 files changed, 55 insertions(+), 55 deletions(-) diff --git a/src/java/org/apache/cassandra/tcm/CMSMembership.java b/src/java/org/apache/cassandra/tcm/CMSMembership.java index d18bfdb663..5b88543be8 100644 --- a/src/java/org/apache/cassandra/tcm/CMSMembership.java +++ b/src/java/org/apache/cassandra/tcm/CMSMembership.java @@ -28,6 +28,8 @@ 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; +import org.apache.cassandra.locator.MetaStrategy; +import org.apache.cassandra.locator.Replica; import org.apache.cassandra.tcm.membership.Directory; import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.ownership.DataPlacement; @@ -78,6 +80,22 @@ public class CMSMembership implements MetadataValue return new CMSMembership(lm, BTreeSet.of(full), BTreeSet.of(joining)); } + public DataPlacement toPlacement(Directory directory) + { + DataPlacement.Builder builder = DataPlacement.builder(); + for (NodeId id : fullMembers) + { + Replica replica = MetaStrategy.replica(directory.endpoint(id)); + builder.withReadReplica(lastModified, replica); + builder.withWriteReplica(lastModified, replica); + } + for(NodeId id : joiningMembers) + { + builder.withWriteReplica(lastModified, MetaStrategy.replica(directory.endpoint(id))); + } + return builder.build(); + } + private CMSMembership() { this(Epoch.EMPTY, diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java index 619283eb0f..4bdaddf3f7 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -269,13 +269,16 @@ public class ClusterMetadata if (epoch.isBefore(Epoch.FIRST) || schema.getKeyspaces().get(SchemaConstants.METADATA_KEYSPACE_NAME).isEmpty()) return placements; - DataPlacement.Builder metaBuilder = DataPlacement.builder(); + DataPlacement metaPlacement; if (epoch.is(Epoch.FIRST)) { // PRE_INITIALIZE_CMS: placements need to be hardcoded to the local address so that the subsequent // INITIALIZE_CMS can be committed - metaBuilder.withReadReplica(Epoch.FIRST, MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort())); - metaBuilder.withWriteReplica(Epoch.FIRST, MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort())); + Replica localReplica = MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort()); + metaPlacement = DataPlacement.builder() + .withReadReplica(Epoch.FIRST, localReplica) + .withWriteReplica(Epoch.FIRST, localReplica) + .build(); } else if (epoch.isAfter(Epoch.FIRST) && directory.isEmpty()) { @@ -292,20 +295,11 @@ public class ClusterMetadata } else { - for (NodeId id : cms.fullMembers()) - { - Replica replica = MetaStrategy.replica(directory.endpoint(id)); - metaBuilder.withReadReplica(cms.lastModified(), replica); - metaBuilder.withWriteReplica(cms.lastModified(), replica); - } - - for(NodeId id : cms.joiningMembers()) - { - metaBuilder.withWriteReplica(cms.lastModified(), MetaStrategy.replica(directory.endpoint(id))); - } + // Build a placement based on the CMS membership + metaPlacement = cms.toPlacement(directory); } return placements.unbuild() - .with(ReplicationParams.meta(this), metaBuilder.build()) + .with(ReplicationParams.meta(this), metaPlacement) .build(); } diff --git a/src/java/org/apache/cassandra/tcm/sequences/CancelCMSReconfiguration.java b/src/java/org/apache/cassandra/tcm/sequences/CancelCMSReconfiguration.java index 4139110236..3dbd9311bd 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/CancelCMSReconfiguration.java +++ b/src/java/org/apache/cassandra/tcm/sequences/CancelCMSReconfiguration.java @@ -25,25 +25,20 @@ import java.util.Map; import org.apache.cassandra.exceptions.ExceptionCode; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.MetaStrategy; -import org.apache.cassandra.locator.Replica; import org.apache.cassandra.schema.DistributedSchema; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.accord.topology.FastPathStrategy; +import org.apache.cassandra.tcm.CMSMembership; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Transformation; import org.apache.cassandra.tcm.membership.Directory; -import org.apache.cassandra.tcm.ownership.DataPlacement; -import org.apache.cassandra.tcm.ownership.DataPlacements; import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; -import static org.apache.cassandra.locator.MetaStrategy.entireRange; - public class CancelCMSReconfiguration implements Transformation { public static final Serializer serializer = new Serializer(); @@ -66,53 +61,46 @@ public class CancelCMSReconfiguration implements Transformation if (reconfigureCMS == null) return new Rejected(ExceptionCode.INVALID, "Can not cancel reconfiguration since there does not seem to be any in-flight"); - ReplicationParams metaParams = ReplicationParams.meta(prev); ClusterMetadata.Transformer transformer = prev.transformer(); - DataPlacement placement = prev.placements.get(metaParams); // Reset any partially completed transition by removing the pending replica from the write group if (reconfigureCMS.next.activeTransition != null) { - InetAddressAndPort pendingEndpoint = prev.directory.endpoint(reconfigureCMS.next.activeTransition.nodeId); - Replica pendingReplica = new Replica(pendingEndpoint, entireRange, true); - placement = placement.unbuild() - .withoutWriteReplica(prev.nextEpoch(), pendingReplica) - .build(); + // see what the placements for the meta keyspace will be after cancelling the active join + CMSMembership cms = prev.cmsMembership.cancelJoining(reconfigureCMS.next.activeTransition.nodeId); + if (!cms.joiningMembers().isEmpty()) + return new Rejected(ExceptionCode.INVALID, + String.format("Placements will be inconsistent if this transformation is applied:" + + "\nFull members %s\nJoining members: %s", + cms.fullMembers(), + cms.joiningMembers())); + + // if all is good, actually cancel the joining member + transformer = transformer.cancelJoiningCMS(reconfigureCMS.next.activeTransition.nodeId); + // Recalculate the replication params for the meta keyspace based on the actual placement as it will be + ReplicationParams recalculated = getAccurateReplication(prev.directory, cms); + + // If they no longer match the replication params in schema, i.e. the transitions completed so far did not + // bring the membership/placements into line with configuration, update schema to match what we actually have + if (!recalculated.equals(ReplicationParams.meta(prev))) + { + KeyspaceMetadata keyspace = prev.schema.getKeyspaceMetadata(SchemaConstants.METADATA_KEYSPACE_NAME); + KeyspaceMetadata newKeyspace = keyspace.withSwapped(new KeyspaceParams(keyspace.params.durableWrites, + recalculated, + FastPathStrategy.simple())); + transformer = transformer.with(new DistributedSchema(prev.schema.getKeyspaces().withAddedOrUpdated(newKeyspace))); + } } - if (!placement.reads.equivalentTo(placement.writes)) - return new Rejected(ExceptionCode.INVALID, String.format("Placements will be inconsistent if this transformation is applied:\nReads %s\nWrites: %s", - placement.reads, - placement.writes)); - - // Reset the replication params for the meta keyspace based on the actual placement in case they no longer match - ReplicationParams fromPlacement = getAccurateReplication(prev.directory, placement); - - // If they no longer match, i.e. the transitions completed so far did not bring the placements into line with - // the configuration, remove the entry keyed by the existing configured params. - DataPlacements.Builder builder = prev.placements.unbuild(); - if (!metaParams.equals(fromPlacement)) - { - builder = builder.without(metaParams); - - // Also update schema with the corrected params - KeyspaceMetadata keyspace = prev.schema.getKeyspaceMetadata(SchemaConstants.METADATA_KEYSPACE_NAME); - KeyspaceMetadata newKeyspace = keyspace.withSwapped(new KeyspaceParams(keyspace.params.durableWrites, fromPlacement, FastPathStrategy.simple())); - transformer = transformer.with(new DistributedSchema(prev.schema.getKeyspaces().withAddedOrUpdated(newKeyspace))); - } - - // finally, add the possibly corrected placement keyed by the possibly corrected params - builder = builder.with(fromPlacement, placement); - transformer = transformer.with(builder.build()); return Transformation.success(transformer.with(prev.inProgressSequences.without(ReconfigureCMS.SequenceKey.instance)) .with(prev.lockedRanges.unlock(reconfigureCMS.next.lockKey)), MetaStrategy.affectedRanges(prev)); } - private ReplicationParams getAccurateReplication(Directory directory, DataPlacement placement) + private ReplicationParams getAccurateReplication(Directory directory, CMSMembership membership) { Map replicasPerDc = new HashMap<>(); - placement.writes.byEndpoint().keySet().forEach(i -> { - String dc = directory.location(directory.peerId(i)).datacenter; + membership.fullMembers().forEach(id -> { + String dc = directory.location(id).datacenter; int count = replicasPerDc.getOrDefault(dc, 0); replicasPerDc.put(dc, ++count); });