From c53e45bddb8fe269cfe181ab9510cbfafcdeae7e Mon Sep 17 00:00:00 2001 From: Sam Tunnicliffe Date: Fri, 13 Jun 2025 16:14:09 +0100 Subject: [PATCH] [CASSANDRA-20736] Support for upgrades from gossip & from earlier versions with TCM --- .../apache/cassandra/tcm/CMSMembership.java | 36 +++++++++++++ .../apache/cassandra/tcm/ClusterMetadata.java | 51 ++++++++++++++++++- .../tcm/ownership/DataPlacements.java | 16 ++++-- 3 files changed, 98 insertions(+), 5 deletions(-) diff --git a/src/java/org/apache/cassandra/tcm/CMSMembership.java b/src/java/org/apache/cassandra/tcm/CMSMembership.java index d83a5b950e..d18bfdb663 100644 --- a/src/java/org/apache/cassandra/tcm/CMSMembership.java +++ b/src/java/org/apache/cassandra/tcm/CMSMembership.java @@ -28,7 +28,10 @@ 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.tcm.membership.Directory; import org.apache.cassandra.tcm.membership.NodeId; +import org.apache.cassandra.tcm.ownership.DataPlacement; +import org.apache.cassandra.tcm.ownership.VersionedEndpoints; import org.apache.cassandra.tcm.serialization.MetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.utils.btree.BTreeSet; @@ -42,6 +45,39 @@ public class CMSMembership implements MetadataValue private final BTreeSet fullMembers; private final BTreeSet joiningMembers; + /** + * Used to derive a CMSMembership when deserializing a ClusterMetadata instance written with a metadata version + * prior to V7. At that time, CMS membership was always inferred from the data placements of the distributed + * cluster metadata keyspace. Read replicas are full members of the CMS and write-only replicas are in the process + * of joining. Note: every read replica must also be a write replica, leaving the CMS is atomic in respect of the + * placements. + * @param placement + * @param directory + * @return + */ + public static CMSMembership reconstruct(DataPlacement placement, Directory directory) + { + SortedSet full = new TreeSet<>(); + SortedSet joining = new TreeSet<>(); + Epoch lm = Epoch.EMPTY; + for (VersionedEndpoints.ForRange endpoints : placement.reads.endpoints) + { + lm = endpoints.lastModified().isAfter(lm) ? endpoints.lastModified() : lm; + endpoints.get().endpoints().forEach(e -> full.add(directory.peerId(e))); + } + + for (VersionedEndpoints.ForRange endpoints : placement.writes.endpoints) + { + lm = endpoints.lastModified().isAfter(lm) ? endpoints.lastModified() : lm; + endpoints.get().endpoints().forEach(e -> { + NodeId id = directory.peerId(e); + if (!full.contains(id)) + joining.add(id); + }); + } + return new CMSMembership(lm, BTreeSet.of(full), BTreeSet.of(joining)); + } + 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 749523dc9b..619283eb0f 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -28,6 +28,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -213,7 +214,7 @@ public class ClusterMetadata this.directory = directory; this.tokenMap = tokenMap; this.accordFastPath = accordFastPath; - this.placements = addMetaPlacement(placements, cmsMembership); + this.placements = maybeAddMetaPlacement(placements, cmsMembership); this.lockedRanges = lockedRanges; this.inProgressSequences = inProgressSequences; this.consensusMigrationState = consensusMigrationState; @@ -263,7 +264,7 @@ public class ClusterMetadata return fullCMSReplicas; } - private DataPlacements addMetaPlacement(DataPlacements placements, CMSMembership cms) + private DataPlacements maybeAddMetaPlacement(DataPlacements placements, CMSMembership cms) { if (epoch.isBefore(Epoch.FIRST) || schema.getKeyspaces().get(SchemaConstants.METADATA_KEYSPACE_NAME).isEmpty()) return placements; @@ -271,9 +272,24 @@ public class ClusterMetadata DataPlacement.Builder metaBuilder = DataPlacement.builder(); 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())); } + else if (epoch.isAfter(Epoch.FIRST) && directory.isEmpty()) + { + // This cluster did not previously upgrade from a gossip based version (i.e. pre-6.0) but did at some point + // run a version prior to MetadataVersion.V7 where we started to encode CMS membership directly. This + // condition implies that we are reconstructing a serialized cluster metadata during replay or else the + // directory should not be empty after Epoch.FIRST as the base state in INITIALIZE_CMS now includes the + // first CMS node. Similarly, if the cluster had previously been running a gossip-based version, the + // directory would contain entries for each of the live nodes at the time of upgrade. + // Given this state, the very next transformation that is/was applied will be to register the node that + // committed the PRE_INITIALIZE_CMS and INTIALIZE_CMS transformations. So we just leave the placements + // untouched as they will already contain that node as an endpoint. + return placements; + } else { for (NodeId id : cms.fullMembers()) @@ -1282,6 +1298,37 @@ public class ClusterMetadata CMSMembership cmsMembership = CMSMembership.EMPTY; if (version.isAtLeast(Version.V9)) cmsMembership = CMSMembership.serializer.deserialize(in, version); + else + { + Optional metadataKs = schema.maybeGetKeyspaceMetadata(SchemaConstants.METADATA_KEYSPACE_NAME); + if (metadataKs.isPresent()) + { + // Pre-V9 the membership of the CMS was always inferred from the placement of the distributed + // metadata keyspace. + // If the directory is not empty the endpoints in the placement must belong to registered nodes, + // so we can derive the CMSMembership using the data placement and directory. + + // If the directory is empty, then the cluster metadata must be the payload of an INITIALIZE_CMS + // transformation of a cluster that began on a post-6.0, pre-MetadataVersion.V9 version. + // In this case, we can and must assume that the initial CMS membership will consist of a single + // node, with the node_id 1. + // Note: the only route to arrive at this scenario is if a cluster is initialized on a post-6.0, + // pre-V9 version and then upgraded to a post-V9 version without any metadata snapshots being taken. + // If there is a snapshot available locally, when the upgraded node starts up it will replay its + // local log from that point. The INITIALIZE_CMS transform will not be replayed. + if (!dir.isEmpty()) + { + DataPlacement placement = placements.get(metadataKs.get().params.replication); + cmsMembership = CMSMembership.reconstruct(placement, dir); + } + else + { + NodeId id = new NodeId(1); + cmsMembership = CMSMembership.EMPTY.startJoining(id).finishJoining(id); + } + placements = placements.unbuild().without(metadataKs.get().params.replication).build(); + } + } return new ClusterMetadata(clusterIdentifier, epoch, diff --git a/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java b/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java index d813e8c287..bb50523bda 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java +++ b/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java @@ -256,9 +256,14 @@ public class DataPlacements extends ReplicationMap implements Met public void serialize(DataPlacements t, DataOutputPlus out, Version version) throws IOException { Map map = t.asMap(); - out.writeInt(map.size()); + + // From V7, placements for the metadata keyspace are derived from CMSMembership, not serialized + int mapSize = version.isBefore(Version.V7) ? map.size() : Math.max(map.size() - 1, 0); + out.writeInt(mapSize); for (Map.Entry entry : map.entrySet()) { + if (version.isAtLeast(Version.V7) && entry.getKey().isMeta()) + continue; ReplicationParams.serializer.serialize(entry.getKey(), out, version); DataPlacement.serializerFor(entry.getKey()).serialize(entry.getValue(), out, version); } @@ -280,9 +285,14 @@ public class DataPlacements extends ReplicationMap implements Met public long serializedSize(DataPlacements t, Version version) { - long size = sizeof(t.size()); - for (Map.Entry entry : t.asMap().entrySet()) + Map map = t.asMap(); + // From V7, placements for the metadata keyspace are derived from CMSMembership, not serialized + int mapSize = version.isBefore(Version.V7) ? map.size() : Math.max(map.size() - 1, 0); + long size = sizeof(mapSize); + for (Map.Entry entry : map.entrySet()) { + if (version.isAtLeast(Version.V7) && entry.getKey().isMeta()) + continue; size += ReplicationParams.serializer.serializedSize(entry.getKey(), version); size += DataPlacement.serializerFor(entry.getKey()).serializedSize(entry.getValue(), version); }