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

[CASSANDRA-20736] Support for upgrades from gossip & from earlier versions with TCM

This commit is contained in:
Sam Tunnicliffe 2025-06-13 16:14:09 +01:00
parent 4ac4b437a3
commit c53e45bddb
3 changed files with 98 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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<CMSMembership>
private final BTreeSet<NodeId> fullMembers;
private final BTreeSet<NodeId> 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<NodeId> full = new TreeSet<>();
SortedSet<NodeId> 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,

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

@ -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<KeyspaceMetadata> 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,

16
src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java
View File

@ -256,9 +256,14 @@ public class DataPlacements extends ReplicationMap<DataPlacement> implements Met
public void serialize(DataPlacements t, DataOutputPlus out, Version version) throws IOException
{
Map<ReplicationParams, DataPlacement> 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<ReplicationParams, DataPlacement> 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<DataPlacement> implements Met
public long serializedSize(DataPlacements t, Version version)
{
long size = sizeof(t.size());
for (Map.Entry<ReplicationParams, DataPlacement> entry : t.asMap().entrySet())
Map<ReplicationParams, DataPlacement> 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<ReplicationParams, DataPlacement> 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);
}