diff --git a/src/java/org/apache/cassandra/locator/CMSPlacementStrategy.java b/src/java/org/apache/cassandra/locator/CMSPlacementStrategy.java index e00289aef6..cb4dfa82c6 100644 --- a/src/java/org/apache/cassandra/locator/CMSPlacementStrategy.java +++ b/src/java/org/apache/cassandra/locator/CMSPlacementStrategy.java @@ -90,7 +90,8 @@ public class CMSPlacementStrategy // Although MetaStrategy has its own entireRange, it uses a custom partitioner which isn't compatible with // regular, non-CMS placements. For that reason, we select replicas here using tokens provided by the - // globally configured partitioner. + // globally configured partitioner. This also has the benefit of making concurrent operations, such as + // bounces/upgrades/etc, safe for the CMS if they are replica aware. Token minToken = DatabaseDescriptor.getPartitioner().getMinimumToken(); EndpointsForRange endpoints = NetworkTopologyStrategy.calculateNaturalReplicas(minToken, new Range<>(minToken, minToken), diff --git a/src/java/org/apache/cassandra/tcm/transformations/cms/AdvanceCMSReconfiguration.java b/src/java/org/apache/cassandra/tcm/transformations/cms/AdvanceCMSReconfiguration.java index 614dd8a4ed..f31a3fe503 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/cms/AdvanceCMSReconfiguration.java +++ b/src/java/org/apache/cassandra/tcm/transformations/cms/AdvanceCMSReconfiguration.java @@ -29,15 +29,12 @@ 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.RangesByEndpoint; -import org.apache.cassandra.locator.Replica; -import org.apache.cassandra.schema.ReplicationParams; +import org.apache.cassandra.tcm.CMSMembership; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.MultiStepOperation; import org.apache.cassandra.tcm.Transformation; import org.apache.cassandra.tcm.membership.NodeId; -import org.apache.cassandra.tcm.ownership.DataPlacement; import org.apache.cassandra.tcm.sequences.InProgressSequences; import org.apache.cassandra.tcm.sequences.LockedRanges; import org.apache.cassandra.tcm.sequences.ReconfigureCMS; @@ -45,7 +42,6 @@ import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; import static org.apache.cassandra.exceptions.ExceptionCode.INVALID; -import static org.apache.cassandra.locator.MetaStrategy.entireRange; import static org.apache.cassandra.tcm.MultiStepOperation.Kind.RECONFIGURE_CMS; /** @@ -154,38 +150,26 @@ public class AdvanceCMSReconfiguration implements Transformation { // Pop the next node to be added from the list diff.additions NodeId addition = diff.additions.get(0); - InetAddressAndPort endpoint = prev.directory.endpoint(addition); - Replica replica = new Replica(endpoint, entireRange, true); - List newAdditions = new ArrayList<>(diff.additions.subList(1, diff.additions.size())); // Check that the candidate is not already a CMS member - ReplicationParams metaParams = ReplicationParams.meta(prev); - RangesByEndpoint readReplicas = prev.placements.get(metaParams).reads.byEndpoint(); - RangesByEndpoint writeReplicas = prev.placements.get(metaParams).writes.byEndpoint(); - if (readReplicas.containsKey(endpoint) || writeReplicas.containsKey(endpoint)) - return new Transformation.Rejected(INVALID, "Endpoint is already a member of CMS"); + CMSMembership cms = prev.cmsMembership; + if (cms.joiningMembers().contains(addition) || cms.fullMembers().contains(addition)) + return new Transformation.Rejected(INVALID, "Endpoint is already a full or joining member of the CMS: " + prev.directory.endpoint(addition)); - - ClusterMetadata.Transformer transformer = prev.transformer(); - // Add the candidate as a write replica - DataPlacement.Builder builder = prev.placements.get(metaParams).unbuild() - .withWriteReplica(prev.nextEpoch(), replica); - transformer.with(prev.placements.unbuild().with(metaParams, builder.build()).build()); + // Add the candidate as a joining member + ClusterMetadata.Transformer transformer = prev.transformer().startJoiningCMS(addition); // Construct a set of sources for the new member to stream log tables from (essentially this is the existing members) - Set streamCandidates = new HashSet<>(); - for (Replica r : prev.placements.get(metaParams).reads.byEndpoint().flattenValues()) - { - if (!replica.equals(r)) - streamCandidates.add(r.endpoint()); - } + Set streamCandidates = prev.fullCMSMembers(); // Set up the next step in the sequence. This encapsulates the entire state of the reconfiguration sequence, // including the remaining add/remove operations and the streaming that needs to be done by the joining node + List newAdditions = new ArrayList<>(diff.additions.subList(1, diff.additions.size())); AdvanceCMSReconfiguration next = next(prev.nextEpoch(), newAdditions, diff.removals, new ReconfigureCMS.ActiveTransition(addition, streamCandidates)); + // Create a new sequence instance with the next step to reflect that the state has progressed. ReconfigureCMS advanced = sequence.advance(next); // Finally, replace the existing reconfiguration sequence with this updated one. @@ -206,21 +190,21 @@ public class AdvanceCMSReconfiguration implements Transformation */ private Transformation.Result finishAdd(ClusterMetadata prev, ReconfigureCMS sequence, NodeId addition) { + // Check that the candidate is already a joining CMS member + CMSMembership cms = prev.cmsMembership; + if (!cms.joiningMembers().contains(addition)) + return new Transformation.Rejected(INVALID, "Endpoint is not a in the process of joining the CMS: " + prev.directory.endpoint(addition)); + // Add the new member as a full read replica, able to participate in quorums for log updates - ReplicationParams metaParams = ReplicationParams.meta(prev); - InetAddressAndPort endpoint = prev.directory.endpoint(addition); - Replica replica = new Replica(endpoint, entireRange, true); - ClusterMetadata.Transformer transformer = prev.transformer(); - DataPlacement.Builder builder = prev.placements.get(metaParams) - .unbuild() - .withReadReplica(prev.nextEpoch(), replica); - transformer = transformer.with(prev.placements.unbuild().with(metaParams, builder.build()).build()); + ClusterMetadata.Transformer transformer = prev.transformer().finishJoiningCMS(addition); // Set up the next step in the sequence. This encapsulates the entire state of the reconfiguration sequence, // which includes the remaining add/remove operations AdvanceCMSReconfiguration next = next(prev.nextEpoch(), diff.additions, diff.removals, null); + // Create a new sequence instance with the next step to reflect that the state has progressed. ReconfigureCMS advanced = sequence.advance(next); + // Finally, replace the existing reconfiguration sequence with this updated one. transformer.with(prev.inProgressSequences.with(ReconfigureCMS.SequenceKey.instance, (ReconfigureCMS old) -> advanced)); return Transformation.success(transformer, MetaStrategy.affectedRanges(prev)); @@ -238,29 +222,25 @@ public class AdvanceCMSReconfiguration implements Transformation List newRemovals = new ArrayList<>(diff.removals.subList(1, diff.removals.size())); // Check that the candidate is actually a CMS member - ClusterMetadata.Transformer transformer = prev.transformer(); + Set cms = prev.fullCMSMemberIds(); InetAddressAndPort endpoint = prev.directory.endpoint(removal); - Replica replica = new Replica(endpoint, entireRange, true); - ReplicationParams metaParams = ReplicationParams.meta(prev); - if (!prev.fullCMSMembers().contains(endpoint)) + if (!prev.fullCMSMemberIds().contains(removal)) return new Transformation.Rejected(INVALID, String.format("%s is not currently a CMS member, cannot remove it", endpoint)); // Check that the candidate is not the only CMS member - DataPlacement.Builder builder = prev.placements.get(metaParams).unbuild(); - builder.reads.withoutReplica(prev.nextEpoch(), replica); - builder.writes.withoutReplica(prev.nextEpoch(), replica); - DataPlacement proposed = builder.build(); - if (proposed.reads.byEndpoint().isEmpty() || proposed.writes.byEndpoint().isEmpty()) + if (cms.size() == 1) return new Transformation.Rejected(INVALID, String.format("Removing %s will leave no nodes in CMS", endpoint)); - // Actually remove the candidate - transformer = transformer.with(prev.placements.unbuild().with(metaParams, proposed).build()); + // Remove the CMS member + ClusterMetadata.Transformer transformer = prev.transformer().leaveCMS(removal); // Set up the next step in the sequence. This encapsulates the entire state of the reconfiguration sequence, // which includes the remaining add/remove operations AdvanceCMSReconfiguration next = next(prev.nextEpoch(), diff.additions, newRemovals, null); + // Create a new sequence instance with the next step to reflect that the state has progressed. ReconfigureCMS advanced = sequence.advance(next); + // Finally, replace the existing reconfiguration sequence with this updated one. transformer.with(prev.inProgressSequences.with(ReconfigureCMS.SequenceKey.instance, (ReconfigureCMS old) -> advanced)); return Transformation.success(transformer, MetaStrategy.affectedRanges(prev)); diff --git a/src/java/org/apache/cassandra/tcm/transformations/cms/PrepareCMSReconfiguration.java b/src/java/org/apache/cassandra/tcm/transformations/cms/PrepareCMSReconfiguration.java index cfb7db66dd..5a8bc70486 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/cms/PrepareCMSReconfiguration.java +++ b/src/java/org/apache/cassandra/tcm/transformations/cms/PrepareCMSReconfiguration.java @@ -29,7 +29,6 @@ import java.util.Map; import java.util.Set; import java.util.function.Function; import java.util.function.Predicate; -import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -83,7 +82,7 @@ public abstract class PrepareCMSReconfiguration implements Transformation logger.info("Proposed CMS reconfiguration resulted in {}", diff); LockedRanges.Key lockKey = LockedRanges.keyFor(prev.nextEpoch()); - Set cms = prev.fullCMSMembers().stream().map(prev.directory::peerId).collect(Collectors.toSet()); + Set cms = prev.fullCMSMemberIds(); Set tmp = new HashSet<>(cms); tmp.addAll(diff.additions); tmp.removeAll(diff.removals); @@ -355,10 +354,7 @@ public abstract class PrepareCMSReconfiguration implements Transformation public static boolean needsReconfiguration(ClusterMetadata metadata) { Map dcRf = extractRf(ReplicationParams.meta(metadata)); - Set currentCms = metadata.fullCMSMembers() - .stream() - .map(metadata.directory::peerId) - .collect(Collectors.toSet()); + Set currentCms = metadata.fullCMSMemberIds(); int expectedSize = dcRf.values().stream().mapToInt(Integer::intValue).sum(); if (currentCms.size() != expectedSize) return true;