diff --git a/src/java/org/apache/cassandra/locator/EndpointsForToken.java b/src/java/org/apache/cassandra/locator/EndpointsForToken.java index fc88fa761b..f0a3f732ba 100644 --- a/src/java/org/apache/cassandra/locator/EndpointsForToken.java +++ b/src/java/org/apache/cassandra/locator/EndpointsForToken.java @@ -25,7 +25,9 @@ import com.google.common.base.Preconditions; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ownership.DataPlacement; import org.apache.cassandra.tcm.ownership.VersionedEndpoints; /** @@ -164,7 +166,11 @@ public class EndpointsForToken extends Endpoints public static VersionedEndpoints.ForToken natural(Keyspace keyspace, Token token) { - return ClusterMetadata.current().placements.get(keyspace.getMetadata().params.replication).reads.forToken(token); + ReplicationParams replication = keyspace.getMetadata().params.replication; + DataPlacement placement = replication.isMeta() + ? ClusterMetadata.current().getCMSPlacement() + : ClusterMetadata.current().placements.get(replication); + return placement.reads.forToken(token); } } diff --git a/src/java/org/apache/cassandra/locator/ReplicaLayout.java b/src/java/org/apache/cassandra/locator/ReplicaLayout.java index 440b2bed73..8ea75c5577 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaLayout.java +++ b/src/java/org/apache/cassandra/locator/ReplicaLayout.java @@ -239,7 +239,9 @@ public abstract class ReplicaLayout> { // todo deduplicate so that "pending" contains "read - write", // which is a hack until we revisit how consistency level handles pending - DataPlacement dataPlacement = metadata.placements.get(ks.params.replication); + DataPlacement dataPlacement = ks.params.replication.isMeta() + ? metadata.getCMSPlacement() + : metadata.placements.get(ks.params.replication); natural = forNonLocalStrategyTokenRead(dataPlacement, token); // perf optimization to avoid double endpoints search and filtering for a typical case // DataPlacement constructor does a deduplication of reads/writes, so we can use cheap == comparision here @@ -394,7 +396,10 @@ public abstract class ReplicaLayout> static EndpointsForRange forNonLocalStategyRangeRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, AbstractBounds range) { - return metadata.placements.get(keyspace.params.replication).reads.forRange(range.right.getToken()).get(); + DataPlacement placement = keyspace.params.replication.isMeta() + ? metadata.getCMSPlacement() + : metadata.placements.get(keyspace.params.replication); + return placement.reads.forRange(range.right.getToken()).get(); } public static EndpointsForToken forNonLocalStrategyTokenRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, Token token) @@ -407,17 +412,11 @@ public abstract class ReplicaLayout> return dataPlacement.reads.forToken(token).get(); } - static EndpointsForToken forNonLocalStrategyTokenWrite(ClusterMetadata metadata, KeyspaceMetadata keyspace, Token token) - { - return forNonLocalStrategyTokenWrite(metadata.placements.get(keyspace.params.replication), token); - } - - static EndpointsForToken forNonLocalStrategyTokenWrite(DataPlacement dataPlacement, Token token) + private static EndpointsForToken forNonLocalStrategyTokenWrite(DataPlacement dataPlacement, Token token) { return dataPlacement.writes.forToken(token).get(); } - static EndpointsForRange forLocalStrategyRange(ClusterMetadata metadata, AbstractReplicationStrategy replicationStrategy, AbstractBounds range) { return replicationStrategy.calculateNaturalReplicas(range.right.getToken(), metadata); diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java index 36bf1c0f14..1815d5c7d4 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -126,6 +126,7 @@ public class ClusterMetadata private Set fullCMSEndpoints; private volatile Map localRangesAllSettled = null; private static final RangesAtEndpoint EMPTY_LOCAL_RANGES = RangesAtEndpoint.empty(FBUtilities.getBroadcastAddressAndPort()); + private DataPlacement cmsDataPlacement; public ClusterMetadata(IPartitioner partitioner) { @@ -214,7 +215,7 @@ public class ClusterMetadata this.directory = directory; this.tokenMap = tokenMap; this.accordFastPath = accordFastPath; - this.placements = maybeAddMetaPlacement(placements, cmsMembership); + this.placements = placements; this.lockedRanges = lockedRanges; this.inProgressSequences = inProgressSequences; this.consensusMigrationState = consensusMigrationState; @@ -222,6 +223,7 @@ public class ClusterMetadata this.locator = Locator.usingDirectory(directory); this.accordStaleReplicas = accordStaleReplicas; this.cmsMembership = cmsMembership; + this.cmsDataPlacement = calculateCMSPlacement(placements, cmsMembership); } public Set fullCMSMemberIds() @@ -264,52 +266,57 @@ public class ClusterMetadata return fullCMSReplicas; } - private DataPlacements maybeAddMetaPlacement(DataPlacements placements, CMSMembership cms) + public DataPlacement getCMSPlacement() + { + return cmsDataPlacement; + } + + private DataPlacement calculateCMSPlacement(DataPlacements placements, CMSMembership cms) { if (epoch.isBefore(Epoch.FIRST) || schema.getKeyspaces().get(SchemaConstants.METADATA_KEYSPACE_NAME).isEmpty()) - return placements; + return DataPlacement.empty(); - DataPlacement metaPlacement; - Epoch previousLastModified = placements.lastModified(); - Epoch nextLastModified; - if (epoch.is(Epoch.FIRST)) + if (directory.isEmpty()) { - // PRE_INITIALIZE_CMS: placements need to be hardcoded to the local address so that the subsequent - // INITIALIZE_CMS can be committed - Replica localReplica = MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort()); - metaPlacement = DataPlacement.builder() - .withReadReplica(Epoch.FIRST, localReplica) - .withWriteReplica(Epoch.FIRST, localReplica) - .build(); - nextLastModified = Epoch.FIRST; - } - 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; + 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 + Replica localReplica = MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort()); + return DataPlacement.builder() + .withReadReplica(Epoch.FIRST, localReplica) + .withWriteReplica(Epoch.FIRST, localReplica) + .build(); + } + else + { + // 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 extract the placements + // associated with the MetaStrategy params as they will already contain that node as an endpoint. + for (ReplicationParams params : placements.keys()) + if (params.isMeta()) + return placements.get(params); + + // This point can only be reached in tests. + // TODO enforce that invariant somehow + Replica localReplica = MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort()); + return DataPlacement.builder() + .withReadReplica(epoch, localReplica) + .withWriteReplica(epoch, localReplica) + .build(); + } } else { // Build a placement based on the CMS membership - metaPlacement = cms.toPlacement(directory); - if (cms.lastModified().isAfter(previousLastModified)) - nextLastModified = cms.lastModified(); - else - nextLastModified = previousLastModified; - + return cms.toPlacement(directory); } - return placements.unbuild() - .with(ReplicationParams.meta(this), metaPlacement) - .build() - .withLastModified(nextLastModified); } public Transformer transformer() @@ -1220,7 +1227,10 @@ public class ClusterMetadata DistributedSchema.serializer.serialize(metadata.schema, out, version); Directory.serializer.serialize(metadata.directory, out, version); TokenMap.serializer.serialize(metadata.tokenMap, out, version); - DataPlacements.serializer.serialize(metadata.placements, out, version); + // Prior to V9, placements for the MetaStrategy keyspace were included in the main DataPlacements + // so when targetting such a version, emulate that. + DataPlacements placements = version.isBefore(Version.V9) ? preV9Placements(metadata) : metadata.placements; + DataPlacements.serializer.serialize(placements, out, version); if (version.isAtLeast(MIN_ACCORD_VERSION)) { AccordFastPath.serializer.serialize(metadata.accordFastPath, out, version); @@ -1239,6 +1249,7 @@ public class ClusterMetadata assert key.valueType.isInstance(value); value.serialize(out, version); } + // From V9 CMS membership is directly encoded in ClusterMetadata if (version.isAtLeast(Version.V9)) CMSMembership.serializer.serialize(metadata.cmsMembership, out, version); } @@ -1381,8 +1392,7 @@ public class ClusterMetadata sizeof(metadata.partitioner.getClass().getCanonicalName()) + DistributedSchema.serializer.serializedSize(metadata.schema, version) + Directory.serializer.serializedSize(metadata.directory, version) + - TokenMap.serializer.serializedSize(metadata.tokenMap, version) + - DataPlacements.serializer.serializedSize(metadata.placements, version); + TokenMap.serializer.serializedSize(metadata.tokenMap, version); if (version.isAtLeast(MIN_ACCORD_VERSION)) { @@ -1394,12 +1404,27 @@ public class ClusterMetadata size += LockedRanges.serializer.serializedSize(metadata.lockedRanges, version) + InProgressSequences.serializer.serializedSize(metadata.inProgressSequences, version); + // Prior to V9, placements for the MetaStrategy keyspace were included in the main DataPlacements + // so when targetting such a version, emulate that. + DataPlacements placements = version.isBefore(Version.V9) ? preV9Placements(metadata) : metadata.placements; + size += DataPlacements.serializer.serializedSize(placements, version); + // From V9 CMS membership is directly encoded in ClusterMetadata if (version.isAtLeast(Version.V9)) size += CMSMembership.serializer.serializedSize(metadata.cmsMembership, version); return size; } + private DataPlacements preV9Placements(ClusterMetadata metadata) + { + if (metadata.cmsDataPlacement.isEmpty()) + return metadata.placements; + + return metadata.placements.unbuild() + .with(ReplicationParams.meta(metadata), metadata.cmsDataPlacement) + .build(); + } + public static IPartitioner getPartitioner(DataInputPlus in, Version version) throws IOException { if (version.isAtLeast(Version.V1)) diff --git a/src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java b/src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java index a78fc8be41..e3a4ae3efd 100644 --- a/src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java +++ b/src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java @@ -70,7 +70,7 @@ public class PaxosBackedProcessor extends AbstractLocalProcessor if (metadata.epoch.isAfter(Epoch.FIRST) && metadata.fullCMSMembers().contains(FBUtilities.getBroadcastAddressAndPort())) return true; return metadata.epoch.isEqualOrBefore(Epoch.FIRST) - && metadata.placements.get(ReplicationParams.meta(metadata)).reads.byEndpoint().keySet().contains(FBUtilities.getBroadcastAddressAndPort()); + && metadata.getCMSPlacement().reads.byEndpoint().keySet().contains(FBUtilities.getBroadcastAddressAndPort()); } @Override diff --git a/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java b/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java index fd1a2618db..6c0ff2012c 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java +++ b/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java @@ -106,6 +106,11 @@ public class DataPlacement return EMPTY; } + public boolean isEmpty() + { + return reads.isEmpty() && writes.isEmpty(); + } + public static Builder builder() { return new Builder(ReplicaGroups.builder(), diff --git a/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java b/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java index bb50523bda..fd992da478 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java +++ b/src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java @@ -46,6 +46,7 @@ import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.db.TypeSizes.sizeof; +import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt; public class DataPlacements extends ReplicationMap implements MetadataValue { @@ -256,14 +257,13 @@ public class DataPlacements extends ReplicationMap implements Met public void serialize(DataPlacements t, DataOutputPlus out, Version version) throws IOException { Map map = t.asMap(); + if (version.isBefore(Version.V9)) + out.writeInt(map.size()); + else + out.writeUnsignedVInt32(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); } @@ -272,7 +272,7 @@ public class DataPlacements extends ReplicationMap implements Met public DataPlacements deserialize(DataInputPlus in, Version version) throws IOException { - int size = in.readInt(); + int size = version.isBefore(Version.V9) ? in.readInt() : in.readUnsignedVInt32(); Map map = Maps.newHashMapWithExpectedSize(size); for (int i = 0; i < size; i++) { @@ -286,13 +286,9 @@ public class DataPlacements extends ReplicationMap implements Met public long serializedSize(DataPlacements t, Version version) { 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); + long size = version.isBefore(Version.V9) ? sizeof(map.size()) : sizeofUnsignedVInt(map.size()); 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); } diff --git a/src/java/org/apache/cassandra/tcm/ownership/UniformRangePlacement.java b/src/java/org/apache/cassandra/tcm/ownership/UniformRangePlacement.java index 3d3b4da374..26ecdc33e2 100644 --- a/src/java/org/apache/cassandra/tcm/ownership/UniformRangePlacement.java +++ b/src/java/org/apache/cassandra/tcm/ownership/UniformRangePlacement.java @@ -329,7 +329,11 @@ public class UniformRangePlacement implements PlacementProvider logger.trace("Calculating data placements for {}", ksMetadata.name); AbstractReplicationStrategy replication = ksMetadata.replicationStrategy; ReplicationParams params = ksMetadata.params.replication; - if (params.isMeta() || params.isLocal()) + if (params.isMeta()) + { + // don't calculate meta strategy placements, these are derived from ClusterMetadata.cmsMembership + } + else if (params.isLocal()) { placements.put(params, metadata.placements.get(params)); } diff --git a/src/java/org/apache/cassandra/tcm/serialization/Version.java b/src/java/org/apache/cassandra/tcm/serialization/Version.java index a9ecfe0727..fb6e59cfd6 100644 --- a/src/java/org/apache/cassandra/tcm/serialization/Version.java +++ b/src/java/org/apache/cassandra/tcm/serialization/Version.java @@ -80,6 +80,7 @@ public enum Version V8(8), /** * - DataPlacements don't include MetaStrategy, replaced by ClusterMetadata.CMSMembership + * - Size of DataPlacements is encoded as vint */ V9(9),