You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sa...@apache.org on 2023/03/14 16:45:27 UTC
[cassandra] 11/25: [CEP-21] Placement and ownership related changes (2/7)
This is an automated email from the ASF dual-hosted git repository.
samt pushed a commit to branch cep-21-tcm
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 8425d3177da20b96c970cc5f07386d1e7eee273a
Author: Sam Tunnicliffe <sa...@apache.org>
AuthorDate: Thu Mar 2 18:45:44 2023 +0000
[CEP-21] Placement and ownership related changes (2/7)
Part 2 of 7 moves most of the data placement/ownership code over to the
TCM structures.
Co-authored-by: Marcus Eriksson <ma...@apache.org>
Co-authored-by: Alex Petrov <ol...@gmail.com>
Co-authored-by: Sam Tunnicliffe <sa...@apache.org>
---
.../cassandra/auth/CassandraRoleManager.java | 9 +-
.../org/apache/cassandra/auth/DCPermissions.java | 3 +-
.../apache/cassandra/batchlog/BatchlogManager.java | 41 ++-
.../statements/schema/AlterKeyspaceStatement.java | 12 +-
.../statements/schema/AlterTableStatement.java | 3 +-
.../statements/schema/CreateKeyspaceStatement.java | 4 +-
.../apache/cassandra/db/SizeEstimatesRecorder.java | 34 ++-
.../org/apache/cassandra/db/view/ViewUtils.java | 17 +-
.../org/apache/cassandra/dht/BootStrapper.java | 35 ++-
.../apache/cassandra/dht/BootstrapDiagnostics.java | 15 +-
.../org/apache/cassandra/dht/BootstrapEvent.java | 10 +-
src/java/org/apache/cassandra/dht/Datacenters.java | 16 +-
.../org/apache/cassandra/dht/IPartitioner.java | 4 +-
.../org/apache/cassandra/dht/RangeStreamer.java | 259 +++++++++--------
.../dht/tokenallocator/OfflineTokenAllocator.java | 9 +-
.../dht/tokenallocator/TokenAllocation.java | 80 ++---
.../cassandra/hints/HintsDispatchExecutor.java | 3 +-
.../locator/AbstractReplicationStrategy.java | 262 +++--------------
.../cassandra/locator/EndpointsForToken.java | 30 +-
.../apache/cassandra/locator/LocalStrategy.java | 52 +---
.../org/apache/cassandra/locator/MetaStrategy.java | 27 +-
.../cassandra/locator/NetworkTopologyStrategy.java | 133 +++++----
.../apache/cassandra/locator/PendingRangeMaps.java | 212 --------------
.../apache/cassandra/locator/ReplicaLayout.java | 101 ++++++-
.../apache/cassandra/locator/SimpleStrategy.java | 36 ++-
.../apache/cassandra/locator/SystemStrategy.java | 7 +-
.../apache/cassandra/schema/KeyspaceMetadata.java | 2 +-
.../apache/cassandra/schema/KeyspaceParams.java | 7 +-
.../apache/cassandra/schema/PartitionDenylist.java | 6 +-
.../apache/cassandra/schema/ReplicationParams.java | 11 +-
.../cassandra/service/ActiveRepairService.java | 106 ++++---
.../apache/cassandra/service/CassandraDaemon.java | 37 ++-
.../service/PendingRangeCalculatorService.java | 98 -------
.../PendingRangeCalculatorServiceDiagnostics.java | 66 -----
.../PendingRangeCalculatorServiceEvent.java | 75 -----
.../apache/cassandra/service/RangeRelocator.java | 323 ---------------------
.../org/apache/cassandra/service/StorageProxy.java | 113 +++----
.../apache/cassandra/service/StorageService.java | 2 +-
.../cassandra/service/StorageServiceMBean.java | 6 +-
.../cassandra/service/paxos/PaxosPrepare.java | 10 +-
.../cassandra/service/paxos/PaxosRepair.java | 39 +--
.../service/paxos/cleanup/PaxosCleanup.java | 18 +-
.../paxos/cleanup/PaxosStartPrepareCleanup.java | 3 -
.../paxos/uncommitted/UncommittedTableData.java | 2 +-
.../service/reads/AbstractReadExecutor.java | 9 +-
.../org/apache/cassandra/tcm/ClusterMetadata.java | 73 +++++
.../cassandra/tcm/compatibility/AsEndpoints.java | 28 --
.../cassandra/tcm/compatibility/AsLocations.java | 38 ---
.../cassandra/tcm/compatibility/AsTokenMap.java | 33 ---
.../tcm/listeners/PlacementsChangeListener.java | 53 ++++
.../org/apache/cassandra/tcm/log/LocalLog.java | 2 +
.../apache/cassandra/tcm/membership/Directory.java | 4 +-
.../apache/cassandra/tcm/ownership/TokenMap.java | 3 +-
53 files changed, 844 insertions(+), 1737 deletions(-)
diff --git a/src/java/org/apache/cassandra/auth/CassandraRoleManager.java b/src/java/org/apache/cassandra/auth/CassandraRoleManager.java
index 84c79273db..0918a5b1e6 100644
--- a/src/java/org/apache/cassandra/auth/CassandraRoleManager.java
+++ b/src/java/org/apache/cassandra/auth/CassandraRoleManager.java
@@ -36,17 +36,18 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.exceptions.*;
+import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.StorageProxy;
-import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
import org.mindrot.jbcrypt.BCrypt;
import static org.apache.cassandra.service.QueryState.forInternalCalls;
@@ -343,7 +344,7 @@ public class CassandraRoleManager implements IRoleManager
*/
private static void setupDefaultRole()
{
- if (StorageService.instance.getTokenMetadata().sortedTokens().isEmpty())
+ if (ClusterMetadata.current().tokenMap.tokens().isEmpty())
throw new IllegalStateException("CassandraRoleManager skipped default role setup: no known tokens in ring");
try
@@ -413,7 +414,7 @@ public class CassandraRoleManager implements IRoleManager
}
catch (RequestValidationException e)
{
- throw new AssertionError(e); // not supposed to happen
+ throw new AssertionError(e + " " + FBUtilities.getJustLocalAddress()); // not supposed to happen
}
}
diff --git a/src/java/org/apache/cassandra/auth/DCPermissions.java b/src/java/org/apache/cassandra/auth/DCPermissions.java
index d04242dd7e..c8a7716f45 100644
--- a/src/java/org/apache/cassandra/auth/DCPermissions.java
+++ b/src/java/org/apache/cassandra/auth/DCPermissions.java
@@ -28,6 +28,7 @@ import com.google.common.collect.Sets;
import org.apache.cassandra.dht.Datacenters;
import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.tcm.ClusterMetadata;
public abstract class DCPermissions
{
@@ -93,7 +94,7 @@ public abstract class DCPermissions
public void validate()
{
- Set<String> unknownDcs = Sets.difference(subset, Datacenters.getValidDatacenters());
+ Set<String> unknownDcs = Sets.difference(subset, Datacenters.getValidDatacenters(ClusterMetadata.current()));
if (!unknownDcs.isEmpty())
{
throw new InvalidRequestException(String.format("Invalid value(s) for DATACENTERS '%s'," +
diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
index 6ff5aee0cf..6fc9507f8e 100644
--- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
+++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
@@ -36,6 +36,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Future;
import org.slf4j.Logger;
@@ -202,7 +204,7 @@ public class BatchlogManager implements BatchlogManagerMBean
// rate limit is in bytes per second. Uses Double.MAX_VALUE if disabled (set to 0 in cassandra.yaml).
// max rate is scaled by the number of nodes in the cluster (same as for HHOM - see CASSANDRA-5272).
- int endpointsCount = StorageService.instance.getTokenMetadata().getSizeOfAllEndpoints();
+ int endpointsCount = ClusterMetadata.current().directory.allAddresses().size();
if (endpointsCount <= 0)
{
logger.trace("Replay cancelled as there are no peers in the ring.");
@@ -233,7 +235,7 @@ public class BatchlogManager implements BatchlogManagerMBean
*/
public void setRate(final int throttleInKB)
{
- int endpointsCount = StorageService.instance.getTokenMetadata().getSizeOfAllEndpoints();
+ int endpointsCount = ClusterMetadata.current().directory.allAddresses().size();
if (endpointsCount > 0)
{
int endpointThrottleInKiB = throttleInKB / endpointsCount;
@@ -470,29 +472,27 @@ public class BatchlogManager implements BatchlogManagerMBean
Set<UUID> hintedNodes)
{
String ks = mutation.getKeyspaceName();
- Keyspace keyspace = Keyspace.open(ks);
Token tk = mutation.key().getToken();
+ ClusterMetadata metadata = ClusterMetadata.current();
+ KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaceMetadata(ks);
// TODO: this logic could do with revisiting at some point, as it is unclear what its rationale is
// we perform a local write, ignoring errors and inline in this thread (potentially slowing replay down)
// effectively bumping CL for locally owned writes and also potentially stalling log replay if an error occurs
// once we decide how it should work, it can also probably be simplified, and avoid constructing a ReplicaPlan directly
- ReplicaLayout.ForTokenWrite liveAndDown = ReplicaLayout.forTokenWriteLiveAndDown(keyspace, tk);
- Replicas.temporaryAssertFull(liveAndDown.all()); // TODO in CASSANDRA-14549
+ ReplicaLayout.ForTokenWrite allReplias = ReplicaLayout.forTokenWriteLiveAndDown(metadata, keyspaceMetadata, tk);
+ ReplicaPlan.ForWrite replicaPlan = forReplayMutation(metadata, Keyspace.open(ks), tk);
- Replica selfReplica = liveAndDown.all().selfIfPresent();
+ Replica selfReplica = allReplias.all().selfIfPresent();
if (selfReplica != null)
mutation.apply();
- ReplicaLayout.ForTokenWrite liveRemoteOnly = liveAndDown.filter(
- r -> FailureDetector.isReplicaAlive.test(r) && r != selfReplica);
-
- for (Replica replica : liveAndDown.all())
+ for (Replica replica : allReplias.all())
{
- if (replica == selfReplica || liveRemoteOnly.all().contains(replica))
+ if (replica == selfReplica || replicaPlan.liveAndDown().contains(replica))
continue;
- UUID hostId = StorageService.instance.getHostIdForEndpoint(replica.endpoint());
+ UUID hostId = metadata.directory.peerId(replica.endpoint()).uuid;
if (null != hostId)
{
HintsService.instance.write(hostId, Hint.create(mutation, writtenAt));
@@ -500,15 +500,26 @@ public class BatchlogManager implements BatchlogManagerMBean
}
}
- ReplicaPlan.ForWrite replicaPlan = new ReplicaPlan.ForWrite(keyspace, liveAndDown.replicationStrategy(),
- ConsistencyLevel.ONE, liveRemoteOnly.pending(), liveRemoteOnly.all(), liveRemoteOnly.all(), liveRemoteOnly.all());
ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(replicaPlan, mutation, nanoTime());
Message<Mutation> message = Message.outWithFlag(MUTATION_REQ, mutation, MessageFlag.CALL_BACK_ON_FAILURE);
- for (Replica replica : liveRemoteOnly.all())
+ for (Replica replica : replicaPlan.liveAndDown())
MessagingService.instance().sendWriteWithCallback(message, replica, handler, false);
return handler;
}
+ public static ReplicaPlan.ForWrite forReplayMutation(ClusterMetadata metadata, Keyspace keyspace, Token token)
+ {
+ ReplicaLayout.ForTokenWrite liveAndDown = ReplicaLayout.forTokenWriteLiveAndDown(metadata, keyspace.getMetadata(), token);
+ Replicas.temporaryAssertFull(liveAndDown.all()); // TODO in CASSANDRA-14549
+
+ Replica selfReplica = liveAndDown.all().selfIfPresent();
+ ReplicaLayout.ForTokenWrite liveRemoteOnly = liveAndDown.filter(r -> FailureDetector.isReplicaAlive.test(r) && r != selfReplica);
+
+ return new ReplicaPlan.ForWrite(keyspace, liveAndDown.replicationStrategy(),
+ ConsistencyLevel.ONE, liveRemoteOnly.pending(), liveRemoteOnly.all(), liveRemoteOnly.all(), liveRemoteOnly.all(),
+ (cm) -> forReplayMutation(cm, keyspace, token),
+ liveAndDown.epoch());
+ }
private static int gcgs(Collection<Mutation> mutations)
{
int gcgs = Integer.MAX_VALUE;
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
index 250b18ffed..6e89fc5364 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterKeyspaceStatement.java
@@ -18,10 +18,8 @@
package org.apache.cassandra.cql3.statements.schema;
import java.util.HashSet;
-import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,7 +34,6 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.LocalStrategy;
@@ -90,15 +87,16 @@ public final class AlterKeyspaceStatement extends AlterSchemaStatement
if (newKeyspace.params.replication.klass.equals(LocalStrategy.class))
throw ire("Unable to use given strategy class: LocalStrategy is reserved for internal use.");
- newKeyspace.params.validate(keyspaceName, state);
- newKeyspace.replicationStrategy.validate();
+ newKeyspace.params.validate(keyspaceName, state, metadata);
+ newKeyspace.replicationStrategy.validate(metadata);
+
validateNoRangeMovements();
validateTransientReplication(keyspace.replicationStrategy, newKeyspace.replicationStrategy);
// Because we used to not properly validate unrecognized options, we only log a warning if we find one.
try
{
- newKeyspace.replicationStrategy.validateExpectedOptions();
+ newKeyspace.replicationStrategy.validateExpectedOptions(metadata);
}
catch (ConfigurationException e)
{
@@ -153,7 +151,9 @@ public final class AlterKeyspaceStatement extends AlterSchemaStatement
default:
return false;
}
+
}).map(e -> metadata.directory.endpoint(e.getKey())).collect(Collectors.toSet());
+
if (!notNormalEndpoints.isEmpty())
{
throw new ConfigurationException("Cannot alter RF while some endpoints are not in normal state (no range movements): " + notNormalEndpoints);
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
index 875f22aca4..c19270d566 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterTableStatement.java
@@ -57,7 +57,6 @@ import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.schema.ViewMetadata;
import org.apache.cassandra.schema.Views;
import org.apache.cassandra.service.ClientState;
-import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
@@ -524,7 +523,7 @@ public abstract class AlterTableStatement extends AlterSchemaStatement
Set<InetAddressAndPort> preC15897nodes = new HashSet<>();
Set<InetAddressAndPort> with2xSStables = new HashSet<>();
Splitter onComma = Splitter.on(',').omitEmptyStrings().trimResults();
- for (InetAddressAndPort node : StorageService.instance.getTokenMetadata().getAllEndpoints())
+ for (InetAddressAndPort node : ClusterMetadata.current().directory.allAddresses())
{
if (MessagingService.instance().versions.knows(node) &&
MessagingService.instance().versions.getRaw(node) < MessagingService.VERSION_40)
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
index 7fdff7b6be..dff503f10e 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/CreateKeyspaceStatement.java
@@ -82,8 +82,8 @@ public final class CreateKeyspaceStatement extends AlterSchemaStatement
if (keyspace.params.replication.klass.equals(LocalStrategy.class))
throw ire("Unable to use given strategy class: LocalStrategy is reserved for internal use.");
- keyspace.params.validate(keyspaceName, state);
- keyspace.replicationStrategy.validateExpectedOptions();
+ keyspace.params.validate(keyspaceName, state, metadata);
+ keyspace.replicationStrategy.validateExpectedOptions(metadata);
return schema.withAddedOrUpdated(keyspace);
}
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index dfd52c5f38..be3b9d05af 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -19,7 +19,9 @@ package org.apache.cassandra.db;
import java.util.*;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,15 +31,17 @@ import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.format.SSTableReader;
-import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaChangeListener;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Refs;
+import static org.apache.cassandra.tcm.compatibility.TokenRingUtils.getAllRanges;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
@@ -62,8 +66,7 @@ public class SizeEstimatesRecorder implements SchemaChangeListener, Runnable
public void run()
{
- TokenMetadata metadata = StorageService.instance.getTokenMetadata().cloneOnlyTokenMap();
- if (!metadata.isMember(FBUtilities.getBroadcastAddressAndPort()))
+ if (!ClusterMetadata.current().directory.allAddresses().contains(FBUtilities.getBroadcastAddressAndPort()))
{
logger.debug("Node is not part of the ring; not recording size estimates");
return;
@@ -73,6 +76,9 @@ public class SizeEstimatesRecorder implements SchemaChangeListener, Runnable
for (Keyspace keyspace : Keyspace.nonLocalStrategy())
{
+ if (keyspace.getMetadata().params.replication.isMeta())
+ continue;
+
// In tools the call to describe_splits_ex() used to be coupled with the call to describe_local_ring() so
// most access was for the local primary range; after creating the size_estimates table this was changed
// to be the primary range.
@@ -88,7 +94,7 @@ public class SizeEstimatesRecorder implements SchemaChangeListener, Runnable
// range. If we publish multiple ranges downstream integrations may start to see duplicate data.
// See CASSANDRA-15637
Collection<Range<Token>> primaryRanges = StorageService.instance.getPrimaryRanges(keyspace.getName());
- Collection<Range<Token>> localPrimaryRanges = StorageService.instance.getLocalPrimaryRange();
+ Collection<Range<Token>> localPrimaryRanges = getLocalPrimaryRange();
boolean rangesAreEqual = primaryRanges.equals(localPrimaryRanges);
for (ColumnFamilyStore table : keyspace.getColumnFamilyStores())
{
@@ -116,6 +122,26 @@ public class SizeEstimatesRecorder implements SchemaChangeListener, Runnable
}
}
+ private static Collection<Range<Token>> getLocalPrimaryRange()
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ NodeId nodeId = metadata.myNodeId();
+ String dc = metadata.directory.location(nodeId).datacenter;
+ Set<Token> tokens = new HashSet<>(metadata.tokenMap.tokens(nodeId));
+
+ // filter tokens to the single DC
+ List<Token> filteredTokens = Lists.newArrayList();
+ for (Token token : metadata.tokenMap.tokens())
+ {
+ NodeId owner = metadata.tokenMap.owner(token);
+ if (dc.equals(metadata.directory.location(owner).datacenter))
+ filteredTokens.add(token);
+ }
+ return getAllRanges(filteredTokens).stream()
+ .filter(t -> tokens.contains(t.right))
+ .collect(Collectors.toList());
+ }
+
@SuppressWarnings("resource")
private static Map<Range<Token>, Pair<Long, Long>> computeSizeEstimates(ColumnFamilyStore table, Collection<Range<Token>> ranges)
{
diff --git a/src/java/org/apache/cassandra/db/view/ViewUtils.java b/src/java/org/apache/cassandra/db/view/ViewUtils.java
index 55a462c31b..4ac63ea702 100644
--- a/src/java/org/apache/cassandra/db/view/ViewUtils.java
+++ b/src/java/org/apache/cassandra/db/view/ViewUtils.java
@@ -24,10 +24,11 @@ import java.util.function.Predicate;
import com.google.common.collect.Iterables;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.EndpointsForToken;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
public final class ViewUtils
{
@@ -57,11 +58,13 @@ public final class ViewUtils
*
* @return Optional.empty() if this method is called using a base token which does not belong to this replica
*/
- public static Optional<Replica> getViewNaturalEndpoint(AbstractReplicationStrategy replicationStrategy, Token baseToken, Token viewToken)
+ public static Optional<Replica> getViewNaturalEndpoint(ClusterMetadata metadata, String keyspace, Token baseToken, Token viewToken)
{
String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getLocalDatacenter();
- EndpointsForToken naturalBaseReplicas = replicationStrategy.getNaturalReplicasForToken(baseToken);
- EndpointsForToken naturalViewReplicas = replicationStrategy.getNaturalReplicasForToken(viewToken);
+ KeyspaceMetadata keyspaceMetadata = metadata.schema.getKeyspaces().getNullable(keyspace);
+
+ EndpointsForToken naturalBaseReplicas = metadata.placements.get(keyspaceMetadata.params.replication).reads.forToken(baseToken);
+ EndpointsForToken naturalViewReplicas = metadata.placements.get(keyspaceMetadata.params.replication).reads.forToken(viewToken);
Optional<Replica> localReplica = Iterables.tryFind(naturalViewReplicas, Replica::isSelf).toJavaUtil();
if (localReplica.isPresent())
@@ -69,7 +72,7 @@ public final class ViewUtils
// We only select replicas from our own DC
// TODO: this is poor encapsulation, leaking implementation details of replication strategy
- Predicate<Replica> isLocalDC = r -> !(replicationStrategy instanceof NetworkTopologyStrategy)
+ Predicate<Replica> isLocalDC = r -> !(keyspaceMetadata.replicationStrategy instanceof NetworkTopologyStrategy)
|| DatabaseDescriptor.getEndpointSnitch().getDatacenter(r).equals(localDataCenter);
// We have to remove any endpoint which is shared between the base and the view, as it will select itself
@@ -84,7 +87,9 @@ public final class ViewUtils
// The replication strategy will be the same for the base and the view, as they must belong to the same keyspace.
// Since the same replication strategy is used, the same placement should be used and we should get the same
// number of replicas for all of the tokens in the ring.
- assert baseReplicas.size() == viewReplicas.size() : "Replication strategy should have the same number of endpoints for the base and the view";
+ assert baseReplicas.size() == viewReplicas.size() :
+ String.format("Replication strategy should have the same number of endpoints for the base (%d) and the view (%d)",
+ baseReplicas.size(), viewReplicas.size());
int baseIdx = -1;
for (int i=0; i<baseReplicas.size(); i++)
diff --git a/src/java/org/apache/cassandra/dht/BootStrapper.java b/src/java/org/apache/cassandra/dht/BootStrapper.java
index 4b1e43400e..0cebd439ce 100644
--- a/src/java/org/apache/cassandra/dht/BootStrapper.java
+++ b/src/java/org/apache/cassandra/dht/BootStrapper.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.dht;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -31,10 +32,9 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.tokenallocator.TokenAllocation;
import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.TokenMetadata;
+import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.streaming.StreamEvent;
import org.apache.cassandra.streaming.StreamEventHandler;
@@ -200,29 +200,26 @@ public class BootStrapper extends ProgressEventNotifierSupport
return tokens;
}
- private static Collection<Token> getSpecifiedTokens(final TokenMetadata metadata,
+ private static Collection<Token> getSpecifiedTokens(final ClusterMetadata metadata,
Collection<String> initialTokens)
{
logger.info("tokens manually specified as {}", initialTokens);
List<Token> tokens = new ArrayList<>(initialTokens.size());
for (String tokenString : initialTokens)
{
- Token token = metadata.partitioner.getTokenFactory().fromString(tokenString);
- if (metadata.getEndpoint(token) != null)
+ Token token = metadata.tokenMap.partitioner().getTokenFactory().fromString(tokenString);
+ if (metadata.tokenMap.owner(token) != null)
throw new ConfigurationException("Bootstrapping to existing token " + tokenString + " is not allowed (decommission/removenode the old node first).");
tokens.add(token);
}
return tokens;
}
- static Collection<Token> allocateTokens(final TokenMetadata metadata,
+ static Collection<Token> allocateTokens(final ClusterMetadata metadata,
InetAddressAndPort address,
String allocationKeyspace,
int numTokens)
{
- if (!FBUtilities.getBroadcastAddressAndPort().equals(InetAddressAndPort.getLoopbackAddress()))
- Gossiper.waitToSettle();
-
Keyspace ks = Keyspace.open(allocationKeyspace);
if (ks == null)
throw new ConfigurationException("Problem opening token allocation keyspace " + allocationKeyspace);
@@ -234,30 +231,36 @@ public class BootStrapper extends ProgressEventNotifierSupport
}
- static Collection<Token> allocateTokens(final TokenMetadata metadata,
+ static Collection<Token> allocateTokens(final ClusterMetadata metadata,
InetAddressAndPort address,
int rf,
int numTokens)
{
- if (!FBUtilities.getBroadcastAddressAndPort().equals(InetAddressAndPort.getLoopbackAddress()))
- Gossiper.waitToSettle();
-
Collection<Token> tokens = TokenAllocation.allocateTokens(metadata, rf, address, numTokens);
BootstrapDiagnostics.tokensAllocated(address, metadata, rf, numTokens, tokens);
return tokens;
}
- public static Collection<Token> getRandomTokens(TokenMetadata metadata, int numTokens)
+ public static Set<Token> getRandomTokens(ClusterMetadata metadata, int numTokens)
{
Set<Token> tokens = new HashSet<>(numTokens);
while (tokens.size() < numTokens)
{
- Token token = metadata.partitioner.getRandomToken();
- if (metadata.getEndpoint(token) == null)
+ Token token = metadata.tokenMap.partitioner().getRandomToken();
+ if (metadata.tokenMap.owner(token) == null)
tokens.add(token);
}
logger.info("Generated random tokens. tokens are {}", tokens);
return tokens;
}
+
+ public String toString()
+ {
+ return "BootStrapper{" +
+ "address=" + address +
+ ", tokens=" + tokens +
+ ", metadata=" + metadata +
+ '}';
+ }
}
diff --git a/src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java b/src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java
index 5c2b46a030..3f3bfbe2e4 100644
--- a/src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java
+++ b/src/java/org/apache/cassandra/dht/BootstrapDiagnostics.java
@@ -19,12 +19,13 @@
package org.apache.cassandra.dht;
import java.util.Collection;
+
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.dht.BootstrapEvent.BootstrapEventType;
import org.apache.cassandra.diag.DiagnosticEventService;
import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.TokenMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
/**
* Utility methods for bootstrap related activities.
@@ -50,38 +51,38 @@ final class BootstrapDiagnostics
ImmutableList.copyOf(initialTokens)));
}
- static void useRandomTokens(InetAddressAndPort address, TokenMetadata metadata, int numTokens, Collection<Token> tokens)
+ static void useRandomTokens(InetAddressAndPort address, ClusterMetadata metadata, int numTokens, Collection<Token> tokens)
{
if (isEnabled(BootstrapEventType.BOOTSTRAP_USING_RANDOM_TOKENS))
service.publish(new BootstrapEvent(BootstrapEventType.BOOTSTRAP_USING_RANDOM_TOKENS,
address,
- metadata.cloneOnlyTokenMap(),
+ metadata,
null,
null,
numTokens,
ImmutableList.copyOf(tokens)));
}
- static void tokensAllocated(InetAddressAndPort address, TokenMetadata metadata,
+ static void tokensAllocated(InetAddressAndPort address, ClusterMetadata metadata,
String allocationKeyspace, int numTokens, Collection<Token> tokens)
{
if (isEnabled(BootstrapEventType.TOKENS_ALLOCATED))
service.publish(new BootstrapEvent(BootstrapEventType.TOKENS_ALLOCATED,
address,
- metadata.cloneOnlyTokenMap(),
+ metadata,
allocationKeyspace,
null,
numTokens,
ImmutableList.copyOf(tokens)));
}
- static void tokensAllocated(InetAddressAndPort address, TokenMetadata metadata,
+ static void tokensAllocated(InetAddressAndPort address, ClusterMetadata metadata,
int rf, int numTokens, Collection<Token> tokens)
{
if (isEnabled(BootstrapEventType.TOKENS_ALLOCATED))
service.publish(new BootstrapEvent(BootstrapEventType.TOKENS_ALLOCATED,
address,
- metadata.cloneOnlyTokenMap(),
+ metadata,
null,
rf,
numTokens,
diff --git a/src/java/org/apache/cassandra/dht/BootstrapEvent.java b/src/java/org/apache/cassandra/dht/BootstrapEvent.java
index 4936c2942a..e5a5cbc078 100644
--- a/src/java/org/apache/cassandra/dht/BootstrapEvent.java
+++ b/src/java/org/apache/cassandra/dht/BootstrapEvent.java
@@ -28,7 +28,7 @@ import com.google.common.collect.ImmutableCollection;
import org.apache.cassandra.diag.DiagnosticEvent;
import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.TokenMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
/**
* DiagnosticEvent implementation for bootstrap related activities.
@@ -38,7 +38,7 @@ final class BootstrapEvent extends DiagnosticEvent
private final BootstrapEventType type;
@Nullable
- private final TokenMetadata tokenMetadata;
+ private final ClusterMetadata metadata;
private final InetAddressAndPort address;
@Nullable
private final String allocationKeyspace;
@@ -47,12 +47,12 @@ final class BootstrapEvent extends DiagnosticEvent
private final Integer numTokens;
private final Collection<Token> tokens;
- BootstrapEvent(BootstrapEventType type, InetAddressAndPort address, @Nullable TokenMetadata tokenMetadata,
+ BootstrapEvent(BootstrapEventType type, InetAddressAndPort address, @Nullable ClusterMetadata metadata,
@Nullable String allocationKeyspace, @Nullable Integer rf, int numTokens, ImmutableCollection<Token> tokens)
{
this.type = type;
this.address = address;
- this.tokenMetadata = tokenMetadata;
+ this.metadata = metadata;
this.allocationKeyspace = allocationKeyspace;
this.rf = rf;
this.numTokens = numTokens;
@@ -76,7 +76,7 @@ final class BootstrapEvent extends DiagnosticEvent
{
// be extra defensive against nulls and bugs
HashMap<String, Serializable> ret = new HashMap<>();
- ret.put("tokenMetadata", String.valueOf(tokenMetadata));
+ ret.put("metadata", metadata.legacyToString());
ret.put("allocationKeyspace", allocationKeyspace);
ret.put("rf", rf);
ret.put("numTokens", numTokens);
diff --git a/src/java/org/apache/cassandra/dht/Datacenters.java b/src/java/org/apache/cassandra/dht/Datacenters.java
index b1d96eb729..2c0328e8d3 100644
--- a/src/java/org/apache/cassandra/dht/Datacenters.java
+++ b/src/java/org/apache/cassandra/dht/Datacenters.java
@@ -22,13 +22,10 @@ import java.util.HashSet;
import java.util.Set;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.locator.IEndpointSnitch;
-import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
public class Datacenters
{
-
private static class DCHandle
{
private static final String thisDc = DatabaseDescriptor.getEndpointSnitch().getLocalDatacenter();
@@ -41,22 +38,15 @@ public class Datacenters
/*
* (non-javadoc) Method to generate list of valid data center names to be used to validate the replication parameters during CREATE / ALTER keyspace operations.
- * All peers of current node are fetched from {@link TokenMetadata} and then a set is build by fetching DC name of each peer.
* @return a set of valid DC names
*/
- public static Set<String> getValidDatacenters()
+ public static Set<String> getValidDatacenters(ClusterMetadata metadata)
{
final Set<String> validDataCenters = new HashSet<>();
- final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
-
// Add data center of localhost.
validDataCenters.add(thisDatacenter());
// Fetch and add DCs of all peers.
- for (InetAddressAndPort peer : StorageService.instance.getTokenMetadata().getAllEndpoints())
- {
- validDataCenters.add(snitch.getDatacenter(peer));
- }
-
+ validDataCenters.addAll(metadata.directory.knownDatacenters());
return validDataCenters;
}
}
diff --git a/src/java/org/apache/cassandra/dht/IPartitioner.java b/src/java/org/apache/cassandra/dht/IPartitioner.java
index ef8ced25b1..1a5518b264 100644
--- a/src/java/org/apache/cassandra/dht/IPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/IPartitioner.java
@@ -26,13 +26,13 @@ import java.util.Random;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
public interface IPartitioner
{
static IPartitioner global()
{
- return StorageService.instance.getTokenMetadata().partitioner;
+ return ClusterMetadata.current().partitioner;
}
static void validate(Collection<? extends AbstractBounds<?>> allBounds)
diff --git a/src/java/org/apache/cassandra/dht/RangeStreamer.java b/src/java/org/apache/cassandra/dht/RangeStreamer.java
index 9b7833b90a..6b75a8d7d6 100644
--- a/src/java/org/apache/cassandra/dht/RangeStreamer.java
+++ b/src/java/org/apache/cassandra/dht/RangeStreamer.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.dht;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -35,9 +36,8 @@ import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Multimap;
-
+import com.google.common.collect.Multimaps;
import org.apache.commons.lang3.StringUtils;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,11 +60,13 @@ import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaCollection;
import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import org.apache.cassandra.locator.Replicas;
-import org.apache.cassandra.locator.TokenMetadata;
+import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.streaming.StreamOperation;
import org.apache.cassandra.streaming.StreamPlan;
import org.apache.cassandra.streaming.StreamResultFuture;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ownership.PlacementForRange;
import org.apache.cassandra.utils.FBUtilities;
import static com.google.common.base.Predicates.and;
@@ -82,17 +84,14 @@ public class RangeStreamer
private static final Logger logger = LoggerFactory.getLogger(RangeStreamer.class);
public static Predicate<Replica> ALIVE_PREDICATE = replica ->
- (!Gossiper.instance.isEnabled() ||
- (Gossiper.instance.getEndpointStateForEndpoint(replica.endpoint()) == null ||
- Gossiper.instance.getEndpointStateForEndpoint(replica.endpoint()).isAlive())) &&
- FailureDetector.instance.isAlive(replica.endpoint());
+ (!Gossiper.instance.isEnabled() ||
+ (Gossiper.instance.getEndpointStateForEndpoint(replica.endpoint()) == null ||
+ Gossiper.instance.getEndpointStateForEndpoint(replica.endpoint()).isAlive())) &&
+ FailureDetector.instance.isAlive(replica.endpoint());
+ private final ClusterMetadata metadata;
/* bootstrap tokens. can be null if replacing the node. */
private final Collection<Token> tokens;
- /* current token ring */
- private final TokenMetadata metadata;
- /* address of this node */
- private final InetAddressAndPort address;
/* streaming description */
private final String description;
private final Map<String, Multimap<InetAddressAndPort, FetchReplica>> toFetch = new HashMap<>();
@@ -272,9 +271,29 @@ public class RangeStreamer
}
}
- public RangeStreamer(TokenMetadata metadata,
+ public static class ExcludedSourcesFilter implements SourceFilter
+ {
+ private final Set<InetAddressAndPort> excludedSources;
+
+ public ExcludedSourcesFilter(Set<InetAddressAndPort> allowedSources)
+ {
+ this.excludedSources = allowedSources;
+ }
+
+ public boolean apply(Replica replica)
+ {
+ return !excludedSources.contains(replica.endpoint());
+ }
+
+ @Override
+ public String message(Replica replica)
+ {
+ return "Filtered " + replica + " out because it was in the excluded set: " + excludedSources;
+ }
+ }
+
+ public RangeStreamer(ClusterMetadata metadata,
Collection<Token> tokens,
- InetAddressAndPort address,
StreamOperation streamOperation,
boolean useStrictConsistency,
IEndpointSnitch snitch,
@@ -282,13 +301,12 @@ public class RangeStreamer
boolean connectSequentially,
int connectionsPerHost)
{
- this(metadata, tokens, address, streamOperation, useStrictConsistency, snitch, stateStore,
+ this(metadata, tokens, streamOperation, useStrictConsistency, snitch, stateStore,
FailureDetector.instance, connectSequentially, connectionsPerHost);
}
- RangeStreamer(TokenMetadata metadata,
+ RangeStreamer(ClusterMetadata metadata,
Collection<Token> tokens,
- InetAddressAndPort address,
StreamOperation streamOperation,
boolean useStrictConsistency,
IEndpointSnitch snitch,
@@ -300,7 +318,6 @@ public class RangeStreamer
Preconditions.checkArgument(streamOperation == StreamOperation.BOOTSTRAP || streamOperation == StreamOperation.REBUILD, streamOperation);
this.metadata = metadata;
this.tokens = tokens;
- this.address = address;
this.description = streamOperation.getDescription();
this.streamPlan = new StreamPlan(streamOperation, connectionsPerHost, connectSequentially, null, PreviewKind.NONE);
this.useStrictConsistency = useStrictConsistency;
@@ -352,7 +369,13 @@ public class RangeStreamer
}
boolean useStrictSource = useStrictSourcesForRanges(strat);
- EndpointsByReplica fetchMap = calculateRangesToFetchWithPreferredEndpoints(replicas, keyspace, useStrictSource);
+ EndpointsByReplica fetchMap = calculateRangesToFetchWithPreferredEndpoints(snitch::sortedByProximity,
+ keyspace.getReplicationStrategy(),
+ replicas,
+ useStrictConsistency,
+ metadata,
+ keyspace.getName(),
+ sourceFilters);
for (Map.Entry<Replica, Replica> entry : fetchMap.flattenEntries())
logger.info("{}: range {} exists on {} for keyspace {}", description, entry.getKey(), entry.getValue(), keyspaceName);
@@ -387,61 +410,61 @@ public class RangeStreamer
* @return true when the node is bootstrapping, useStrictConsistency is true and # of nodes in the cluster is more than # of replica
*/
private boolean useStrictSourcesForRanges(AbstractReplicationStrategy strat)
+ {
+ return useStrictSourcesForRanges(strat, metadata, snitch, useStrictConsistency, tokens);
+ }
+
+ private static boolean useStrictSourcesForRanges(AbstractReplicationStrategy strat, ClusterMetadata metadata, IEndpointSnitch snitch, boolean useStrictConsistency, Collection<Token> tokens)
{
boolean res = useStrictConsistency && tokens != null;
-
+
if (res)
{
int nodes = 0;
+ HashSet<InetAddressAndPort> allOtherNodes = new HashSet<>(metadata.directory.allAddresses());
+ allOtherNodes.remove(FBUtilities.getBroadcastAddressAndPort());
+
if (strat instanceof NetworkTopologyStrategy)
{
- ImmutableMultimap<String, InetAddressAndPort> dc2Nodes = metadata.getDC2AllEndpoints(snitch);
+ ImmutableMultimap<String, InetAddressAndPort> dc2Nodes = Multimaps.index(allOtherNodes, snitch::getDatacenter);
NetworkTopologyStrategy ntps = (NetworkTopologyStrategy) strat;
for (String dc : dc2Nodes.keySet())
nodes += ntps.getReplicationFactor(dc).allReplicas > 0 ? dc2Nodes.get(dc).size() : 0;
}
else
- nodes = metadata.getSizeOfAllEndpoints();
-
+ nodes = allOtherNodes.size();
+
res = nodes > strat.getReplicationFactor().allReplicas;
}
-
+
return res;
}
- /**
- * Wrapper method to assemble the arguments for invoking the implementation with RangeStreamer's parameters
- */
- private EndpointsByReplica calculateRangesToFetchWithPreferredEndpoints(ReplicaCollection<?> fetchRanges, Keyspace keyspace, boolean useStrictConsistency)
+ private static EndpointsForRange endpointsForWrappedRange(Range<Token> maybeWrappingRange, PlacementForRange placements)
{
- AbstractReplicationStrategy strat = keyspace.getReplicationStrategy();
-
- TokenMetadata tmd = metadata.cloneOnlyTokenMap();
-
- TokenMetadata tmdAfter = null;
-
- if (tokens != null)
+ EndpointsForRange endpoints;
+ if (maybeWrappingRange.isWrapAround())
{
- // Pending ranges
- tmdAfter = tmd.cloneOnlyTokenMap();
- tmdAfter.updateNormalTokens(tokens, address);
+ EndpointsForRange wrappedEndpointsForRange = null;
+ for (Range<Token> r : maybeWrappingRange.unwrap())
+ {
+ EndpointsForRange ep1 = placements.forRange(r);
+ if (wrappedEndpointsForRange == null)
+ wrappedEndpointsForRange = ep1;
+ assert wrappedEndpointsForRange.endpoints().equals(ep1.endpoints()) : wrappedEndpointsForRange +" != " + ep1;
+ }
+ EndpointsForRange.Builder builder = EndpointsForRange.builder(maybeWrappingRange);
+ for (InetAddressAndPort ep : wrappedEndpointsForRange.endpoints())
+ builder.add(new Replica(ep, maybeWrappingRange, wrappedEndpointsForRange.byEndpoint().get(ep).isFull()));
+ endpoints = builder.build();
}
- else if (useStrictConsistency)
+ else
{
- throw new IllegalArgumentException("Can't ask for strict consistency and not supply tokens");
+ endpoints = placements.forRange(maybeWrappingRange);
}
-
- return calculateRangesToFetchWithPreferredEndpoints(snitch::sortedByProximity,
- strat,
- fetchRanges,
- useStrictConsistency,
- tmd,
- tmdAfter,
- keyspace.getName(),
- sourceFilters);
-
+ return endpoints;
}
/**
@@ -454,17 +477,13 @@ public class RangeStreamer
AbstractReplicationStrategy strat,
ReplicaCollection<?> fetchRanges,
boolean useStrictConsistency,
- TokenMetadata tmdBefore,
- TokenMetadata tmdAfter,
+ ClusterMetadata metadata,
String keyspace,
Collection<SourceFilter> sourceFilters)
{
- EndpointsByRange rangeAddresses = strat.getRangeAddresses(tmdBefore);
-
InetAddressAndPort localAddress = FBUtilities.getBroadcastAddressAndPort();
- logger.debug ("Keyspace: {}", keyspace);
+ logger.debug("Keyspace: {}", keyspace);
logger.debug("To fetch RN: {}", fetchRanges);
- logger.debug("Fetch ranges: {}", rangeAddresses);
Predicate<Replica> testSourceFilters = and(sourceFilters);
Function<EndpointsForRange, EndpointsForRange> sorted =
@@ -480,76 +499,72 @@ public class RangeStreamer
Predicate<Replica> isSufficient = r -> toFetch.isTransient() || r.isFull();
logger.debug("To fetch {}", toFetch);
- for (Range<Token> range : rangeAddresses.keySet())
- {
- if (!range.contains(toFetch.range()))
- continue;
-
- final EndpointsForRange oldEndpoints = sorted.apply(rangeAddresses.get(range));
-
- //Ultimately we populate this with whatever is going to be fetched from to satisfy toFetch
- //It could be multiple endpoints and we must fetch from all of them if they are there
- //With transient replication and strict consistency this is to get the full data from a full replica and
- //transient data from the transient replica losing data
- EndpointsForRange sources;
- //Due to CASSANDRA-5953 we can have a higher RF than we have endpoints.
- //So we need to be careful to only be strict when endpoints == RF
- boolean isStrictConsistencyApplicable = useStrictConsistency && (oldEndpoints.size() == strat.getReplicationFactor().allReplicas);
- if (isStrictConsistencyApplicable)
- {
- EndpointsForRange strictEndpoints;
- //Start with two sets of who replicates the range before and who replicates it after
- EndpointsForRange newEndpoints = strat.calculateNaturalReplicas(toFetch.range().right, tmdAfter);
- logger.debug("Old endpoints {}", oldEndpoints);
- logger.debug("New endpoints {}", newEndpoints);
+ ReplicationParams params = metadata.schema.getKeyspaces().get(keyspace).get().params.replication;
+ EndpointsForRange oldEndpoints = endpointsForWrappedRange(toFetch.range(), metadata.placements.get(params).reads);
+
+ //Ultimately we populate this with whatever is going to be fetched from to satisfy toFetch
+ //It could be multiple endpoints and we must fetch from all of them if they are there
+ //With transient replication and strict consistency this is to get the full data from a full replica and
+ //transient data from the transient replica losing data
+ EndpointsForRange sources;
+ //Due to CASSANDRA-5953 we can have a higher RF than we have endpoints.
+ //So we need to be careful to only be strict when endpoints == RF
+ boolean isStrictConsistencyApplicable = useStrictConsistency && (oldEndpoints.size() == strat.getReplicationFactor().allReplicas);
+ if (isStrictConsistencyApplicable)
+ {
+ EndpointsForRange strictEndpoints;
- // Remove new endpoints from old endpoints based on address
- strictEndpoints = oldEndpoints.without(newEndpoints.endpoints());
+ //Start with two sets of who replicates the range before and who replicates it after
+ EndpointsForRange newEndpoints = endpointsForWrappedRange(toFetch.range(), metadata.placements.get(params).writes);
+ logger.debug("Old endpoints {}", oldEndpoints);
+ logger.debug("New endpoints {}", newEndpoints);
- if (strictEndpoints.size() > 1)
- throw new AssertionError("Expected <= 1 endpoint but found " + strictEndpoints);
+ // Remove new endpoints from old endpoints based on address
+ strictEndpoints = oldEndpoints.without(newEndpoints.endpoints());
- //We have to check the source filters here to see if they will remove any replicas
- //required for strict consistency
- if (!all(strictEndpoints, testSourceFilters))
- throw new IllegalStateException("Necessary replicas for strict consistency were removed by source filters: " + buildErrorMessage(sourceFilters, strictEndpoints));
+ if (strictEndpoints.size() > 1)
+ throw new AssertionError("Expected <= 1 endpoint but found " + strictEndpoints);
- //If we are transitioning from transient to full and and the set of replicas for the range is not changing
- //we might end up with no endpoints to fetch from by address. In that case we can pick any full replica safely
- //since we are already a transient replica and the existing replica remains.
- //The old behavior where we might be asked to fetch ranges we don't need shouldn't occur anymore.
- //So it's an error if we don't find what we need.
- if (strictEndpoints.isEmpty() && toFetch.isTransient())
- throw new AssertionError("If there are no endpoints to fetch from then we must be transitioning from transient to full for range " + toFetch);
+ //We have to check the source filters here to see if they will remove any replicas
+ //required for strict consistency
+ if (!all(strictEndpoints, testSourceFilters))
+ throw new IllegalStateException("Necessary replicas for strict consistency were removed by source filters: " + buildErrorMessage(sourceFilters, strictEndpoints));
- if (!any(strictEndpoints, isSufficient))
- {
- // need an additional replica; include all our filters, to ensure we include a matching node
- Optional<Replica> fullReplica = Iterables.<Replica>tryFind(oldEndpoints, and(isSufficient, testSourceFilters)).toJavaUtil();
- if (fullReplica.isPresent())
- strictEndpoints = Endpoints.concat(strictEndpoints, EndpointsForRange.of(fullReplica.get()));
- else
- throw new IllegalStateException("Couldn't find any matching sufficient replica out of " + buildErrorMessage(sourceFilters, oldEndpoints));
- }
+ //If we are transitioning from transient to full and and the set of replicas for the range is not changing
+ //we might end up with no endpoints to fetch from by address. In that case we can pick any full replica safely
+ //since we are already a transient replica and the existing replica remains.
+ //The old behavior where we might be asked to fetch ranges we don't need shouldn't occur anymore.
+ //So it's an error if we don't find what we need.
+ if (strictEndpoints.isEmpty() && toFetch.isTransient())
+ throw new AssertionError("If there are no endpoints to fetch from then we must be transitioning from transient to full for range " + toFetch);
- sources = strictEndpoints;
- }
- else
+ if (!any(strictEndpoints, isSufficient))
{
- //Without strict consistency we have given up on correctness so no point in fetching from
- //a random full + transient replica since it's also likely to lose data
- //Also apply testSourceFilters that were given to us so we can safely select a single source
- sources = sorted.apply(oldEndpoints.filter(and(isSufficient, testSourceFilters)));
- //Limit it to just the first possible source, we don't need more than one and downstream
- //will fetch from every source we supply
- sources = sources.size() > 0 ? sources.subList(0, 1) : sources;
+ // need an additional replica; include all our filters, to ensure we include a matching node
+ Optional<Replica> fullReplica = Iterables.<Replica>tryFind(oldEndpoints, and(isSufficient, testSourceFilters)).toJavaUtil();
+ if (fullReplica.isPresent())
+ strictEndpoints = Endpoints.concat(strictEndpoints, EndpointsForRange.of(fullReplica.get()));
+ else
+ throw new IllegalStateException("Couldn't find any matching sufficient replica out of " + buildErrorMessage(sourceFilters, oldEndpoints));
}
- // storing range and preferred endpoint set
- rangesToFetchWithPreferredEndpoints.putAll(toFetch, sources, Conflict.NONE);
- logger.debug("Endpoints to fetch for {} are {}", toFetch, sources);
+ sources = strictEndpoints;
}
+ else
+ {
+ //Without strict consistency we have given up on correctness so no point in fetching from
+ //a random full + transient replica since it's also likely to lose data
+ //Also apply testSourceFilters that were given to us so we can safely select a single source
+ sources = sorted.apply(oldEndpoints.filter(and(isSufficient, testSourceFilters)));
+ //Limit it to just the first possible source, we don't need more than one and downstream
+ //will fetch from every source we supply
+ sources = sources.size() > 0 ? sources.subList(0, 1) : sources;
+ }
+
+ // storing range and preferred endpoint set
+ rangesToFetchWithPreferredEndpoints.putAll(toFetch, sources, Conflict.NONE);
+ logger.debug("Endpoints to fetch for {} are {}", toFetch, sources);
EndpointsForRange addressList = rangesToFetchWithPreferredEndpoints.getIfPresent(toFetch);
if (addressList == null)
@@ -669,7 +684,7 @@ public class RangeStreamer
if(entry.getKey().equals(FBUtilities.getBroadcastAddressAndPort()))
{
throw new IllegalStateException("Trying to stream locally. Range: " + entry.getValue()
- + " in keyspace " + keyspace);
+ + " in keyspace " + keyspace);
}
if (!rangesWithSources.get(entry.getValue()).endpoints().contains(entry.getKey()))
@@ -708,7 +723,7 @@ public class RangeStreamer
else
{
// Filter out already streamed ranges
- SystemKeyspace.AvailableRanges available = stateStore.getAvailableRanges(keyspace, metadata.partitioner);
+ SystemKeyspace.AvailableRanges available = stateStore.getAvailableRanges(keyspace, metadata.tokenMap.partitioner());
Predicate<FetchReplica> isAvailable = fetch -> {
boolean isInFull = available.full.contains(fetch.local.range());
@@ -758,13 +773,13 @@ public class RangeStreamer
InetAddressAndPort self = FBUtilities.getBroadcastAddressAndPort();
RangesAtEndpoint full = remaining.stream()
- .filter(pair -> pair.remote.isFull())
- .map(pair -> pair.local)
- .collect(RangesAtEndpoint.collector(self));
+ .filter(pair -> pair.remote.isFull())
+ .map(pair -> pair.local)
+ .collect(RangesAtEndpoint.collector(self));
RangesAtEndpoint transientReplicas = remaining.stream()
- .filter(pair -> pair.remote.isTransient())
- .map(pair -> pair.local)
- .collect(RangesAtEndpoint.collector(self));
+ .filter(pair -> pair.remote.isTransient())
+ .map(pair -> pair.local)
+ .collect(RangesAtEndpoint.collector(self));
logger.debug("Source and our replicas {}", fetchReplicas);
logger.debug("Source {} Keyspace {} streaming full {} transient {}", source, keyspace, full, transientReplicas);
diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java b/src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java
index 6a382c0d51..d46c043ed5 100644
--- a/src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java
+++ b/src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java
@@ -36,7 +36,7 @@ import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.SimpleSnitch;
-import org.apache.cassandra.locator.TokenMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.OutputHandler;
public class OfflineTokenAllocator
@@ -115,7 +115,6 @@ public class OfflineTokenAllocator
private static class MultinodeAllocator
{
private final FakeSnitch fakeSnitch;
- private final TokenMetadata fakeMetadata;
private final TokenAllocation allocation;
private final Map<Integer, SummaryStatistics> lastCheckPoint = Maps.newHashMap();
private final OutputHandler logger;
@@ -123,8 +122,7 @@ public class OfflineTokenAllocator
private MultinodeAllocator(int rf, int numTokens, OutputHandler logger, IPartitioner partitioner)
{
this.fakeSnitch = new FakeSnitch();
- this.fakeMetadata = new TokenMetadata(fakeSnitch).cloneWithNewPartitioner(partitioner);
- this.allocation = TokenAllocation.create(fakeSnitch, fakeMetadata, rf, numTokens);
+ this.allocation = TokenAllocation.create(fakeSnitch, new ClusterMetadata(partitioner), rf, numTokens);
this.logger = logger;
}
@@ -133,7 +131,8 @@ public class OfflineTokenAllocator
// Update snitch and token metadata info
InetAddressAndPort fakeNodeAddressAndPort = getLoopbackAddressWithPort(nodeId);
fakeSnitch.nodeByRack.put(fakeNodeAddressAndPort, rackId);
- fakeMetadata.updateTopology(fakeNodeAddressAndPort);
+ // todo:
+ //fakeMetadata.updateTopology(fakeNodeAddressAndPort);
// Allocate tokens
Collection<Token> tokens = allocation.allocate(fakeNodeAddressAndPort);
diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java
index e3e81dc116..bca8532a44 100644
--- a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java
+++ b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java
@@ -27,6 +27,7 @@ import java.util.TreeMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import com.google.common.collect.Multimap;
import org.apache.commons.math3.stat.descriptive.SummaryStatistics;
@@ -41,56 +42,55 @@ import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.locator.SimpleStrategy;
-import org.apache.cassandra.locator.TokenMetadata;
-import org.apache.cassandra.locator.TokenMetadata.Topology;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
public class TokenAllocation
{
public static final double WARN_STDEV_GROWTH = 0.05;
private static final Logger logger = LoggerFactory.getLogger(TokenAllocation.class);
- final TokenMetadata tokenMetadata;
+ final ClusterMetadata metadata;
final AbstractReplicationStrategy replicationStrategy;
final int numTokens;
final Map<String, Map<String, StrategyAdapter>> strategyByRackDc = new HashMap<>();
- private TokenAllocation(TokenMetadata tokenMetadata, AbstractReplicationStrategy replicationStrategy, int numTokens)
+ private TokenAllocation(ClusterMetadata metadata, AbstractReplicationStrategy replicationStrategy, int numTokens)
{
- this.tokenMetadata = tokenMetadata.cloneOnlyTokenMap();
+ this.metadata = metadata;
this.replicationStrategy = replicationStrategy;
this.numTokens = numTokens;
}
- public static Collection<Token> allocateTokens(final TokenMetadata tokenMetadata,
+ public static Collection<Token> allocateTokens(final ClusterMetadata metadata,
final AbstractReplicationStrategy rs,
final InetAddressAndPort endpoint,
int numTokens)
{
- return create(tokenMetadata, rs, numTokens).allocate(endpoint);
+ return create(metadata, rs, numTokens).allocate(endpoint);
}
- public static Collection<Token> allocateTokens(final TokenMetadata tokenMetadata,
+ public static Collection<Token> allocateTokens(final ClusterMetadata metadata,
final int replicas,
final InetAddressAndPort endpoint,
int numTokens)
{
- return create(DatabaseDescriptor.getEndpointSnitch(), tokenMetadata, replicas, numTokens).allocate(endpoint);
+ return create(DatabaseDescriptor.getEndpointSnitch(), metadata, replicas, numTokens).allocate(endpoint);
}
- static TokenAllocation create(IEndpointSnitch snitch, TokenMetadata tokenMetadata, int replicas, int numTokens)
+ static TokenAllocation create(IEndpointSnitch snitch, ClusterMetadata metadata, int replicas, int numTokens)
{
// We create a fake NTS replication strategy with the specified RF in the local DC
HashMap<String, String> options = new HashMap<>();
options.put(snitch.getLocalDatacenter(), Integer.toString(replicas));
- NetworkTopologyStrategy fakeReplicationStrategy = new NetworkTopologyStrategy(null, tokenMetadata, snitch, options);
+ NetworkTopologyStrategy fakeReplicationStrategy = new NetworkTopologyStrategy(null, options);
- TokenAllocation allocator = new TokenAllocation(tokenMetadata, fakeReplicationStrategy, numTokens);
- return allocator;
+ return new TokenAllocation(metadata, fakeReplicationStrategy, numTokens);
}
- static TokenAllocation create(TokenMetadata tokenMetadata, AbstractReplicationStrategy rs, int numTokens)
+ static TokenAllocation create(ClusterMetadata metadata, AbstractReplicationStrategy rs, int numTokens)
{
- return new TokenAllocation(tokenMetadata, rs, numTokens);
+ return new TokenAllocation(metadata, rs, numTokens);
}
Collection<Token> allocate(InetAddressAndPort endpoint)
@@ -100,7 +100,6 @@ public class TokenAllocation
tokens = strategy.adjustForCrossDatacenterClashes(tokens);
SummaryStatistics os = strategy.replicatedOwnershipStats();
- tokenMetadata.updateNormalTokens(tokens, endpoint);
SummaryStatistics ns = strategy.replicatedOwnershipStats();
logger.info("Selected tokens {}", tokens);
@@ -142,12 +141,14 @@ public class TokenAllocation
final TokenAllocator<InetAddressAndPort> createAllocator()
{
NavigableMap<Token, InetAddressAndPort> sortedTokens = new TreeMap<>();
- for (Map.Entry<Token, InetAddressAndPort> en : tokenMetadata.getNormalAndBootstrappingTokenToEndpointMap().entrySet())
+
+ for (Map.Entry<Token, NodeId> en : metadata.tokenMap.asMap().entrySet())
{
- if (inAllocationRing(en.getValue()))
- sortedTokens.put(en.getKey(), en.getValue());
+ InetAddressAndPort endpoint = metadata.directory.endpoint(en.getValue());
+ if (inAllocationRing(endpoint))
+ sortedTokens.put(en.getKey(), endpoint);
}
- return TokenAllocatorFactory.createTokenAllocator(sortedTokens, this, tokenMetadata.partitioner);
+ return TokenAllocatorFactory.createTokenAllocator(sortedTokens, this, metadata.tokenMap.partitioner());
}
final Collection<Token> adjustForCrossDatacenterClashes(Collection<Token> tokens)
@@ -156,9 +157,10 @@ public class TokenAllocation
for (Token t : tokens)
{
- while (tokenMetadata.getEndpoint(t) != null)
+ while (metadata.tokenMap.owner(t) != null)
{
- InetAddressAndPort other = tokenMetadata.getEndpoint(t);
+ NodeId nodeId = metadata.tokenMap.owner(t);
+ InetAddressAndPort other = metadata.directory.endpoint(nodeId);
if (inAllocationRing(other))
throw new ConfigurationException(String.format("Allocated token %s already assigned to node %s. Is another node also allocating tokens?", t, other));
t = t.increaseSlightly();
@@ -175,7 +177,10 @@ public class TokenAllocation
{
// Filter only in the same allocation ring
if (inAllocationRing(en.getKey()))
- stat.addValue(en.getValue() / tokenMetadata.getTokens(en.getKey()).size());
+ {
+ NodeId nodeId = metadata.directory.peerId(en.getKey());
+ stat.addValue(en.getValue() / metadata.tokenMap.tokens(nodeId).size());
+ }
}
return stat;
}
@@ -184,7 +189,7 @@ public class TokenAllocation
private Map<InetAddressAndPort, Double> evaluateReplicatedOwnership()
{
Map<InetAddressAndPort, Double> ownership = Maps.newHashMap();
- List<Token> sortedTokens = tokenMetadata.sortedTokens();
+ List<Token> sortedTokens = metadata.tokenMap.tokens();
if (sortedTokens.isEmpty())
return ownership;
@@ -205,7 +210,7 @@ public class TokenAllocation
{
double size = current.size(next);
Token representative = current.getPartitioner().midpoint(current, next);
- for (InetAddressAndPort n : replicationStrategy.calculateNaturalReplicas(representative, tokenMetadata).endpoints())
+ for (InetAddressAndPort n : replicationStrategy.calculateNaturalReplicas(representative, metadata).endpoints())
{
Double v = ownership.get(n);
ownership.put(n, v != null ? v + size : size);
@@ -215,8 +220,9 @@ public class TokenAllocation
private StrategyAdapter getOrCreateStrategy(InetAddressAndPort endpoint)
{
- String dc = replicationStrategy.snitch.getDatacenter(endpoint);
- String rack = replicationStrategy.snitch.getRack(endpoint);
+ IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
+ String dc = snitch.getDatacenter(endpoint);
+ String rack = snitch.getRack(endpoint);
return getOrCreateStrategy(dc, rack);
}
@@ -228,7 +234,7 @@ public class TokenAllocation
private StrategyAdapter createStrategy(String dc, String rack)
{
if (replicationStrategy instanceof NetworkTopologyStrategy)
- return createStrategy(tokenMetadata, (NetworkTopologyStrategy) replicationStrategy, dc, rack);
+ return createStrategy(metadata, (NetworkTopologyStrategy) replicationStrategy, dc, rack);
if (replicationStrategy instanceof SimpleStrategy)
return createStrategy((SimpleStrategy) replicationStrategy);
throw new ConfigurationException("Token allocation does not support replication strategy " + replicationStrategy.getClass().getSimpleName());
@@ -236,37 +242,37 @@ public class TokenAllocation
private StrategyAdapter createStrategy(final SimpleStrategy rs)
{
- return createStrategy(rs.snitch, null, null, rs.getReplicationFactor().allReplicas, false);
+ return createStrategy(DatabaseDescriptor.getEndpointSnitch(), null, null, rs.getReplicationFactor().allReplicas, false);
}
- private StrategyAdapter createStrategy(TokenMetadata tokenMetadata, NetworkTopologyStrategy strategy, String dc, String rack)
+ private StrategyAdapter createStrategy(ClusterMetadata metadata, NetworkTopologyStrategy strategy, String dc, String rack)
{
int replicas = strategy.getReplicationFactor(dc).allReplicas;
- Topology topology = tokenMetadata.getTopology();
// if topology hasn't been setup yet for this dc+rack then treat it as a separate unit
- int racks = topology.getDatacenterRacks().get(dc) != null && topology.getDatacenterRacks().get(dc).containsKey(rack)
- ? topology.getDatacenterRacks().get(dc).asMap().size()
+ Multimap<String, InetAddressAndPort> datacenterRacks = metadata.directory.datacenterRacks(dc);
+ int racks = datacenterRacks != null && datacenterRacks.containsKey(rack)
+ ? datacenterRacks.asMap().size()
: 1;
if (replicas <= 1)
{
// each node is treated as separate and replicates once
- return createStrategy(strategy.snitch, dc, null, 1, false);
+ return createStrategy(DatabaseDescriptor.getEndpointSnitch(), dc, null, 1, false);
}
else if (racks == replicas)
{
// each node is treated as separate and replicates once, with separate allocation rings for each rack
- return createStrategy(strategy.snitch, dc, rack, 1, false);
+ return createStrategy(DatabaseDescriptor.getEndpointSnitch(), dc, rack, 1, false);
}
else if (racks > replicas)
{
// group by rack
- return createStrategy(strategy.snitch, dc, null, replicas, true);
+ return createStrategy(DatabaseDescriptor.getEndpointSnitch(), dc, null, replicas, true);
}
else if (racks == 1)
{
- return createStrategy(strategy.snitch, dc, null, replicas, false);
+ return createStrategy(DatabaseDescriptor.getEndpointSnitch(), dc, null, replicas, false);
}
throw new ConfigurationException(String.format("Token allocation failed: the number of racks %d in datacenter %s is lower than its replication factor %d.",
diff --git a/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java b/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java
index 540f5bd85d..29db414be4 100644
--- a/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java
+++ b/src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java
@@ -38,6 +38,7 @@ import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.apache.cassandra.utils.concurrent.Future;
@@ -216,7 +217,7 @@ final class HintsDispatchExecutor
// The goal is to bound maximum hints traffic going towards a particular node from the rest of the cluster,
// not total outgoing hints traffic from this node. This is why the rate limiter is not shared between
// all the dispatch tasks (as there will be at most one dispatch task for a particular host id at a time).
- int nodesCount = isTransfer ? 1 : Math.max(1, StorageService.instance.getTokenMetadata().getAllEndpoints().size() - 1);
+ int nodesCount = isTransfer ? 1 : Math.max(1, ClusterMetadata.current().directory.allAddresses().size() - 1);
double throttleInBytes = DatabaseDescriptor.getHintedHandoffThrottleInKiB() * 1024.0 / nodesCount;
this.rateLimiter = RateLimiter.create(throttleInBytes == 0 ? Double.MAX_VALUE : throttleInBytes);
}
diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
index d5a54971ee..ae2278dab1 100644
--- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
+++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
@@ -22,22 +22,17 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Collections;
-import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
+import java.util.*;
import com.google.common.base.Preconditions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.WriteType;
import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.dht.RingPosition;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
@@ -46,79 +41,26 @@ import org.apache.cassandra.service.AbstractWriteResponseHandler;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.DatacenterSyncWriteResponseHandler;
import org.apache.cassandra.service.DatacenterWriteResponseHandler;
-import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.WriteResponseHandler;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.utils.FBUtilities;
-import org.cliffc.high_scale_lib.NonBlockingHashMap;
/**
* A abstract parent for all replication strategies.
*/
public abstract class AbstractReplicationStrategy
{
- private static final Logger logger = LoggerFactory.getLogger(AbstractReplicationStrategy.class);
-
public final Map<String, String> configOptions;
protected final String keyspaceName;
- private final TokenMetadata tokenMetadata;
- private final ReplicaCache<Token, EndpointsForRange> replicas = new ReplicaCache<>();
- public IEndpointSnitch snitch;
- protected AbstractReplicationStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
+ protected AbstractReplicationStrategy(String keyspaceName, Map<String, String> configOptions)
{
- assert snitch != null;
- assert tokenMetadata != null;
- this.tokenMetadata = tokenMetadata;
- this.snitch = snitch;
this.configOptions = configOptions == null ? Collections.<String, String>emptyMap() : configOptions;
this.keyspaceName = keyspaceName;
}
- public EndpointsForRange getCachedReplicas(long ringVersion, Token t)
- {
- return replicas.get(ringVersion, t);
- }
-
- /**
- * get the (possibly cached) endpoints that should store the given Token.
- * Note that while the endpoints are conceptually a Set (no duplicates will be included),
- * we return a List to avoid an extra allocation when sorting by proximity later
- * @param searchPosition the position the natural endpoints are requested for
- * @return a copy of the natural endpoints for the given token
- */
- public EndpointsForToken getNaturalReplicasForToken(RingPosition<?> searchPosition)
- {
- return getNaturalReplicas(searchPosition).forToken(searchPosition.getToken());
- }
-
- public EndpointsForRange getNaturalReplicas(RingPosition<?> searchPosition)
- {
- Token searchToken = searchPosition.getToken();
- long currentRingVersion = tokenMetadata.getRingVersion();
- Token keyToken = TokenRingUtils.firstToken(tokenMetadata.sortedTokens(), searchToken);
- EndpointsForRange endpoints = getCachedReplicas(currentRingVersion, keyToken);
- if (endpoints == null)
- {
- TokenMetadata tm = tokenMetadata.cachedOnlyTokenMap();
- // if our cache got invalidated, it's possible there is a new token to account for too
- keyToken = TokenRingUtils.firstToken(tm.sortedTokens(), searchToken);
- endpoints = calculateNaturalReplicas(searchToken, tm);
- replicas.put(tm.getRingVersion(), keyToken, endpoints);
- }
-
- return endpoints;
- }
-
- public Replica getLocalReplicaFor(RingPosition<?> searchPosition)
- {
- return getNaturalReplicas(searchPosition)
- .byEndpoint()
- .get(FBUtilities.getBroadcastAddressAndPort());
- }
-
/**
* Calculate the natural endpoints for the given token. Endpoints are returned in the order
* they occur in the ring following the searchToken, as defined by the replication strategy.
@@ -128,14 +70,12 @@ public abstract class AbstractReplicationStrategy
* {@link org.apache.cassandra.service.StorageService#getPrimaryRangesForEndpoint(String, InetAddressAndPort)}
* which is in turn relied on by various components like repair and size estimate calculations.
*
- * @see #getNaturalReplicasForToken(org.apache.cassandra.dht.RingPosition)
- *
- * @param tokenMetadata the token metadata used to find the searchToken, e.g. contains token to endpoint
+ * @param metadata the token metadata used to find the searchToken, e.g. contains token to endpoint
* mapping information
* @param searchToken the token to find the natural endpoints for
* @return a copy of the natural endpoints for the given token
*/
- public abstract EndpointsForRange calculateNaturalReplicas(Token searchToken, TokenMetadata tokenMetadata);
+ public abstract EndpointsForRange calculateNaturalReplicas(Token searchToken, ClusterMetadata metadata);
public abstract DataPlacement calculateDataPlacement(List<Range<Token>> ranges, ClusterMetadata metadata);
@@ -209,90 +149,74 @@ public abstract class AbstractReplicationStrategy
{
return getReplicationFactor().hasTransientReplicas();
}
-
/*
* NOTE: this is pretty inefficient. also the inverse (getRangeAddresses) below.
* this is fine as long as we don't use this on any critical path.
* (fixing this would probably require merging tokenmetadata into replicationstrategy,
* so we could cache/invalidate cleanly.)
*/
- public RangesByEndpoint getAddressReplicas(TokenMetadata metadata)
+ public RangesByEndpoint getAddressReplicas(ClusterMetadata metadata)
{
RangesByEndpoint.Builder map = new RangesByEndpoint.Builder();
-
- for (Token token : metadata.sortedTokens())
+ List<Token> tokens = metadata.tokenMap.tokens();
+ for (Token token : tokens)
{
- Range<Token> range = metadata.getPrimaryRangeFor(token);
- for (Replica replica : calculateNaturalReplicas(token, metadata))
+ for (Range<Token> range : TokenRingUtils.getPrimaryRangesFor(tokens, Collections.singleton(token)))
{
- // LocalStrategy always returns (min, min] ranges for it's replicas, so we skip the check here
- Preconditions.checkState(range.equals(replica.range()) || this instanceof LocalStrategy);
- map.put(replica.endpoint(), replica);
+ for (Replica replica : calculateNaturalReplicas(token, metadata))
+ {
+ // SystemStrategy always returns (min, min] ranges for it's replicas, so we skip the check here
+ Preconditions.checkState(range.equals(replica.range()) || this instanceof SystemStrategy);
+ map.put(replica.endpoint(), replica);
+ }
}
}
return map.build();
}
- public RangesAtEndpoint getAddressReplicas(TokenMetadata metadata, InetAddressAndPort endpoint)
+ public RangesAtEndpoint getAddressReplicas(ClusterMetadata metadata, InetAddressAndPort endpoint)
{
RangesAtEndpoint.Builder builder = RangesAtEndpoint.builder(endpoint);
- for (Token token : metadata.sortedTokens())
+ List<Token> tokens = metadata.tokenMap.tokens();
+ for (Token token : tokens)
{
- Range<Token> range = metadata.getPrimaryRangeFor(token);
- Replica replica = calculateNaturalReplicas(token, metadata)
- .byEndpoint().get(endpoint);
- if (replica != null)
+ for (Range<Token> range : TokenRingUtils.getPrimaryRangesFor(tokens, Collections.singleton(token)))
{
- // LocalStrategy always returns (min, min] ranges for it's replicas, so we skip the check here
- Preconditions.checkState(range.equals(replica.range()) || this instanceof LocalStrategy);
- builder.add(replica, Conflict.DUPLICATE);
+ Replica replica = calculateNaturalReplicas(token, metadata)
+ .byEndpoint().get(endpoint);
+ if (replica != null)
+ {
+ // SystemStrategy always returns (min, min] ranges for it's replicas, so we skip the check here
+ Preconditions.checkState(range.equals(replica.range()) || this instanceof SystemStrategy);
+ builder.add(replica, Conflict.DUPLICATE);
+ }
}
}
return builder.build();
}
- public EndpointsByRange getRangeAddresses(TokenMetadata metadata)
+ public EndpointsByRange getRangeAddresses(ClusterMetadata metadata)
{
EndpointsByRange.Builder map = new EndpointsByRange.Builder();
-
- for (Token token : metadata.sortedTokens())
+ List<Token> tokens = metadata.tokenMap.tokens();
+ for (Token token : tokens)
{
- Range<Token> range = metadata.getPrimaryRangeFor(token);
- for (Replica replica : calculateNaturalReplicas(token, metadata))
+ for (Range<Token> range : TokenRingUtils.getPrimaryRangesFor(tokens, Collections.singleton(token)))
{
- // LocalStrategy always returns (min, min] ranges for it's replicas, so we skip the check here
- Preconditions.checkState(range.equals(replica.range()) || this instanceof LocalStrategy);
- map.put(range, replica);
+ for (Replica replica : calculateNaturalReplicas(token, metadata))
+ {
+ // SystemStrategy always returns (min, min] ranges for it's replicas, so we skip the check here
+ Preconditions.checkState(range.equals(replica.range()) || this instanceof SystemStrategy);
+ map.put(range, replica);
+ }
}
}
return map.build();
}
- public RangesByEndpoint getAddressReplicas()
- {
- return getAddressReplicas(tokenMetadata.cloneOnlyTokenMap());
- }
-
- public RangesAtEndpoint getAddressReplicas(InetAddressAndPort endpoint)
- {
- return getAddressReplicas(tokenMetadata.cloneOnlyTokenMap(), endpoint);
- }
-
- public RangesAtEndpoint getPendingAddressRanges(TokenMetadata metadata, Token pendingToken, InetAddressAndPort pendingAddress)
- {
- return getPendingAddressRanges(metadata, Collections.singleton(pendingToken), pendingAddress);
- }
-
- public RangesAtEndpoint getPendingAddressRanges(TokenMetadata metadata, Collection<Token> pendingTokens, InetAddressAndPort pendingAddress)
- {
- TokenMetadata temp = metadata.cloneOnlyTokenMap();
- temp.updateNormalTokens(pendingTokens, pendingAddress);
- return getAddressReplicas(temp, pendingAddress);
- }
-
public abstract void validateOptions() throws ConfigurationException;
@Deprecated // use #maybeWarnOnOptions(ClientState) instead
@@ -312,7 +236,7 @@ public abstract class AbstractReplicationStrategy
* The empty collection means that no options are accepted, but null means
* that any option is accepted.
*/
- public Collection<String> recognizedOptions()
+ public Collection<String> recognizedOptions(ClusterMetadata metadata)
{
// We default to null for backward compatibility sake
return null;
@@ -320,17 +244,15 @@ public abstract class AbstractReplicationStrategy
private static AbstractReplicationStrategy createInternal(String keyspaceName,
Class<? extends AbstractReplicationStrategy> strategyClass,
- TokenMetadata tokenMetadata,
- IEndpointSnitch snitch,
Map<String, String> strategyOptions)
throws ConfigurationException
{
AbstractReplicationStrategy strategy;
- Class<?>[] parameterTypes = new Class[] {String.class, TokenMetadata.class, IEndpointSnitch.class, Map.class};
+ Class<?>[] parameterTypes = new Class[] {String.class, Map.class};
try
{
Constructor<? extends AbstractReplicationStrategy> constructor = strategyClass.getConstructor(parameterTypes);
- strategy = constructor.newInstance(keyspaceName, tokenMetadata, snitch, strategyOptions);
+ strategy = constructor.newInstance(keyspaceName, strategyOptions);
}
catch (InvocationTargetException e)
{
@@ -349,38 +271,11 @@ public abstract class AbstractReplicationStrategy
{
return createReplicationStrategy(keyspaceName, replicationParams.klass, replicationParams.options);
}
-
- public static AbstractReplicationStrategy createReplicationStrategy(String keyspaceName,
- Class<? extends AbstractReplicationStrategy> strategyClass,
- Map<String, String> strategyOptions)
- {
- AbstractReplicationStrategy strategy = createInternal(keyspaceName,
- strategyClass,
- StorageService.instance.getTokenMetadata(),
- DatabaseDescriptor.getEndpointSnitch(),
- strategyOptions);
- strategy.validateOptions();
- return strategy;
- }
-
public static AbstractReplicationStrategy createReplicationStrategy(String keyspaceName,
Class<? extends AbstractReplicationStrategy> strategyClass,
- TokenMetadata tokenMetadata,
- IEndpointSnitch snitch,
Map<String, String> strategyOptions)
{
- AbstractReplicationStrategy strategy = createInternal(keyspaceName, strategyClass, tokenMetadata, snitch, strategyOptions);
-
- // Because we used to not properly validate unrecognized options, we only log a warning if we find one.
- try
- {
- strategy.validateExpectedOptions();
- }
- catch (ConfigurationException e)
- {
- logger.warn("Ignoring {}", e.getMessage());
- }
-
+ AbstractReplicationStrategy strategy = createInternal(keyspaceName, strategyClass, strategyOptions);
strategy.validateOptions();
return strategy;
}
@@ -420,13 +315,12 @@ public abstract class AbstractReplicationStrategy
public static void validateReplicationStrategy(String keyspaceName,
Class<? extends AbstractReplicationStrategy> strategyClass,
- TokenMetadata tokenMetadata,
- IEndpointSnitch snitch,
+ ClusterMetadata metadata,
Map<String, String> strategyOptions,
ClientState state) throws ConfigurationException
{
- AbstractReplicationStrategy strategy = createInternal(keyspaceName, strategyClass, tokenMetadata, snitch, strategyOptions);
- strategy.validateExpectedOptions();
+ AbstractReplicationStrategy strategy = createInternal(keyspaceName, strategyClass, strategyOptions);
+ strategy.validateExpectedOptions(metadata);
strategy.validateOptions();
strategy.maybeWarnOnOptions(state);
if (strategy.hasTransientReplicas() && !DatabaseDescriptor.isTransientReplicationEnabled())
@@ -473,9 +367,9 @@ public abstract class AbstractReplicationStrategy
}
}
- public void validate() throws ConfigurationException
+ public void validate(ClusterMetadata snapshot) throws ConfigurationException
{
- validateExpectedOptions();
+ validateExpectedOptions(snapshot);
validateOptions();
maybeWarnOnOptions();
if (hasTransientReplicas() && !DatabaseDescriptor.isTransientReplicationEnabled())
@@ -484,76 +378,16 @@ public abstract class AbstractReplicationStrategy
}
}
- public void validateExpectedOptions() throws ConfigurationException
+ public void validateExpectedOptions(ClusterMetadata snapshot) throws ConfigurationException
{
- Collection<String> expectedOptions = recognizedOptions();
+ Collection<String> expectedOptions = recognizedOptions(snapshot);
if (expectedOptions == null)
return;
for (String key : configOptions.keySet())
{
if (!expectedOptions.contains(key))
- throw new ConfigurationException(String.format("Unrecognized strategy option {%s} passed to %s for keyspace %s", key, getClass().getSimpleName(), keyspaceName));
- }
- }
-
- static class ReplicaCache<K, V>
- {
- private final AtomicReference<ReplicaHolder<K, V>> cachedReplicas = new AtomicReference<>(new ReplicaHolder<>(0, 4));
-
- V get(long ringVersion, K keyToken)
- {
- ReplicaHolder<K, V> replicaHolder = maybeClearAndGet(ringVersion);
- if (replicaHolder == null)
- return null;
-
- return replicaHolder.replicas.get(keyToken);
- }
-
- void put(long ringVersion, K keyToken, V endpoints)
- {
- ReplicaHolder<K, V> current = maybeClearAndGet(ringVersion);
- if (current != null)
- {
- // if we have the same ringVersion, but already know about the keyToken the endpoints should be the same
- current.replicas.putIfAbsent(keyToken, endpoints);
- }
- }
-
- ReplicaHolder<K, V> maybeClearAndGet(long ringVersion)
- {
- ReplicaHolder<K, V> current = cachedReplicas.get();
- if (ringVersion == current.ringVersion)
- return current;
- else if (ringVersion < current.ringVersion) // things have already moved on
- return null;
-
- // If ring version has changed, create a fresh replica holder and try to replace the current one.
- // This may race with other threads that have the same new ring version and one will win and the loosers
- // will be garbage collected
- ReplicaHolder<K, V> cleaned = new ReplicaHolder<>(ringVersion, current.replicas.size());
- cachedReplicas.compareAndSet(current, cleaned);
-
- // A new ring version may have come along while making the new holder, so re-check the
- // reference and return the ring version if the same, otherwise return null as there is no point
- // in using it.
- current = cachedReplicas.get();
- if (ringVersion == current.ringVersion)
- return current;
- else
- return null;
- }
- }
-
- static class ReplicaHolder<K, V>
- {
- private final long ringVersion;
- private final NonBlockingHashMap<K, V> replicas;
-
- ReplicaHolder(long ringVersion, int expectedEntries)
- {
- this.ringVersion = ringVersion;
- this.replicas = new NonBlockingHashMap<>(expectedEntries);
+ throw new ConfigurationException(String.format("Unrecognized strategy option {%s} passed to %s for keyspace %s. Expected options: %s", key, getClass().getSimpleName(), keyspaceName, expectedOptions));
}
}
}
diff --git a/src/java/org/apache/cassandra/locator/EndpointsForToken.java b/src/java/org/apache/cassandra/locator/EndpointsForToken.java
index 70cd76325c..69d61bcbbf 100644
--- a/src/java/org/apache/cassandra/locator/EndpointsForToken.java
+++ b/src/java/org/apache/cassandra/locator/EndpointsForToken.java
@@ -22,8 +22,7 @@ import com.google.common.base.Preconditions;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.schema.TableMetadata;
-import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
import java.util.Arrays;
import java.util.Collection;
@@ -158,31 +157,6 @@ public class EndpointsForToken extends Endpoints<EndpointsForToken>
public static EndpointsForToken natural(Keyspace keyspace, Token token)
{
- return keyspace.getReplicationStrategy().getNaturalReplicasForToken(token);
- }
-
- public static EndpointsForToken natural(AbstractReplicationStrategy replicationStrategy, Token token)
- {
- return replicationStrategy.getNaturalReplicasForToken(token);
- }
-
- public static EndpointsForToken natural(TableMetadata table, Token token)
- {
- return natural(Keyspace.open(table.keyspace), token);
- }
-
- public static EndpointsForToken pending(TableMetadata table, Token token)
- {
- return pending(table.keyspace, token);
- }
-
- public static EndpointsForToken pending(Keyspace keyspace, Token token)
- {
- return pending(keyspace.getName(), token);
- }
-
- public static EndpointsForToken pending(String keyspace, Token token)
- {
- return StorageService.instance.getTokenMetadata().pendingEndpointsForToken(token, keyspace);
+ return ClusterMetadata.current().placements.get(keyspace.getMetadata().params.replication).reads.forToken(token);
}
}
diff --git a/src/java/org/apache/cassandra/locator/LocalStrategy.java b/src/java/org/apache/cassandra/locator/LocalStrategy.java
index 3f41ef806f..c850257dbc 100644
--- a/src/java/org/apache/cassandra/locator/LocalStrategy.java
+++ b/src/java/org/apache/cassandra/locator/LocalStrategy.java
@@ -17,15 +17,11 @@
*/
package org.apache.cassandra.locator;
-import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.dht.RingPosition;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.transformations.cms.EntireRange;
@@ -33,61 +29,27 @@ import org.apache.cassandra.tcm.transformations.cms.EntireRange;
public class LocalStrategy extends SystemStrategy
{
private static final ReplicationFactor RF = ReplicationFactor.fullOnly(1);
-// private final EndpointsForRange replicas;
- public LocalStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
+ public LocalStrategy(String keyspaceName, Map<String, String> configOptions)
{
- super(keyspaceName, tokenMetadata, snitch, configOptions);
-// replicas = EndpointsForRange.of(
-// new Replica(FBUtilities.getBroadcastAddressAndPort(),
-// DatabaseDescriptor.getPartitioner().getMinimumToken(),
-// DatabaseDescriptor.getPartitioner().getMinimumToken(),
-// true
-// )
-// );
+ super(keyspaceName, configOptions);
}
@Override
- public DataPlacement calculateDataPlacement(List<Range<Token>> ranges, ClusterMetadata metadata)
- {
- return EntireRange.placement;
- }
-
- @Override
- public ReplicationFactor getReplicationFactor()
- {
- return RF;
- }
-
- /**
- * We need to override this even if we override calculateNaturalReplicas,
- * because the default implementation depends on token calculations but
- * LocalStrategy may be used before tokens are set up.
- */
- @Override
- public EndpointsForRange getNaturalReplicas(RingPosition<?> searchPosition)
+ public EndpointsForRange calculateNaturalReplicas(Token token, ClusterMetadata metadata)
{
return EntireRange.localReplicas;
}
- public EndpointsForRange calculateNaturalReplicas(Token token, TokenMetadata metadata)
- {
- return EntireRange.localReplicas;
- }
-
- public void validateOptions() throws ConfigurationException
- {
- }
-
@Override
- public void maybeWarnOnOptions()
+ public DataPlacement calculateDataPlacement(List<Range<Token>> ranges, ClusterMetadata metadata)
{
+ return EntireRange.placement;
}
@Override
- public Collection<String> recognizedOptions()
+ public ReplicationFactor getReplicationFactor()
{
- // LocalStrategy doesn't expect any options.
- return Collections.emptySet();
+ return RF;
}
}
diff --git a/src/java/org/apache/cassandra/locator/MetaStrategy.java b/src/java/org/apache/cassandra/locator/MetaStrategy.java
index 28272088da..d1406835bf 100644
--- a/src/java/org/apache/cassandra/locator/MetaStrategy.java
+++ b/src/java/org/apache/cassandra/locator/MetaStrategy.java
@@ -19,47 +19,38 @@ package org.apache.cassandra.locator;
import java.util.List;
import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
import org.apache.cassandra.dht.Range;
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.PlacementForRange;
-import org.apache.cassandra.tcm.transformations.cms.EntireRange;
+
+import static org.apache.cassandra.tcm.transformations.cms.EntireRange.entireRange;
public class MetaStrategy extends SystemStrategy
{
- public MetaStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
+ public MetaStrategy(String keyspaceName, Map<String, String> configOptions)
{
- super(keyspaceName, tokenMetadata, snitch, configOptions);
+ super(keyspaceName, configOptions);
}
@Override
- public EndpointsForRange calculateNaturalReplicas(Token token, TokenMetadata tokenMetadata)
+ public EndpointsForRange calculateNaturalReplicas(Token token, ClusterMetadata metadata)
{
- return replicas();
+ return metadata.placements.get(ReplicationParams.meta()).reads.forRange(entireRange);
}
@Override
public DataPlacement calculateDataPlacement(List<Range<Token>> ranges, ClusterMetadata metadata)
{
- PlacementForRange placement = PlacementForRange.builder(1).withReplicaGroup(replicas()).build();
- return new DataPlacement(placement, placement);
+ return metadata.placements.get(ReplicationParams.meta());
}
- private static EndpointsForRange replicas()
- {
- Set<InetAddressAndPort> members = ClusterMetadata.current().fullCMSMembers();
- return EndpointsForRange.builder(EntireRange.entireRange, members.size())
- .addAll(members.stream().map(EntireRange::replica).collect(Collectors.toList()))
- .build();
- }
@Override
public ReplicationFactor getReplicationFactor()
{
- int rf = ClusterMetadata.current().fullCMSMembers().size();
+ int rf = ClusterMetadata.current().placements.get(ReplicationParams.meta()).writes.forRange(entireRange).byEndpoint.size();
return ReplicationFactor.fullOnly(rf);
}
diff --git a/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java b/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
index 49be984c44..4405ed6ee6 100644
--- a/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
+++ b/src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
@@ -17,39 +17,43 @@
*/
package org.apache.cassandra.locator;
-import java.util.*;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.guardrails.Guardrails;
-import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
+import com.google.common.collect.ImmutableMultimap;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Multimaps;
+import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.dht.Datacenters;
import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.ClientWarn;
-import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadata;
-import org.apache.cassandra.tcm.compatibility.AsEndpoints;
-import org.apache.cassandra.tcm.compatibility.AsLocations;
-import org.apache.cassandra.tcm.compatibility.AsTokenMap;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
+import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.PlacementForRange;
+import org.apache.cassandra.tcm.ownership.TokenMap;
import org.apache.cassandra.utils.FBUtilities;
-import com.google.common.collect.ImmutableMultimap;
-import com.google.common.collect.Multimap;
-import com.google.common.collect.Multimaps;
-import com.google.common.collect.Sets;
-
/**
* <p>
* This Replication Strategy takes a property file that gives the intended
@@ -73,9 +77,9 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
private final ReplicationFactor aggregateRf;
private static final Logger logger = LoggerFactory.getLogger(NetworkTopologyStrategy.class);
- public NetworkTopologyStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions) throws ConfigurationException
+ public NetworkTopologyStrategy(String keyspaceName, Map<String, String> configOptions) throws ConfigurationException
{
- super(keyspaceName, tokenMetadata, snitch, configOptions);
+ super(keyspaceName, configOptions);
int replicas = 0;
int trans = 0;
@@ -180,31 +184,52 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
}
}
+ @Override
+ public DataPlacement calculateDataPlacement(List<Range<Token>> ranges, ClusterMetadata metadata)
+ {
+ return calculateDataPlacement(ranges, metadata.directory, metadata.tokenMap);
+ }
+
+ private DataPlacement calculateDataPlacement(List<Range<Token>> ranges,
+ Directory directory,
+ TokenMap tokenMap)
+ {
+ PlacementForRange.Builder builder = PlacementForRange.builder();
+ for (Range<Token> range : ranges)
+ {
+ builder.withReplicaGroup(calculateNaturalReplicas(range.right,
+ range,
+ directory,
+ tokenMap,
+ datacenters));
+ }
+
+ PlacementForRange built = builder.build();
+ return new DataPlacement(built, built);
+ }
+
/**
* calculate endpoints in one pass through the tokens by tracking our progress in each DC.
*/
@Override
- public EndpointsForRange calculateNaturalReplicas(Token searchToken, TokenMetadata tokenMetadata)
+ public EndpointsForRange calculateNaturalReplicas(Token searchToken, ClusterMetadata metadata)
{
// we want to preserve insertion order so that the first added endpoint becomes primary
- ArrayList<Token> sortedTokens = tokenMetadata.sortedTokens();
- Token replicaEnd = TokenRingUtils.firstToken(sortedTokens, searchToken);
- Token replicaStart = TokenRingUtils.getPredecessor(sortedTokens, replicaEnd);
- Range<Token> replicatedRange = new Range<>(replicaStart, replicaEnd);
- return calculateNaturalReplicas(searchToken, replicatedRange, tokenMetadata, tokenMetadata, tokenMetadata, datacenters);
+ Range<Token> replicatedRange = TokenRingUtils.getRange(metadata.tokenMap.tokens(), searchToken);
+ return calculateNaturalReplicas(searchToken, replicatedRange, metadata.directory, metadata.tokenMap, datacenters);
}
- private EndpointsForRange calculateNaturalReplicas(Token searchToken,
- Range<Token> replicatedRange,
- AsEndpoints endpoints,
- AsLocations locations,
- AsTokenMap tokens,
- Map<String, ReplicationFactor> datacenters)
+ private static EndpointsForRange calculateNaturalReplicas(Token searchToken,
+ Range<Token> replicatedRange,
+ Directory directory,
+ TokenMap tokens,
+ Map<String, ReplicationFactor> datacenters)
{
EndpointsForRange.Builder builder = new EndpointsForRange.Builder(replicatedRange);
Set<Location> seenRacks = new HashSet<>();
- assert !locations.allDatacenterEndpoints().isEmpty() && !locations.allDatacenterRacks().isEmpty() : "not aware of any cluster members";
+ // Check if we have exhausted all the members/racks of a DC
+ assert !directory.allDatacenterEndpoints().isEmpty() && !directory.allDatacenterRacks().isEmpty() : "not aware of any cluster members";
int dcsToFill = 0;
Map<String, DatacenterEndpoints> dcs = new HashMap<>(datacenters.size() * 2);
@@ -214,12 +239,16 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
{
String dc = en.getKey();
ReplicationFactor rf = en.getValue();
- int nodeCount = sizeOrZero(locations.datacenterEndpoints(dc));
+ int nodeCount = sizeOrZero(directory.datacenterEndpoints(dc));
if (rf.allReplicas <= 0 || nodeCount <= 0)
continue;
- DatacenterEndpoints dcEndpoints = new DatacenterEndpoints(rf, sizeOrZero(locations.datacenterRacks(dc)), nodeCount, builder, seenRacks);
+ DatacenterEndpoints dcEndpoints = new DatacenterEndpoints(rf,
+ sizeOrZero(directory.datacenterRacks(dc)),
+ nodeCount,
+ builder,
+ seenRacks);
dcs.put(dc, dcEndpoints);
++dcsToFill;
}
@@ -229,8 +258,8 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
{
Token next = tokenIter.next();
NodeId owner = tokens.owner(next);
- InetAddressAndPort ep = endpoints.endpoint(owner);
- Location location = locations.location(owner);
+ InetAddressAndPort ep = directory.endpoint(owner);
+ Location location = directory.location(owner);
DatacenterEndpoints dcEndpoints = dcs.get(location.datacenter);
if (dcEndpoints != null && dcEndpoints.addEndpointAndCheckIfDone(ep, location, replicatedRange))
--dcsToFill;
@@ -238,30 +267,12 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
return builder.build();
}
- @Override
- public DataPlacement calculateDataPlacement(List<Range<Token>> ranges,
- ClusterMetadata metadata)
- {
- PlacementForRange.Builder builder = PlacementForRange.builder();
- for (Range<Token> range : ranges)
- {
- builder.withReplicaGroup(calculateNaturalReplicas(range.right,
- range,
- metadata.directory, // AsEndpoints
- metadata.directory, // AsLocations
- metadata.tokenMap,
- datacenters));
- }
- PlacementForRange built = builder.build();
- return new DataPlacement(built, built);
- }
-
- private int sizeOrZero(Multimap<?, ?> collection)
+ private static int sizeOrZero(Multimap<?, ?> collection)
{
return collection != null ? collection.asMap().size() : 0;
}
- private int sizeOrZero(Collection<?> collection)
+ private static int sizeOrZero(Collection<?> collection)
{
return collection != null ? collection.size() : 0;
}
@@ -284,10 +295,10 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
}
@Override
- public Collection<String> recognizedOptions()
+ public Collection<String> recognizedOptions(ClusterMetadata metadata)
{
// only valid options are valid DC names.
- return Datacenters.getValidDatacenters();
+ return Datacenters.getValidDatacenters(metadata);
}
/**
@@ -327,7 +338,7 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
if (replication != null) {
ReplicationFactor defaultReplicas = ReplicationFactor.fromString(replication);
- Datacenters.getValidDatacenters()
+ Datacenters.getValidDatacenters(ClusterMetadata.current())
.forEach(dc -> options.putIfAbsent(dc, defaultReplicas.toParseableString()));
}
@@ -335,7 +346,7 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
}
@Override
- public void validateExpectedOptions() throws ConfigurationException
+ public void validateExpectedOptions(ClusterMetadata metadata) throws ConfigurationException
{
// Do not accept query with no data centers specified.
if (this.configOptions.isEmpty())
@@ -344,11 +355,11 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
}
// Validate the data center names
- super.validateExpectedOptions();
+ super.validateExpectedOptions(metadata);
if (keyspaceName.equalsIgnoreCase(SchemaConstants.AUTH_KEYSPACE_NAME))
{
- Set<String> differenceSet = Sets.difference((Set<String>) recognizedOptions(), configOptions.keySet());
+ Set<String> differenceSet = Sets.difference((Set<String>) recognizedOptions(metadata), configOptions.keySet());
if (!differenceSet.isEmpty())
{
throw new ConfigurationException("Following datacenters have active nodes and must be present in replication options for keyspace " + SchemaConstants.AUTH_KEYSPACE_NAME + ": " + differenceSet.toString());
@@ -373,10 +384,10 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
{
if (!SchemaConstants.isSystemKeyspace(keyspaceName))
{
- ImmutableMultimap<String, InetAddressAndPort> dcsNodes = Multimaps.index(StorageService.instance.getTokenMetadata().getAllMembers(), snitch::getDatacenter);
+ ImmutableMultimap<String, InetAddressAndPort> dcsNodes = Multimaps.index(ClusterMetadata.current().directory.allAddresses(),
+ DatabaseDescriptor.getEndpointSnitch()::getDatacenter);
for (Entry<String, String> e : this.configOptions.entrySet())
{
-
String dc = e.getKey();
ReplicationFactor rf = getReplicationFactor(dc);
Guardrails.minimumReplicationFactor.guard(rf.fullReplicas, keyspaceName, false, state);
diff --git a/src/java/org/apache/cassandra/locator/PendingRangeMaps.java b/src/java/org/apache/cassandra/locator/PendingRangeMaps.java
deleted file mode 100644
index 14870c3b10..0000000000
--- a/src/java/org/apache/cassandra/locator/PendingRangeMaps.java
+++ /dev/null
@@ -1,212 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.cassandra.locator;
-
-import com.google.common.collect.Iterators;
-
-import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
-
-import java.util.*;
-
-import static org.apache.cassandra.config.CassandraRelevantProperties.LINE_SEPARATOR;
-
-public class PendingRangeMaps implements Iterable<Map.Entry<Range<Token>, EndpointsForRange.Builder>>
-{
- /**
- * We have for NavigableMap to be able to search for ranges containing a token efficiently.
- *
- * First two are for non-wrap-around ranges, and the last two are for wrap-around ranges.
- */
- // ascendingMap will sort the ranges by the ascending order of right token
- private final NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingMap;
-
- /**
- * sorting end ascending, if ends are same, sorting begin descending, so that token (end, end) will
- * come before (begin, end] with the same end, and (begin, end) will be selected in the tailMap.
- */
- private static final Comparator<Range<Token>> ascendingComparator = (o1, o2) -> {
- int res = o1.right.compareTo(o2.right);
- if (res != 0)
- return res;
-
- return o2.left.compareTo(o1.left);
- };
-
- // ascendingMap will sort the ranges by the descending order of left token
- private final NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingMap;
-
- /**
- * sorting begin descending, if begins are same, sorting end descending, so that token (begin, begin) will
- * come after (begin, end] with the same begin, and (begin, end) won't be selected in the tailMap.
- */
- private static final Comparator<Range<Token>> descendingComparator = (o1, o2) -> {
- int res = o2.left.compareTo(o1.left);
- if (res != 0)
- return res;
-
- // if left tokens are same, sort by the descending of the right tokens.
- return o2.right.compareTo(o1.right);
- };
-
- // these two maps are for warp around ranges.
- private final NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingMapForWrapAround;
-
- /**
- * for wrap around range (begin, end], which begin > end.
- * Sorting end ascending, if ends are same, sorting begin ascending,
- * so that token (end, end) will come before (begin, end] with the same end, and (begin, end] will be selected in
- * the tailMap.
- */
- private static final Comparator<Range<Token>> ascendingComparatorForWrapAround = (o1, o2) -> {
- int res = o1.right.compareTo(o2.right);
- if (res != 0)
- return res;
-
- return o1.left.compareTo(o2.left);
- };
-
- private final NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingMapForWrapAround;
-
- /**
- * for wrap around ranges, which begin > end.
- * Sorting end ascending, so that token (begin, begin) will come after (begin, end] with the same begin,
- * and (begin, end) won't be selected in the tailMap.
- */
- private static final Comparator<Range<Token>> descendingComparatorForWrapAround = (o1, o2) -> {
- int res = o2.left.compareTo(o1.left);
- if (res != 0)
- return res;
- return o1.right.compareTo(o2.right);
- };
-
- public PendingRangeMaps()
- {
- this.ascendingMap = new TreeMap<>(ascendingComparator);
- this.descendingMap = new TreeMap<>(descendingComparator);
- this.ascendingMapForWrapAround = new TreeMap<>(ascendingComparatorForWrapAround);
- this.descendingMapForWrapAround = new TreeMap<>(descendingComparatorForWrapAround);
- }
-
- static final void addToMap(Range<Token> range,
- Replica replica,
- NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingMap,
- NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingMap)
- {
- EndpointsForRange.Builder replicas = ascendingMap.get(range);
- if (replicas == null)
- {
- replicas = new EndpointsForRange.Builder(range,1);
- ascendingMap.put(range, replicas);
- descendingMap.put(range, replicas);
- }
- replicas.add(replica, Conflict.DUPLICATE);
- }
-
- public void addPendingRange(Range<Token> range, Replica replica)
- {
- if (Range.isWrapAround(range.left, range.right))
- {
- addToMap(range, replica, ascendingMapForWrapAround, descendingMapForWrapAround);
- }
- else
- {
- addToMap(range, replica, ascendingMap, descendingMap);
- }
- }
-
- static final void addIntersections(EndpointsForToken.Builder replicasToAdd,
- NavigableMap<Range<Token>, EndpointsForRange.Builder> smallerMap,
- NavigableMap<Range<Token>, EndpointsForRange.Builder> biggerMap)
- {
- // find the intersection of two sets
- for (Range<Token> range : smallerMap.keySet())
- {
- EndpointsForRange.Builder replicas = biggerMap.get(range);
- if (replicas != null)
- {
- replicasToAdd.addAll(replicas);
- }
- }
- }
-
- public EndpointsForToken pendingEndpointsFor(Token token)
- {
- EndpointsForToken.Builder replicas = EndpointsForToken.builder(token);
-
- Range<Token> searchRange = new Range<>(token, token);
-
- // search for non-wrap-around maps
- NavigableMap<Range<Token>, EndpointsForRange.Builder> ascendingTailMap = ascendingMap.tailMap(searchRange, true);
- NavigableMap<Range<Token>, EndpointsForRange.Builder> descendingTailMap = descendingMap.tailMap(searchRange, false);
-
- // add intersections of two maps
- if (ascendingTailMap.size() < descendingTailMap.size())
- {
- addIntersections(replicas, ascendingTailMap, descendingTailMap);
- }
- else
- {
- addIntersections(replicas, descendingTailMap, ascendingTailMap);
- }
-
- // search for wrap-around sets
- ascendingTailMap = ascendingMapForWrapAround.tailMap(searchRange, true);
- descendingTailMap = descendingMapForWrapAround.tailMap(searchRange, false);
-
- // add them since they are all necessary.
- for (Map.Entry<Range<Token>, EndpointsForRange.Builder> entry : ascendingTailMap.entrySet())
- {
- replicas.addAll(entry.getValue());
- }
- for (Map.Entry<Range<Token>, EndpointsForRange.Builder> entry : descendingTailMap.entrySet())
- {
- replicas.addAll(entry.getValue());
- }
-
- return replicas.build();
- }
-
- public String printPendingRanges()
- {
- StringBuilder sb = new StringBuilder();
-
- for (Map.Entry<Range<Token>, EndpointsForRange.Builder> entry : this)
- {
- Range<Token> range = entry.getKey();
-
- for (Replica replica : entry.getValue())
- {
- sb.append(replica).append(':').append(range);
- sb.append(LINE_SEPARATOR.getString());
- }
- }
-
- return sb.toString();
- }
-
- @Override
- public Iterator<Map.Entry<Range<Token>, EndpointsForRange.Builder>> iterator()
- {
- return Iterators.concat(ascendingMap.entrySet().iterator(), ascendingMapForWrapAround.entrySet().iterator());
- }
-}
diff --git a/src/java/org/apache/cassandra/locator/ReplicaLayout.java b/src/java/org/apache/cassandra/locator/ReplicaLayout.java
index 351e837a86..4d4d6fc363 100644
--- a/src/java/org/apache/cassandra/locator/ReplicaLayout.java
+++ b/src/java/org/apache/cassandra/locator/ReplicaLayout.java
@@ -25,6 +25,10 @@ import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.tcm.Epoch;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.tcm.ownership.TokenMap;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.FBUtilities;
import java.util.Set;
@@ -40,11 +44,23 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
private final E natural;
// the snapshot of the replication strategy that corresponds to the replica layout
private final AbstractReplicationStrategy replicationStrategy;
+ private final Epoch epoch;
ReplicaLayout(AbstractReplicationStrategy replicationStrategy, E natural)
+ {
+ this(replicationStrategy, natural, Epoch.EMPTY);
+ }
+
+ ReplicaLayout(AbstractReplicationStrategy replicationStrategy, E natural, Epoch epoch)
{
this.replicationStrategy = replicationStrategy;
this.natural = natural;
+ this.epoch = epoch;
+ }
+
+ public Epoch epoch()
+ {
+ return epoch;
}
/**
@@ -83,6 +99,11 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
super(replicationStrategy, natural);
}
+ public ForTokenRead(AbstractReplicationStrategy replicationStrategy, EndpointsForToken natural, Epoch epoch)
+ {
+ super(replicationStrategy, natural, epoch);
+ }
+
@Override
public Token token()
{
@@ -201,13 +222,40 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
* only responsibility is to fetch the 'natural' and 'pending' replicas, then resolve any conflicts
* {@link ReplicaLayout#haveWriteConflicts(Endpoints, Endpoints)}
*/
- public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(Keyspace keyspace, Token token)
+ public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(Keyspace ks, Token token)
{
- // TODO: these should be cached, not the natural replicas
- // TODO: race condition to fetch these. implications??
- AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy();
- EndpointsForToken natural = EndpointsForToken.natural(replicationStrategy, token);
- EndpointsForToken pending = EndpointsForToken.pending(keyspace, token);
+ return forTokenWriteLiveAndDown(ks.getMetadata(), token);
+ }
+
+ public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(KeyspaceMetadata ks, Token token)
+ {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ return forTokenWriteLiveAndDown(metadata, ks, token);
+ }
+
+ // TODO: cleanup/remove Keyspace overloads
+ public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(ClusterMetadata metadata, Keyspace ks, Token token)
+ {
+ return forTokenWriteLiveAndDown(metadata, ks.getMetadata(), token);
+ }
+
+ public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(ClusterMetadata metadata, KeyspaceMetadata ks, Token token)
+ {
+ AbstractReplicationStrategy replicationStrategy = ks.replicationStrategy;
+ EndpointsForToken natural;
+ EndpointsForToken pending;
+ if (ks.params.replication.isLocal())
+ {
+ natural = forLocalStrategyToken(metadata, replicationStrategy, token);
+ pending = EndpointsForToken.empty(token);
+ }
+ else
+ {
+ // todo deduplicate so that "pending" contains "read - write",
+ // which is a hack until we revisit how consistency level handles pending
+ natural = forNonLocalStrategyTokenRead(metadata, ks, token);
+ pending = forNonLocalStrategyTokenWrite(metadata, ks, token).without(natural.endpoints());
+ }
return forTokenWrite(replicationStrategy, natural, pending);
}
@@ -325,12 +373,14 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
* @return the read layout for a token - this includes only live natural replicas, i.e. those that are not pending
* and not marked down by the failure detector. these are reverse sorted by the badness score of the configured snitch
*/
- public static ReplicaLayout.ForTokenRead forTokenReadLiveSorted(AbstractReplicationStrategy replicationStrategy, Token token)
+ static ReplicaLayout.ForTokenRead forTokenReadLiveSorted(ClusterMetadata metadata, Keyspace keyspace, AbstractReplicationStrategy replicationStrategy, Token token)
{
- EndpointsForToken replicas = replicationStrategy.getNaturalReplicasForToken(token);
+ EndpointsForToken replicas = keyspace.getMetadata().params.replication.isLocal()
+ ? forLocalStrategyToken(metadata, replicationStrategy, token)
+ : forNonLocalStrategyTokenRead(metadata, keyspace.getMetadata(), token);
replicas = DatabaseDescriptor.getEndpointSnitch().sortedByProximity(FBUtilities.getBroadcastAddressAndPort(), replicas);
replicas = replicas.filter(FailureDetector.isReplicaAlive);
- return new ReplicaLayout.ForTokenRead(replicationStrategy, replicas);
+ return new ReplicaLayout.ForTokenRead(replicationStrategy, replicas, metadata.epoch);
}
/**
@@ -338,12 +388,41 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
* @return the read layout for a range - this includes only live natural replicas, i.e. those that are not pending
* and not marked down by the failure detector. these are reverse sorted by the badness score of the configured snitch
*/
- static ReplicaLayout.ForRangeRead forRangeReadLiveSorted(AbstractReplicationStrategy replicationStrategy, AbstractBounds<PartitionPosition> range)
+ static ReplicaLayout.ForRangeRead forRangeReadLiveSorted(ClusterMetadata metadata, Keyspace keyspace, AbstractReplicationStrategy replicationStrategy, AbstractBounds<PartitionPosition> range)
{
- EndpointsForRange replicas = replicationStrategy.getNaturalReplicas(range.right);
+ EndpointsForRange replicas = keyspace.getMetadata().params.replication.isLocal()
+ ? forLocalStrategyRange(metadata, replicationStrategy, range)
+ : forNonLocalStategyRangeRead(metadata, keyspace.getMetadata(), range);
+
replicas = DatabaseDescriptor.getEndpointSnitch().sortedByProximity(FBUtilities.getBroadcastAddressAndPort(), replicas);
replicas = replicas.filter(FailureDetector.isReplicaAlive);
return new ReplicaLayout.ForRangeRead(replicationStrategy, range, replicas);
}
+ static EndpointsForRange forNonLocalStategyRangeRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, AbstractBounds<PartitionPosition> range)
+ {
+ TokenMap tokenMap = metadata.tokenMap;
+ Token token = tokenMap.nextToken(tokenMap.tokens(), range.right.getToken());
+ return metadata.placements.get(keyspace.params.replication).reads.forRange(token);
+ }
+
+ static EndpointsForToken forNonLocalStrategyTokenRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, Token token)
+ {
+ return metadata.placements.get(keyspace.params.replication).reads.forToken(token);
+ }
+
+ static EndpointsForToken forNonLocalStrategyTokenWrite(ClusterMetadata metadata, KeyspaceMetadata keyspace, Token token)
+ {
+ return metadata.placements.get(keyspace.params.replication).writes.forToken(token);
+ }
+
+ static EndpointsForRange forLocalStrategyRange(ClusterMetadata metadata, AbstractReplicationStrategy replicationStrategy, AbstractBounds<PartitionPosition> range)
+ {
+ return replicationStrategy.calculateNaturalReplicas(range.right.getToken(), metadata);
+ }
+
+ static EndpointsForToken forLocalStrategyToken(ClusterMetadata metadata, AbstractReplicationStrategy replicationStrategy, Token t)
+ {
+ return replicationStrategy.calculateNaturalReplicas(t, metadata).forToken(t);
+ }
}
diff --git a/src/java/org/apache/cassandra/locator/SimpleStrategy.java b/src/java/org/apache/cassandra/locator/SimpleStrategy.java
index 1384fa8f9c..eb41f6b116 100644
--- a/src/java/org/apache/cassandra/locator/SimpleStrategy.java
+++ b/src/java/org/apache/cassandra/locator/SimpleStrategy.java
@@ -17,12 +17,7 @@
*/
package org.apache.cassandra.locator;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,10 +32,10 @@ import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadata;
-import org.apache.cassandra.tcm.compatibility.AsEndpoints;
-import org.apache.cassandra.tcm.compatibility.AsTokenMap;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
+import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.ownership.TokenMap;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.PlacementForRange;
@@ -57,13 +52,14 @@ public class SimpleStrategy extends AbstractReplicationStrategy
private static final Logger logger = LoggerFactory.getLogger(SimpleStrategy.class);
private final ReplicationFactor rf;
- public SimpleStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
+ public SimpleStrategy(String keyspaceName, Map<String, String> configOptions)
{
- super(keyspaceName, tokenMetadata, snitch, configOptions);
+ super(keyspaceName, configOptions);
validateOptionsInternal(configOptions);
this.rf = ReplicationFactor.fromString(this.configOptions.get(REPLICATION_FACTOR));
}
+
@Override
public DataPlacement calculateDataPlacement(List<Range<Token>> ranges, ClusterMetadata metadata)
{
@@ -76,20 +72,21 @@ public class SimpleStrategy extends AbstractReplicationStrategy
}
@Override
- public EndpointsForRange calculateNaturalReplicas(Token token, TokenMetadata metadata)
+ public EndpointsForRange calculateNaturalReplicas(Token token, ClusterMetadata metadata)
{
- ArrayList<Token> ring = metadata.sortedTokens();
- Token replicaEnd = TokenRingUtils.firstToken(ring, token);
- Token replicaStart = TokenRingUtils.getPredecessor(ring, replicaEnd);
- Range<Token> replicaRange = new Range<>(replicaStart, replicaEnd);
- return calculateNaturalReplicas(token, ring, replicaRange, metadata, metadata);
+ List<Token> ring = metadata.tokenMap.tokens();
+ if (ring.isEmpty())
+ return EndpointsForRange.empty(new Range<>(metadata.tokenMap.partitioner().getMinimumToken(), metadata.tokenMap.partitioner().getMinimumToken()));
+
+ Range<Token> replicaRange = TokenRingUtils.getRange(ring, token);
+ return calculateNaturalReplicas(token, ring, replicaRange, metadata.directory, metadata.tokenMap);
}
private EndpointsForRange calculateNaturalReplicas(Token token,
List<Token> ring,
Range<Token> replicaRange,
- AsEndpoints endpoints,
- AsTokenMap tokens)
+ Directory endpoints,
+ TokenMap tokens)
{
if (ring.isEmpty())
return EndpointsForRange.empty(new Range<>(tokens.partitioner().getMinimumToken(), token.getPartitioner().getMinimumToken()));
@@ -111,6 +108,7 @@ public class SimpleStrategy extends AbstractReplicationStrategy
return replicas.build();
}
+
@Override
public ReplicationFactor getReplicationFactor()
{
@@ -153,7 +151,7 @@ public class SimpleStrategy extends AbstractReplicationStrategy
}
@Override
- public Collection<String> recognizedOptions()
+ public Collection<String> recognizedOptions(ClusterMetadata metadata)
{
return Collections.singleton(REPLICATION_FACTOR);
}
diff --git a/src/java/org/apache/cassandra/locator/SystemStrategy.java b/src/java/org/apache/cassandra/locator/SystemStrategy.java
index 062d9b7003..fa8ebf0c66 100644
--- a/src/java/org/apache/cassandra/locator/SystemStrategy.java
+++ b/src/java/org/apache/cassandra/locator/SystemStrategy.java
@@ -23,15 +23,16 @@ import java.util.Collections;
import java.util.Map;
import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.tcm.ClusterMetadata;
/**
* A replication strategy that is uniform for all tokens and ignores TokenMetadata
*/
public abstract class SystemStrategy extends AbstractReplicationStrategy
{
- public SystemStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
+ public SystemStrategy(String keyspaceName, Map<String, String> configOptions)
{
- super(keyspaceName, tokenMetadata, snitch, configOptions);
+ super(keyspaceName, configOptions);
}
public void validateOptions() throws ConfigurationException
@@ -44,7 +45,7 @@ public abstract class SystemStrategy extends AbstractReplicationStrategy
}
@Override
- public Collection<String> recognizedOptions()
+ public Collection<String> recognizedOptions(ClusterMetadata metadata)
{
// SystemStrategy doesn't expect any options.
return Collections.emptySet();
diff --git a/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java b/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
index d2d63e9093..21f7e35df3 100644
--- a/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
+++ b/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
@@ -324,7 +324,7 @@ public final class KeyspaceMetadata implements SchemaElement
name));
}
- params.validate(name, null);
+ params.validate(name, null, metadata);
tablesAndViews().forEach(TableMetadata::validate);
Set<String> indexNames = new HashSet<>();
diff --git a/src/java/org/apache/cassandra/schema/KeyspaceParams.java b/src/java/org/apache/cassandra/schema/KeyspaceParams.java
index d960a709ad..9cfaaa17f6 100644
--- a/src/java/org/apache/cassandra/schema/KeyspaceParams.java
+++ b/src/java/org/apache/cassandra/schema/KeyspaceParams.java
@@ -24,10 +24,11 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.base.Objects;
+import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
-import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
@@ -99,9 +100,9 @@ public final class KeyspaceParams
return new KeyspaceParams(true, ReplicationParams.nts(args));
}
- public void validate(String name, ClientState state)
+ public void validate(String name, ClientState state, ClusterMetadata metadata)
{
- replication.validate(name, state);
+ replication.validate(name, state, metadata);
}
@Override
diff --git a/src/java/org/apache/cassandra/schema/PartitionDenylist.java b/src/java/org/apache/cassandra/schema/PartitionDenylist.java
index 53fcb5e2b1..24bff57fa2 100644
--- a/src/java/org/apache/cassandra/schema/PartitionDenylist.java
+++ b/src/java/org/apache/cassandra/schema/PartitionDenylist.java
@@ -44,8 +44,8 @@ import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RequestExecutionException;
-import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.reads.range.RangeCommands;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.NoSpamLogger;
@@ -433,13 +433,13 @@ public class PartitionDenylist
final Set<ByteBuffer> keys = new HashSet<>();
final NavigableSet<Token> tokens = new TreeSet<>();
-
+ ClusterMetadata metadata = ClusterMetadata.current();
int processed = 0;
for (final UntypedResultSet.Row row : results)
{
final ByteBuffer key = row.getBlob("key");
keys.add(key);
- tokens.add(StorageService.instance.getTokenMetadata().partitioner.getToken(key));
+ tokens.add(metadata.tokenMap.partitioner().getToken(key));
processed++;
if (processed >= limit)
diff --git a/src/java/org/apache/cassandra/schema/ReplicationParams.java b/src/java/org/apache/cassandra/schema/ReplicationParams.java
index a24b1d3813..dbe3e5ba24 100644
--- a/src/java/org/apache/cassandra/schema/ReplicationParams.java
+++ b/src/java/org/apache/cassandra/schema/ReplicationParams.java
@@ -26,20 +26,17 @@ import com.google.common.base.MoreObjects;
import com.google.common.base.Objects;
import com.google.common.collect.ImmutableMap;
-import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CqlBuilder;
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.AbstractReplicationStrategy;
-import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.locator.MetaStrategy;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.locator.SimpleStrategy;
-import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.service.ClientState;
-import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.FBUtilities;
@@ -107,12 +104,10 @@ public final class ReplicationParams
return new ReplicationParams(NetworkTopologyStrategy.class, options);
}
- public void validate(String name, ClientState state)
+ public void validate(String name, ClientState state, ClusterMetadata metadata)
{
// Attempt to instantiate the ARS, which will throw a ConfigurationException if the options aren't valid.
- TokenMetadata tmd = StorageService.instance.getTokenMetadata();
- IEndpointSnitch eps = DatabaseDescriptor.getEndpointSnitch();
- AbstractReplicationStrategy.validateReplicationStrategy(name, klass, tmd, eps, options, state);
+ AbstractReplicationStrategy.validateReplicationStrategy(name, klass, metadata, options, state);
}
public static ReplicationParams fromMap(Map<String, String> map) {
diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java
index eaba5d944e..a25d71e490 100644
--- a/src/java/org/apache/cassandra/service/ActiveRepairService.java
+++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java
@@ -19,7 +19,15 @@ package org.apache.cassandra.service;
import java.io.IOException;
import java.net.UnknownHostException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledFuture;
@@ -27,10 +35,10 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
-import javax.management.openmbean.CompositeData;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import java.util.stream.Collectors;
+import javax.management.openmbean.CompositeData;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
@@ -40,35 +48,20 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
-
-import org.apache.cassandra.concurrent.ExecutorPlus;
-import org.apache.cassandra.config.Config;
-import org.apache.cassandra.config.DurationSpec;
-import org.apache.cassandra.db.compaction.CompactionManager;
-import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.locator.AbstractReplicationStrategy;
-import org.apache.cassandra.locator.EndpointsByRange;
-import org.apache.cassandra.locator.EndpointsForRange;
-import org.apache.cassandra.utils.ExecutorUtils;
-import org.apache.cassandra.repair.state.CoordinatorState;
-import org.apache.cassandra.repair.state.ParticipateState;
-import org.apache.cassandra.repair.state.ValidationState;
-import org.apache.cassandra.utils.Simulate;
-import org.apache.cassandra.locator.EndpointsForToken;
-import org.apache.cassandra.schema.Schema;
-import org.apache.cassandra.schema.TableMetadata;
-import org.apache.cassandra.streaming.PreviewKind;
-import org.apache.cassandra.utils.TimeUUID;
-import org.apache.cassandra.utils.concurrent.CountDownLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.concurrent.ScheduledExecutors;
+import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
@@ -78,26 +71,25 @@ import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
import org.apache.cassandra.gms.IFailureDetectionEventListener;
import org.apache.cassandra.gms.IFailureDetector;
import org.apache.cassandra.gms.VersionedValue;
+import org.apache.cassandra.locator.EndpointsByRange;
+import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.metrics.RepairMetrics;
-import org.apache.cassandra.net.RequestCallback;
-import org.apache.cassandra.net.Verb;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.RequestCallback;
+import org.apache.cassandra.net.Verb;
import org.apache.cassandra.repair.CommonRange;
import org.apache.cassandra.repair.NoSuchRepairSessionException;
-import org.apache.cassandra.service.paxos.PaxosRepair;
-import org.apache.cassandra.service.paxos.cleanup.PaxosCleanup;
import org.apache.cassandra.repair.RepairJobDesc;
import org.apache.cassandra.repair.RepairParallelism;
import org.apache.cassandra.repair.RepairSession;
import org.apache.cassandra.repair.consistent.CoordinatorSessions;
import org.apache.cassandra.repair.consistent.LocalSessions;
+import org.apache.cassandra.repair.consistent.RepairedState;
import org.apache.cassandra.repair.consistent.admin.CleanupSummary;
import org.apache.cassandra.repair.consistent.admin.PendingStats;
import org.apache.cassandra.repair.consistent.admin.RepairStats;
-import org.apache.cassandra.repair.consistent.RepairedState;
import org.apache.cassandra.repair.consistent.admin.SchemaArgsParser;
import org.apache.cassandra.repair.messages.CleanupMessage;
import org.apache.cassandra.repair.messages.PrepareMessage;
@@ -105,11 +97,23 @@ import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.repair.messages.SyncResponse;
import org.apache.cassandra.repair.messages.ValidationResponse;
+import org.apache.cassandra.repair.state.CoordinatorState;
+import org.apache.cassandra.repair.state.ParticipateState;
+import org.apache.cassandra.repair.state.ValidationState;
import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.paxos.PaxosRepair;
+import org.apache.cassandra.service.paxos.cleanup.PaxosCleanup;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.streaming.PreviewKind;
+import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.utils.Simulate;
+import org.apache.cassandra.utils.TimeUUID;
+import org.apache.cassandra.utils.concurrent.CountDownLatch;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.FutureCombiner;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
@@ -120,12 +124,17 @@ import static java.util.Collections.synchronizedSet;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.config.Config.RepairCommandPoolFullStrategy.reject;
-import static org.apache.cassandra.config.DatabaseDescriptor.*;
+import static org.apache.cassandra.config.DatabaseDescriptor.getRepairCommandPoolFullStrategy;
+import static org.apache.cassandra.config.DatabaseDescriptor.getRepairCommandPoolSize;
+import static org.apache.cassandra.config.DatabaseDescriptor.getRepairStateExpires;
+import static org.apache.cassandra.config.DatabaseDescriptor.getRepairStateSize;
+import static org.apache.cassandra.config.DatabaseDescriptor.getRpcTimeout;
+import static org.apache.cassandra.config.DatabaseDescriptor.paxosRepairEnabled;
import static org.apache.cassandra.net.Message.out;
import static org.apache.cassandra.net.Verb.PREPARE_MSG;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
-import static org.apache.cassandra.utils.Simulate.With.MONITORS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch;
/**
@@ -535,10 +544,10 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
EndpointsForRange neighbors = replicaSets.get(rangeSuperSet).withoutSelf();
+ ClusterMetadata metadata = ClusterMetadata.current();
if (dataCenters != null && !dataCenters.isEmpty())
{
- TokenMetadata.Topology topology = ss.getTokenMetadata().cloneOnlyTokenMap().getTopology();
- Multimap<String, InetAddressAndPort> dcEndpointsMap = topology.getDatacenterEndpoints();
+ Multimap<String, InetAddressAndPort> dcEndpointsMap = metadata.directory.allDatacenterEndpoints();
Iterable<InetAddressAndPort> dcEndpoints = concat(transform(dataCenters, dcEndpointsMap::get));
return neighbors.select(dcEndpoints, true);
}
@@ -765,9 +774,9 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
}
/**
- * We assume when calling this method that a parent session for the provided identifier
- * exists, and that session is still in progress. When it doesn't, that should mean either
- * {@link #abort(Predicate, String)} or {@link #failRepair(TimeUUID, String)} have removed it.
+ * We assume when calling this method that a parent session for the provided identifier
+ * exists, and that session is still in progress. When it doesn't, that should mean either
+ * {@link #abort(Predicate, String)} or failRepair(UUID, String) have removed it.
*
* @param parentSessionId an identifier for an active parent repair session
* @return the {@link ParentRepairSession} associated with the provided identifier
@@ -1038,18 +1047,27 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
logger.warn("Not running paxos repair for topology change because there are no ranges to repair");
return ImmediateFuture.success(null);
}
- List<TableMetadata> tables = Lists.newArrayList(Schema.instance.getKeyspaceMetadata(ksName).tables);
+ ClusterMetadata metadata = ClusterMetadata.current();
+ List<TableMetadata> tables = Lists.newArrayList(metadata.schema.getKeyspaces().getNullable(ksName).tables);
List<Future<Void>> futures = new ArrayList<>(ranges.size() * tables.size());
Keyspace keyspace = Keyspace.open(ksName);
- AbstractReplicationStrategy replication = keyspace.getReplicationStrategy();
+
for (Range<Token> range: ranges)
{
for (TableMetadata table : tables)
{
- Set<InetAddressAndPort> endpoints = replication.getNaturalReplicas(range.right).filter(FailureDetector.isReplicaAlive).endpoints();
+ Set<InetAddressAndPort> endpoints = metadata.placements
+ .get(keyspace.getMetadata().params.replication)
+ .reads
+ .forRange(range)
+ .filter(FailureDetector.isReplicaAlive).endpoints();
if (!PaxosRepair.hasSufficientLiveNodesForTopologyChange(keyspace, range, endpoints))
{
- Set<InetAddressAndPort> downEndpoints = replication.getNaturalReplicas(range.right).filter(e -> !endpoints.contains(e)).endpoints();
+ Set<InetAddressAndPort> downEndpoints = metadata.placements
+ .get(keyspace.getMetadata().params.replication)
+ .reads
+ .forRange(range)
+ .filter(e -> !endpoints.contains(e.endpoint())).endpoints();
downEndpoints.removeAll(endpoints);
throw new RuntimeException(String.format("Insufficient live nodes to repair paxos for %s in %s for %s.\n" +
@@ -1064,12 +1082,12 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
"Skipping this check can lead to paxos correctness issues",
range, ksName, reason, downEndpoints));
}
- EndpointsForToken pending = StorageService.instance.getTokenMetadata().pendingEndpointsForToken(range.right, ksName);
- if (pending.size() > 1 && !Boolean.getBoolean("cassandra.paxos_repair_allow_multiple_pending_unsafe"))
+ // todo: can probably be removed with TrM
+ if (ClusterMetadata.current().hasPendingRangesFor(keyspace.getMetadata(), range.right) && !Boolean.getBoolean("cassandra.paxos_repair_allow_multiple_pending_unsafe"))
{
- throw new RuntimeException(String.format("Cannot begin paxos auto repair for %s in %s.%s, multiple pending endpoints exist for range (%s). " +
+ throw new RuntimeException(String.format("Cannot begin paxos auto repair for %s in %s.%s, multiple pending endpoints exist for range (metadata = %s). " +
"Set -Dcassandra.paxos_repair_allow_multiple_pending_unsafe=true to skip this check",
- range, table.keyspace, table.name, pending));
+ range, table.keyspace, table.name, ClusterMetadata.current()));
}
Future<Void> future = PaxosCleanup.cleanup(endpoints, table, Collections.singleton(range), false, repairCommandExecutor());
@@ -1146,4 +1164,4 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
}
return null;
}
-}
+}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 5ea65c6915..4aea5d47aa 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -81,6 +81,8 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.service.paxos.PaxosState;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.InProgressSequence;
import org.apache.cassandra.tcm.Startup;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JMXServerUtils;
@@ -294,7 +296,7 @@ public class CassandraDaemon
SystemKeyspaceMigrator41.migrate();
// Populate token metadata before flushing, for token-aware sstable partitioning (#6696)
- StorageService.instance.populateTokenMetadata();
+// StorageService.instance.populateTokenMetadata();
// try
// {
@@ -371,7 +373,7 @@ public class CassandraDaemon
}
// Re-populate token metadata after commit log recover (new peers might be loaded onto system keyspace #10293)
- StorageService.instance.populateTokenMetadata();
+// StorageService.instance.populateTokenMetadata();
try
{
@@ -838,26 +840,31 @@ public class CassandraDaemon
public void validateTransportsCanStart()
{
+ ClusterMetadata metadata = ClusterMetadata.current();
+ InProgressSequence startupSequence = metadata.inProgressSequences.get(metadata.myNodeId());
+
// We only start transports if bootstrap has completed and we're not in survey mode, OR if we are in
// survey mode and streaming has completed but we're not using auth.
// OR if we have not joined the ring yet.
- if (StorageService.instance.hasJoined())
+ if (startupSequence != null)
+ {
+ throw new IllegalStateException("Not starting client transports because startup sequence has not completed");
+ }
+
+ if (StorageService.instance.isSurveyMode())
{
- if (StorageService.instance.isSurveyMode())
+ if (DatabaseDescriptor.getAuthenticator().requireAuthentication())
{
- if (StorageService.instance.isBootstrapMode() || DatabaseDescriptor.getAuthenticator().requireAuthentication())
- {
- throw new IllegalStateException("Not starting client transports in write_survey mode as it's bootstrapping or " +
- "auth is enabled");
- }
+ throw new IllegalStateException("Not starting client transports in write_survey mode as it's not fully started yet " +
+ "auth is enabled");
}
- else
+ }
+ else
+ {
+ if (!SystemKeyspace.bootstrapComplete())
{
- if (!SystemKeyspace.bootstrapComplete())
- {
- throw new IllegalStateException("Node is not yet bootstrapped completely. Use nodetool to check bootstrap" +
- " state and resume. For more, see `nodetool help bootstrap`");
- }
+ throw new IllegalStateException("Node is not yet bootstrapped completely. Use nodetool to check bootstrap" +
+ " state and resume. For more, see `nodetool help bootstrap`");
}
}
}
diff --git a/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java b/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java
deleted file mode 100644
index 837bdd882e..0000000000
--- a/src/java/org/apache/cassandra/service/PendingRangeCalculatorService.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service;
-
-import java.util.Collection;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import com.google.common.annotations.VisibleForTesting;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.concurrent.SequentialExecutorPlus;
-import org.apache.cassandra.concurrent.SequentialExecutorPlus.AtLeastOnceTrigger;
-import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.locator.AbstractReplicationStrategy;
-import org.apache.cassandra.schema.Schema;
-import org.apache.cassandra.utils.ExecutorUtils;
-
-import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
-import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
-
-public class PendingRangeCalculatorService
-{
- public static final PendingRangeCalculatorService instance = new PendingRangeCalculatorService();
-
- private static final Logger logger = LoggerFactory.getLogger(PendingRangeCalculatorService.class);
-
- // the executor will only run a single range calculation at a time while keeping at most one task queued in order
- // to trigger an update only after the most recent state change and not for each update individually
- private final SequentialExecutorPlus executor = executorFactory()
- .withJmxInternal()
- .configureSequential("PendingRangeCalculator")
- .withRejectedExecutionHandler((r, e) -> {}) // silently handle rejected tasks, this::update takes care of bookkeeping
- .build();
-
- private final AtLeastOnceTrigger update = executor.atLeastOnceTrigger(() -> {
- PendingRangeCalculatorServiceDiagnostics.taskStarted(1);
- long start = currentTimeMillis();
- Collection<String> keyspaces = Schema.instance.getNonLocalStrategyKeyspaces().names();
- for (String keyspaceName : keyspaces)
- calculatePendingRanges(Keyspace.open(keyspaceName).getReplicationStrategy(), keyspaceName);
- if (logger.isTraceEnabled())
- logger.trace("Finished PendingRangeTask for {} keyspaces in {}ms", keyspaces.size(), currentTimeMillis() - start);
- PendingRangeCalculatorServiceDiagnostics.taskFinished();
- });
-
- public PendingRangeCalculatorService()
- {
- }
-
- public void update()
- {
- boolean success = update.trigger();
- if (!success) PendingRangeCalculatorServiceDiagnostics.taskRejected(1);
- else PendingRangeCalculatorServiceDiagnostics.taskCountChanged(1);
- }
-
- public void blockUntilFinished()
- {
- update.sync();
- }
-
-
- public void executeWhenFinished(Runnable runnable)
- {
- update.runAfter(runnable);
- }
-
- // public & static for testing purposes
- public static void calculatePendingRanges(AbstractReplicationStrategy strategy, String keyspaceName)
- {
- StorageService.instance.getTokenMetadata().calculatePendingRanges(strategy, keyspaceName);
- }
-
- @VisibleForTesting
- public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
- {
- ExecutorUtils.shutdownNowAndWait(timeout, unit, executor);
- }
-
-}
diff --git a/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceDiagnostics.java b/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceDiagnostics.java
deleted file mode 100644
index 2b677d1fe6..0000000000
--- a/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceDiagnostics.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service;
-
-import org.apache.cassandra.diag.DiagnosticEventService;
-import org.apache.cassandra.service.PendingRangeCalculatorServiceEvent.PendingRangeCalculatorServiceEventType;
-
-/**
- * Utility methods for diagnostic events related to {@link PendingRangeCalculatorService}.
- */
-final class PendingRangeCalculatorServiceDiagnostics
-{
- private static final DiagnosticEventService service = DiagnosticEventService.instance();
-
- private PendingRangeCalculatorServiceDiagnostics()
- {
- }
-
- static void taskStarted(int taskCount)
- {
- if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_STARTED))
- service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_STARTED,
- taskCount));
- }
-
- static void taskFinished()
- {
- if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_FINISHED_SUCCESSFULLY))
- service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_FINISHED_SUCCESSFULLY));
- }
-
- static void taskRejected(int taskCount)
- {
- if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_EXECUTION_REJECTED))
- service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_EXECUTION_REJECTED,
- taskCount));
- }
-
- static void taskCountChanged(int taskCount)
- {
- if (isEnabled(PendingRangeCalculatorServiceEventType.TASK_COUNT_CHANGED))
- service.publish(new PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType.TASK_COUNT_CHANGED,
- taskCount));
- }
-
- private static boolean isEnabled(PendingRangeCalculatorServiceEventType type)
- {
- return service.isEnabled(PendingRangeCalculatorServiceEvent.class, type);
- }
-}
diff --git a/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceEvent.java b/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceEvent.java
deleted file mode 100644
index f25517347c..0000000000
--- a/src/java/org/apache/cassandra/service/PendingRangeCalculatorServiceEvent.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service;
-
-import java.io.Serializable;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.cassandra.diag.DiagnosticEvent;
-
-/**
- * Events related to {@link PendingRangeCalculatorService}.
- */
-final class PendingRangeCalculatorServiceEvent extends DiagnosticEvent
-{
- private final PendingRangeCalculatorServiceEventType type;
- private final int taskCount;
-
- public enum PendingRangeCalculatorServiceEventType
- {
- TASK_STARTED,
- TASK_FINISHED_SUCCESSFULLY,
- TASK_EXECUTION_REJECTED,
- TASK_COUNT_CHANGED
- }
-
- PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType type)
- {
- this(type, -1);
- }
-
- PendingRangeCalculatorServiceEvent(PendingRangeCalculatorServiceEventType type,
- int taskCount)
- {
- this.type = type;
- this.taskCount = taskCount;
- }
-
- public int getTaskCount()
- {
- return taskCount;
- }
-
- public PendingRangeCalculatorServiceEventType getType()
- {
- return type;
- }
-
- public Map<String, Serializable> toMap()
- {
- // be extra defensive against nulls and bugs
- if (taskCount < 0)
- return Collections.emptyMap();
- HashMap<String, Serializable> ret = new HashMap<>();
- ret.put("taskCount", taskCount);
- return ret;
- }
-}
diff --git a/src/java/org/apache/cassandra/service/RangeRelocator.java b/src/java/org/apache/cassandra/service/RangeRelocator.java
deleted file mode 100644
index b63c105bd2..0000000000
--- a/src/java/org/apache/cassandra/service/RangeRelocator.java
+++ /dev/null
@@ -1,323 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.Future;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Multimap;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.dht.Range;
-import org.apache.cassandra.dht.RangeStreamer;
-import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.gms.FailureDetector;
-import org.apache.cassandra.locator.AbstractReplicationStrategy;
-import org.apache.cassandra.locator.EndpointsByReplica;
-import org.apache.cassandra.locator.EndpointsForRange;
-import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.RangesAtEndpoint;
-import org.apache.cassandra.locator.RangesByEndpoint;
-import org.apache.cassandra.locator.Replica;
-import org.apache.cassandra.locator.TokenMetadata;
-import org.apache.cassandra.streaming.StreamOperation;
-import org.apache.cassandra.streaming.StreamPlan;
-import org.apache.cassandra.streaming.StreamState;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Pair;
-
-@VisibleForTesting
-public class RangeRelocator
-{
- private static final Logger logger = LoggerFactory.getLogger(StorageService.class);
-
- private final StreamPlan streamPlan = new StreamPlan(StreamOperation.RELOCATION);
- private final InetAddressAndPort localAddress = FBUtilities.getBroadcastAddressAndPort();
- private final TokenMetadata tokenMetaCloneAllSettled;
- // clone to avoid concurrent modification in calculateNaturalReplicas
- private final TokenMetadata tokenMetaClone;
- private final Collection<Token> tokens;
- private final List<String> keyspaceNames;
-
-
- RangeRelocator(Collection<Token> tokens, List<String> keyspaceNames, TokenMetadata tmd)
- {
- this.tokens = tokens;
- this.keyspaceNames = keyspaceNames;
- this.tokenMetaCloneAllSettled = tmd.cloneAfterAllSettled();
- // clone to avoid concurrent modification in calculateNaturalReplicas
- this.tokenMetaClone = tmd.cloneOnlyTokenMap();
- }
-
- @VisibleForTesting
- public RangeRelocator()
- {
- this.tokens = null;
- this.keyspaceNames = null;
- this.tokenMetaCloneAllSettled = null;
- this.tokenMetaClone = null;
- }
-
- /**
- * Wrapper that supplies accessors to the real implementations of the various dependencies for this method
- */
- private static Multimap<InetAddressAndPort, RangeStreamer.FetchReplica> calculateRangesToFetchWithPreferredEndpoints(RangesAtEndpoint fetchRanges,
- AbstractReplicationStrategy strategy,
- String keyspace,
- TokenMetadata tmdBefore,
- TokenMetadata tmdAfter)
- {
- EndpointsByReplica preferredEndpoints =
- RangeStreamer.calculateRangesToFetchWithPreferredEndpoints(DatabaseDescriptor.getEndpointSnitch()::sortedByProximity,
- strategy,
- fetchRanges,
- StorageService.useStrictConsistency,
- tmdBefore,
- tmdAfter,
- keyspace,
- Arrays.asList(new RangeStreamer.FailureDetectorSourceFilter(FailureDetector.instance),
- new RangeStreamer.ExcludeLocalNodeFilter()));
- return RangeStreamer.convertPreferredEndpointsToWorkMap(preferredEndpoints);
- }
-
- /**
- * calculating endpoints to stream current ranges to if needed
- * in some situations node will handle current ranges as part of the new ranges
- **/
- public static RangesByEndpoint calculateRangesToStreamWithEndpoints(RangesAtEndpoint streamRanges,
- AbstractReplicationStrategy strat,
- TokenMetadata tmdBefore,
- TokenMetadata tmdAfter)
- {
- RangesByEndpoint.Builder endpointRanges = new RangesByEndpoint.Builder();
- for (Replica toStream : streamRanges)
- {
- //If the range we are sending is full only send it to the new full replica
- //There will also be a new transient replica we need to send the data to, but not
- //the repaired data
- EndpointsForRange oldEndpoints = strat.calculateNaturalReplicas(toStream.range().right, tmdBefore);
- EndpointsForRange newEndpoints = strat.calculateNaturalReplicas(toStream.range().right, tmdAfter);
- logger.debug("Need to stream {}, current endpoints {}, new endpoints {}", toStream, oldEndpoints, newEndpoints);
-
- for (Replica newEndpoint : newEndpoints)
- {
- Replica oldEndpoint = oldEndpoints.byEndpoint().get(newEndpoint.endpoint());
-
- // Nothing to do
- if (newEndpoint.equals(oldEndpoint))
- continue;
-
- // Completely new range for this endpoint
- if (oldEndpoint == null)
- {
- if (toStream.isTransient() && newEndpoint.isFull())
- throw new AssertionError(String.format("Need to stream %s, but only have %s which is transient and not full", newEndpoint, toStream));
-
- for (Range<Token> intersection : newEndpoint.range().intersectionWith(toStream.range()))
- {
- endpointRanges.put(newEndpoint.endpoint(), newEndpoint.decorateSubrange(intersection));
- }
- }
- else
- {
- Set<Range<Token>> subsToStream = Collections.singleton(toStream.range());
-
- //First subtract what we already have
- if (oldEndpoint.isFull() == newEndpoint.isFull() || oldEndpoint.isFull())
- subsToStream = toStream.range().subtract(oldEndpoint.range());
-
- //Now we only stream what is still replicated
- subsToStream.stream()
- .flatMap(range -> range.intersectionWith(newEndpoint.range()).stream())
- .forEach(tokenRange -> endpointRanges.put(newEndpoint.endpoint(), newEndpoint.decorateSubrange(tokenRange)));
- }
- }
- }
- return endpointRanges.build();
- }
-
- public void calculateToFromStreams()
- {
- logger.debug("Current tmd: {}, Updated tmd: {}", tokenMetaClone, tokenMetaCloneAllSettled);
-
- for (String keyspace : keyspaceNames)
- {
- // replication strategy of the current keyspace
- AbstractReplicationStrategy strategy = Keyspace.open(keyspace).getReplicationStrategy();
-
- logger.info("Calculating ranges to stream and request for keyspace {}", keyspace);
- //From what I have seen we only ever call this with a single token from StorageService.move(Token)
- for (Token newToken : tokens)
- {
- Collection<Token> currentTokens = tokenMetaClone.getTokens(localAddress);
- if (currentTokens.size() > 1 || currentTokens.isEmpty())
- {
- throw new AssertionError("Unexpected current tokens: " + currentTokens);
- }
-
- // calculated parts of the ranges to request/stream from/to nodes in the ring
- Pair<RangesAtEndpoint, RangesAtEndpoint> streamAndFetchOwnRanges;
-
- //In the single node token move there is nothing to do and Range subtraction is broken
- //so it's easier to just identify this case up front.
- if (tokenMetaClone.getTopology().getDatacenterEndpoints().get(DatabaseDescriptor.getEndpointSnitch().getLocalDatacenter()).size() > 1)
- {
- // getting collection of the currently used ranges by this keyspace
- RangesAtEndpoint currentReplicas = strategy.getAddressReplicas(localAddress);
-
- // collection of ranges which this node will serve after move to the new token
- RangesAtEndpoint updatedReplicas = strategy.getPendingAddressRanges(tokenMetaClone, newToken, localAddress);
-
- streamAndFetchOwnRanges = calculateStreamAndFetchRanges(currentReplicas, updatedReplicas);
- }
- else
- {
- streamAndFetchOwnRanges = Pair.create(RangesAtEndpoint.empty(localAddress), RangesAtEndpoint.empty(localAddress));
- }
-
- RangesByEndpoint rangesToStream = calculateRangesToStreamWithEndpoints(streamAndFetchOwnRanges.left, strategy, tokenMetaClone, tokenMetaCloneAllSettled);
- logger.info("Endpoint ranges to stream to " + rangesToStream);
-
- // stream ranges
- for (InetAddressAndPort address : rangesToStream.keySet())
- {
- logger.debug("Will stream range {} of keyspace {} to endpoint {}", rangesToStream.get(address), keyspace, address);
- RangesAtEndpoint ranges = rangesToStream.get(address);
- streamPlan.transferRanges(address, keyspace, ranges);
- }
-
- Multimap<InetAddressAndPort, RangeStreamer.FetchReplica> rangesToFetch = calculateRangesToFetchWithPreferredEndpoints(streamAndFetchOwnRanges.right, strategy, keyspace, tokenMetaClone, tokenMetaCloneAllSettled);
-
- // stream requests
- rangesToFetch.asMap().forEach((address, sourceAndOurReplicas) -> {
- RangesAtEndpoint full = sourceAndOurReplicas.stream()
- .filter(pair -> pair.remote.isFull())
- .map(pair -> pair.local)
- .collect(RangesAtEndpoint.collector(localAddress));
- RangesAtEndpoint trans = sourceAndOurReplicas.stream()
- .filter(pair -> pair.remote.isTransient())
- .map(pair -> pair.local)
- .collect(RangesAtEndpoint.collector(localAddress));
- logger.debug("Will request range {} of keyspace {} from endpoint {}", rangesToFetch.get(address), keyspace, address);
- streamPlan.requestRanges(address, keyspace, full, trans);
- });
-
- logger.debug("Keyspace {}: work map {}.", keyspace, rangesToFetch);
- }
- }
- }
-
- /**
- * Calculate pair of ranges to stream/fetch for given two range collections
- * (current ranges for keyspace and ranges after move to new token)
- *
- * With transient replication the added wrinkle is that if a range transitions from full to transient then
- * we need to stream the range despite the fact that we are retaining it as transient. Some replica
- * somewhere needs to transition from transient to full and we will be the source.
- *
- * If the range is transient and is transitioning to full then always fetch even if the range was already transient
- * since a transiently replicated obviously needs to fetch data to become full.
- *
- * This why there is a continue after checking for instersection because intersection is not sufficient reason
- * to do the subtraction since we might need to stream/fetch data anyways.
- *
- * @param currentRanges collection of the ranges by current token
- * @param updatedRanges collection of the ranges after token is changed
- * @return pair of ranges to stream/fetch for given current and updated range collections
- */
- public static Pair<RangesAtEndpoint, RangesAtEndpoint> calculateStreamAndFetchRanges(RangesAtEndpoint currentRanges, RangesAtEndpoint updatedRanges)
- {
- RangesAtEndpoint.Builder toStream = RangesAtEndpoint.builder(currentRanges.endpoint());
- RangesAtEndpoint.Builder toFetch = RangesAtEndpoint.builder(currentRanges.endpoint());
- logger.debug("Calculating toStream");
- computeRanges(currentRanges, updatedRanges, toStream);
-
- logger.debug("Calculating toFetch");
- computeRanges(updatedRanges, currentRanges, toFetch);
-
- logger.debug("To stream {}", toStream);
- logger.debug("To fetch {}", toFetch);
- return Pair.create(toStream.build(), toFetch.build());
- }
-
- private static void computeRanges(RangesAtEndpoint srcRanges, RangesAtEndpoint dstRanges, RangesAtEndpoint.Builder ranges)
- {
- for (Replica src : srcRanges)
- {
- boolean intersect = false;
- RangesAtEndpoint remainder = null;
- for (Replica dst : dstRanges)
- {
- logger.debug("Comparing {} and {}", src, dst);
- // Stream the full range if there's no intersection
- if (!src.intersectsOnRange(dst))
- continue;
-
- // If we're transitioning from full to transient
- if (src.isFull() && dst.isTransient())
- continue;
-
- if (remainder == null)
- {
- remainder = src.subtractIgnoreTransientStatus(dst.range());
- }
- else
- {
- // Re-subtract ranges to avoid overstreaming in cases when the single range is split or merged
- RangesAtEndpoint.Builder newRemainder = new RangesAtEndpoint.Builder(remainder.endpoint());
- for (Replica replica : remainder)
- newRemainder.addAll(replica.subtractIgnoreTransientStatus(dst.range()));
- remainder = newRemainder.build();
- }
- intersect = true;
- }
-
- if (!intersect)
- {
- assert remainder == null;
- logger.debug(" Doesn't intersect adding {}", src);
- ranges.add(src); // should stream whole old range
- }
- else
- {
- ranges.addAll(remainder);
- logger.debug(" Intersects adding {}", remainder);
- }
- }
- }
-
- public Future<StreamState> stream()
- {
- return streamPlan.execute();
- }
-
- public boolean streamsNeeded()
- {
- return !streamPlan.isEmpty();
- }
-}
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 0aafa9539f..83d6062dba 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -43,7 +43,6 @@ import com.google.common.base.Preconditions;
import com.google.common.cache.CacheLoader;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.Uninterruptibles;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,13 +93,11 @@ import org.apache.cassandra.exceptions.RequestTimeoutException;
import org.apache.cassandra.exceptions.UnavailableException;
import org.apache.cassandra.exceptions.WriteFailureException;
import org.apache.cassandra.exceptions.WriteTimeoutException;
-import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.hints.Hint;
import org.apache.cassandra.hints.HintsService;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.EndpointsForToken;
-import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaLayout;
@@ -134,6 +131,7 @@ import org.apache.cassandra.service.reads.AbstractReadExecutor;
import org.apache.cassandra.service.reads.ReadCallback;
import org.apache.cassandra.service.reads.range.RangeCommands;
import org.apache.cassandra.service.reads.repair.ReadRepair;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.triggers.TriggerExecutor;
import org.apache.cassandra.utils.Clock;
@@ -146,12 +144,9 @@ import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.CountDownLatch;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
+import static com.google.common.collect.Iterables.concat;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
-
-import static com.google.common.collect.Iterables.concat;
-import static org.apache.commons.lang3.StringUtils.join;
-
import static org.apache.cassandra.db.ConsistencyLevel.SERIAL;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casReadMetrics;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casWriteMetrics;
@@ -179,6 +174,7 @@ import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch;
+import static org.apache.commons.lang3.StringUtils.join;
public class StorageProxy implements StorageProxyMBean
{
@@ -969,25 +965,27 @@ public class StorageProxy implements StorageProxyMBean
private static void hintMutation(Mutation mutation)
{
+ ClusterMetadata metadata = ClusterMetadata.current();
String keyspaceName = mutation.getKeyspaceName();
Token token = mutation.key().getToken();
// local writes can timeout, but cannot be dropped (see LocalMutationRunnable and CASSANDRA-6510),
// so there is no need to hint or retry.
- EndpointsForToken replicasToHint = ReplicaLayout.forTokenWriteLiveAndDown(Keyspace.open(keyspaceName), token)
+ EndpointsForToken replicasToHint = ReplicaLayout.forTokenWriteLiveAndDown(metadata, Keyspace.open(keyspaceName), token)
.all()
.filter(StorageProxy::shouldHint);
submitHint(mutation, replicasToHint, null);
}
- public boolean appliesLocally(Mutation mutation)
+ public static boolean appliesLocally(Mutation mutation)
{
+ ClusterMetadata metadata = ClusterMetadata.current();
String keyspaceName = mutation.getKeyspaceName();
Token token = mutation.key().getToken();
InetAddressAndPort local = FBUtilities.getBroadcastAddressAndPort();
- return ReplicaLayout.forTokenWriteLiveAndDown(Keyspace.open(keyspaceName), token)
+ return ReplicaLayout.forTokenWriteLiveAndDown(metadata, Keyspace.open(keyspaceName), token)
.all().endpoints().contains(local);
}
@@ -1008,6 +1006,7 @@ public class StorageProxy implements StorageProxyMBean
long startTime = nanoTime();
+ ClusterMetadata metadata = ClusterMetadata.current();
try
{
@@ -1024,23 +1023,25 @@ public class StorageProxy implements StorageProxyMBean
List<WriteResponseHandlerWrapper> wrappers = new ArrayList<>(mutations.size());
//non-local mutations rely on the base mutation commit-log entry for eventual consistency
Set<Mutation> nonLocalMutations = new HashSet<>(mutations);
- Token baseToken = StorageService.instance.getTokenMetadata().partitioner.getToken(dataKey);
+ Token baseToken = metadata.tokenMap.partitioner().getToken(dataKey);
ConsistencyLevel consistencyLevel = ConsistencyLevel.ONE;
//Since the base -> view replication is 1:1 we only need to store the BL locally
- ReplicaPlan.ForWrite replicaPlan = ReplicaPlans.forLocalBatchlogWrite();
+ ReplicaPlan.ForWrite localReplicaPlan = ReplicaPlans.forLocalBatchlogWrite();
BatchlogCleanup cleanup = new BatchlogCleanup(mutations.size(),
- () -> asyncRemoveFromBatchlog(replicaPlan, batchUUID));
+ () -> asyncRemoveFromBatchlog(localReplicaPlan, batchUUID));
// add a handler for each mutation - includes checking availability, but doesn't initiate any writes, yet
for (Mutation mutation : mutations)
{
String keyspaceName = mutation.getKeyspaceName();
Token tk = mutation.key().getToken();
- AbstractReplicationStrategy replicationStrategy = Keyspace.open(keyspaceName).getReplicationStrategy();
- Optional<Replica> pairedEndpoint = ViewUtils.getViewNaturalEndpoint(replicationStrategy, baseToken, tk);
- EndpointsForToken pendingReplicas = StorageService.instance.getTokenMetadata().pendingEndpointsForToken(tk, keyspaceName);
+ Function<ClusterMetadata, Optional<Replica>> pairedEndpointSupplier = (cm) -> ViewUtils.getViewNaturalEndpoint(cm, keyspaceName, baseToken, tk);
+ Function<ClusterMetadata, EndpointsForToken> pendingReplicasSupplier = (cm) -> cm.pendingEndpointsFor(Keyspace.open(keyspaceName).getMetadata(), tk);
+
+ Optional<Replica> pairedEndpoint = pairedEndpointSupplier.apply(metadata);
+ EndpointsForToken pendingReplicas = pendingReplicasSupplier.apply(metadata);
// if there are no paired endpoints there are probably range movements going on, so we write to the local batchlog to replay later
if (!pairedEndpoint.isPresent())
@@ -1076,13 +1077,17 @@ public class StorageProxy implements StorageProxyMBean
}
else
{
- ReplicaLayout.ForTokenWrite liveAndDown = ReplicaLayout.forTokenWrite(replicationStrategy,
- EndpointsForToken.of(tk, pairedEndpoint.get()),
- pendingReplicas);
+ Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> computeReplicas = (cm) -> {
+ return ReplicaLayout.forTokenWrite(Keyspace.open(keyspaceName).getReplicationStrategy(),
+ EndpointsForToken.of(tk, pairedEndpointSupplier.apply(cm).get()),
+ pendingReplicasSupplier.apply(cm));
+ };
+
+ ReplicaPlan.ForWrite replicaPlan = ReplicaPlans.forWrite(metadata, Keyspace.open(keyspaceName), consistencyLevel, computeReplicas, ReplicaPlans.writeAll);
+
wrappers.add(wrapViewBatchResponseHandler(mutation,
consistencyLevel,
- consistencyLevel,
- liveAndDown,
+ replicaPlan,
baseComplete,
WriteType.BATCH,
cleanup,
@@ -1410,16 +1415,13 @@ public class StorageProxy implements StorageProxyMBean
* Keeps track of ViewWriteMetrics
*/
private static WriteResponseHandlerWrapper wrapViewBatchResponseHandler(Mutation mutation,
- ConsistencyLevel consistencyLevel,
ConsistencyLevel batchConsistencyLevel,
- ReplicaLayout.ForTokenWrite liveAndDown,
+ ReplicaPlan.ForWrite replicaPlan,
AtomicLong baseComplete,
WriteType writeType,
BatchlogResponseHandler.BatchlogCleanup cleanup,
long queryStartNanoTime)
{
- Keyspace keyspace = Keyspace.open(mutation.getKeyspaceName());
- ReplicaPlan.ForWrite replicaPlan = ReplicaPlans.forWrite(keyspace, consistencyLevel, liveAndDown, ReplicaPlans.writeAll);
AbstractReplicationStrategy replicationStrategy = replicaPlan.replicationStrategy();
AbstractWriteResponseHandler<IMutation> writeHandler = replicationStrategy.getWriteResponseHandler(replicaPlan, () -> {
long delay = Math.max(0, currentTimeMillis() - baseComplete.get());
@@ -1702,7 +1704,8 @@ public class StorageProxy implements StorageProxyMBean
*/
public static AbstractWriteResponseHandler<IMutation> mutateCounter(CounterMutation cm, String localDataCenter, long queryStartNanoTime) throws UnavailableException, OverloadedException
{
- Replica replica = findSuitableReplica(cm.getKeyspaceName(), cm.key(), localDataCenter, cm.consistency());
+ ClusterMetadata metadata = ClusterMetadata.current();
+ Replica replica = ReplicaPlans.findCounterLeaderReplica(metadata, cm.getKeyspaceName(), cm.key(), localDataCenter, cm.consistency());
if (replica.isSelf())
{
@@ -1716,15 +1719,17 @@ public class StorageProxy implements StorageProxyMBean
Token tk = cm.key().getToken();
// we build this ONLY to perform the sufficiency check that happens on construction
- ReplicaPlans.forWrite(keyspace, cm.consistency(), tk, ReplicaPlans.writeAll);
+ ReplicaPlans.forWrite(metadata, keyspace, cm.consistency(), tk, ReplicaPlans.writeAll);
// This host isn't a replica, so mark the request as being remote. If this host is a
// replica, applyCounterMutationOnCoordinator() in the branch above will call performWrite(), and
// there we'll mark a local request against the metrics.
writeMetrics.remoteRequests.mark();
+ ReplicaPlan.ForWrite forWrite = ReplicaPlans.forForwardingCounterWrite(metadata, keyspace, tk,
+ clm -> ReplicaPlans.findCounterLeaderReplica(clm, cm.getKeyspaceName(), cm.key(), localDataCenter, cm.consistency()));
// Forward the actual update to the chosen leader replica
- AbstractWriteResponseHandler<IMutation> responseHandler = new WriteResponseHandler<>(ReplicaPlans.forForwardingCounterWrite(keyspace, tk, replica),
+ AbstractWriteResponseHandler<IMutation> responseHandler = new WriteResponseHandler<>(forWrite,
WriteType.COUNTER, null, queryStartNanoTime);
Tracing.trace("Enqueuing counter update to {}", replica);
@@ -1734,53 +1739,6 @@ public class StorageProxy implements StorageProxyMBean
}
}
- /**
- * Find a suitable replica as leader for counter update.
- * For now, we pick a random replica in the local DC (or ask the snitch if
- * there is no replica alive in the local DC).
- * TODO: if we track the latency of the counter writes (which makes sense
- * contrarily to standard writes since there is a read involved), we could
- * trust the dynamic snitch entirely, which may be a better solution. It
- * is unclear we want to mix those latencies with read latencies, so this
- * may be a bit involved.
- */
- private static Replica findSuitableReplica(String keyspaceName, DecoratedKey key, String localDataCenter, ConsistencyLevel cl) throws UnavailableException
- {
- Keyspace keyspace = Keyspace.open(keyspaceName);
- IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
- AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy();
- EndpointsForToken replicas = replicationStrategy.getNaturalReplicasForToken(key);
-
- // CASSANDRA-13043: filter out those endpoints not accepting clients yet, maybe because still bootstrapping
- replicas = replicas.filter(replica -> StorageService.instance.isRpcReady(replica.endpoint()));
-
- // CASSANDRA-17411: filter out endpoints that are not alive
- replicas = replicas.filter(replica -> FailureDetector.instance.isAlive(replica.endpoint()));
-
- // TODO have a way to compute the consistency level
- if (replicas.isEmpty())
- throw UnavailableException.create(cl, cl.blockFor(replicationStrategy), 0);
-
- List<Replica> localReplicas = new ArrayList<>(replicas.size());
-
- for (Replica replica : replicas)
- if (snitch.getDatacenter(replica).equals(localDataCenter))
- localReplicas.add(replica);
-
- if (localReplicas.isEmpty())
- {
- // If the consistency required is local then we should not involve other DCs
- if (cl.isDatacenterLocal())
- throw UnavailableException.create(cl, cl.blockFor(replicationStrategy), 0);
-
- // No endpoint in local DC, pick the closest endpoint according to the snitch
- replicas = snitch.sortedByProximity(FBUtilities.getBroadcastAddressAndPort(), replicas);
- return replicas.get(0);
- }
-
- return localReplicas.get(ThreadLocalRandom.current().nextInt(localReplicas.size()));
- }
-
// Must be called on a replica of the mutation. This replica becomes the
// leader of this mutation.
public static AbstractWriteResponseHandler<IMutation> applyCounterMutationOnLeader(CounterMutation cm, String localDataCenter, Runnable callback, long queryStartNanoTime)
@@ -2081,11 +2039,12 @@ public class StorageProxy implements StorageProxyMBean
AbstractReadExecutor[] reads = new AbstractReadExecutor[cmdCount];
+ ClusterMetadata metadata = ClusterMetadata.current();
// Get the replica locations, sorted by response time according to the snitch, and create a read executor
// for type of speculation we'll use in this read
for (int i=0; i<cmdCount; i++)
{
- reads[i] = AbstractReadExecutor.getReadExecutor(commands.get(i), consistencyLevel, queryStartNanoTime);
+ reads[i] = AbstractReadExecutor.getReadExecutor(metadata, commands.get(i), consistencyLevel, queryStartNanoTime);
if (reads[i].hasLocalRead())
readMetrics.localRequests.mark();
@@ -2094,7 +2053,7 @@ public class StorageProxy implements StorageProxyMBean
}
// sends a data request to the closest replica, and a digest request to the others. If we have a speculating
- // read executoe, we'll only send read requests to enough replicas to satisfy the consistency level
+ // read executor, we'll only send read requests to enough replicas to satisfy the consistency level
for (int i=0; i<cmdCount; i++)
{
reads[i].executeAsync();
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 0e00bb9002..c88d68e79f 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -5575,7 +5575,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
@Override
public List<String> describeCMS()
{
- return ClusterMetadata.current().cmsMembers().stream().sorted().map(ep -> ep.toString()).collect(Collectors.toList());
+ return ClusterMetadata.current().fullCMSMembers().stream().sorted().map(Object::toString).collect(Collectors.toList());
}
@Override
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index 087e59587e..ed0814934b 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -1089,6 +1089,7 @@ public interface StorageServiceMBean extends NotificationEmitter
/** Returns a map of schema version -> list of endpoints reporting that version that we need schema updates for */
@Deprecated
public Map<String, Set<InetAddress>> getOutstandingSchemaVersions();
+ @Deprecated
public Map<String, Set<String>> getOutstandingSchemaVersionsWithPort();
// see CASSANDRA-3200
@@ -1186,6 +1187,7 @@ public interface StorageServiceMBean extends NotificationEmitter
public void setSkipStreamDiskSpaceCheck(boolean value);
public boolean getSkipStreamDiskSpaceCheck();
- public void addToCms(List<String> ignoredEndpoints);
- public List<String> describeCMS();
+ void addToCms(List<String> endpoint);
+ List<String> describeCMS();
+ public void sealPeriod();
}
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
index a7f58e1383..bf85d49edc 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepare.java
@@ -44,6 +44,7 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.metrics.PaxosMetrics;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
@@ -51,7 +52,6 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
-import org.apache.cassandra.service.PendingRangeCalculatorService;
import org.apache.cassandra.service.paxos.PaxosPrepare.Status.Outcome;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.vint.VIntCoding;
@@ -63,7 +63,6 @@ import static org.apache.cassandra.net.Verb.PAXOS2_PREPARE_REQ;
import static org.apache.cassandra.net.Verb.PAXOS2_PREPARE_RSP;
import static org.apache.cassandra.service.paxos.Ballot.Flag.NONE;
import static org.apache.cassandra.service.paxos.Commit.*;
-import static org.apache.cassandra.service.paxos.Commit.CompareResult.WAS_REPROPOSED_BY;
import static org.apache.cassandra.service.paxos.Paxos.*;
import static org.apache.cassandra.service.paxos.PaxosPrepare.Status.Outcome.*;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
@@ -469,13 +468,14 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
Stage.GOSSIP.executor().execute(() -> {
Gossiper.instance.notifyFailureDetector(permitted.gossipInfo);
Gossiper.instance.applyStateLocally(permitted.gossipInfo);
-
// TODO: We should also wait for schema pulls/pushes, however this would be quite an involved change to MigrationManager
// (which currently drops some migration tasks on the floor).
// Note it would be fine for us to fail to complete the migration task and simply treat this response as a failure/timeout.
// once any pending ranges have been calculated, refresh our Participants list and submit the promise
- PendingRangeCalculatorService.instance.executeWhenFinished(() -> permittedOrTerminateIfElectorateMismatch(permitted, from));
+ // todo: verify that this is correct, we no longer have any pending ranges, just call this immediately
+ // PendingRangeCalculatorService.instance.executeWhenFinished(() -> permittedOrTerminateIfElectorateMismatch(permitted, from));
+ permittedOrTerminateIfElectorateMismatch(permitted, from);
});
}
@@ -1005,6 +1005,8 @@ public class PaxosPrepare extends PaxosRequestCallback<PaxosPrepare.Response> im
@Override
public void doVerb(Message<Request> message)
{
+ ClusterMetadataService.instance().maybeCatchup(message.epoch());
+
Response response = execute(message.payload, message.from());
if (response == null)
MessagingService.instance().respondWithFailure(UNKNOWN, message);
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java b/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java
index d88323cbcf..e00e1019fd 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java
@@ -42,10 +42,6 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.exceptions.UnavailableException;
-import org.apache.cassandra.gms.ApplicationState;
-import org.apache.cassandra.gms.EndpointState;
-import org.apache.cassandra.gms.Gossiper;
-import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@@ -58,6 +54,8 @@ import org.apache.cassandra.net.RequestCallbackWithFailure;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.CassandraVersion;
import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
@@ -545,7 +543,7 @@ public class PaxosRepair extends AbstractPaxosRepair
*/
public static boolean hasSufficientLiveNodesForTopologyChange(Keyspace keyspace, Range<Token> range, Collection<InetAddressAndPort> liveEndpoints)
{
- return hasSufficientLiveNodesForTopologyChange(keyspace.getReplicationStrategy().getNaturalReplicasForToken(range.right).endpoints(),
+ return hasSufficientLiveNodesForTopologyChange(ClusterMetadata.current().placements.get(keyspace.getMetadata().params.replication).reads.forRange(range).endpoints(),
liveEndpoints,
DatabaseDescriptor.getEndpointSnitch()::getDatacenter,
DatabaseDescriptor.paxoTopologyRepairNoDcChecks(),
@@ -658,40 +656,21 @@ public class PaxosRepair extends AbstractPaxosRepair
return (version.major == 4 && version.minor > 0) || version.major > 4;
}
- static String getPeerVersion(InetAddressAndPort peer)
+ static boolean validatePeerCompatibility(ClusterMetadata metadata, Replica peer)
{
- EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(peer);
- if (epState == null)
- return null;
-
- VersionedValue value = epState.getApplicationState(ApplicationState.RELEASE_VERSION);
- if (value == null)
- return null;
-
- try
- {
- return value.value;
- }
- catch (IllegalArgumentException e)
- {
- return null;
- }
- }
-
- static boolean validatePeerCompatibility(Replica peer)
- {
- String versionString = getPeerVersion(peer.endpoint());
- CassandraVersion version = versionString != null ? new CassandraVersion(versionString) : null;
+ NodeId nodeId = metadata.directory.peerId(peer.endpoint());
+ CassandraVersion version = metadata.directory.version(nodeId).cassandraVersion;
boolean result = validateVersionCompatibility(version);
if (!result)
- logger.info("PaxosRepair isn't supported by {} on version {}", peer, versionString);
+ logger.info("PaxosRepair isn't supported by {} on version {}", peer, version);
return result;
}
static boolean validatePeerCompatibility(TableMetadata table, Range<Token> range)
{
+ ClusterMetadata metadata = ClusterMetadata.current();
Participants participants = Participants.get(table, range.right, ConsistencyLevel.SERIAL);
- return Iterables.all(participants.all, PaxosRepair::validatePeerCompatibility);
+ return Iterables.all(participants.all, (participant) -> validatePeerCompatibility(metadata, participant));
}
public static boolean validatePeerCompatibility(TableMetadata table, Collection<Range<Token>> ranges)
diff --git a/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosCleanup.java b/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosCleanup.java
index 6eb1ebd574..c71577d389 100644
--- a/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosCleanup.java
+++ b/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosCleanup.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.service.paxos.cleanup;
import java.util.Collection;
-import java.util.List;
import java.util.concurrent.Executor;
import java.util.function.Consumer;
@@ -36,13 +35,11 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.RangesAtEndpoint;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
-import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.paxos.Ballot;
-import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.concurrent.AsyncFuture;
import org.apache.cassandra.utils.concurrent.Future;
@@ -120,18 +117,7 @@ public class PaxosCleanup extends AsyncFuture<Void> implements Runnable
private static boolean isOutOfRange(String ksName, Collection<Range<Token>> repairRanges)
{
Keyspace keyspace = Keyspace.open(ksName);
- List<Range<Token>> localRanges = Range.normalize(keyspace.getReplicationStrategy()
- .getAddressReplicas()
- .get(FBUtilities.getBroadcastAddressAndPort())
- .ranges());
-
- RangesAtEndpoint pendingRanges = StorageService.instance.getTokenMetadata().getPendingRanges(ksName, FBUtilities.getBroadcastAddressAndPort());
- if (!pendingRanges.isEmpty())
- {
- localRanges.addAll(pendingRanges.ranges());
- localRanges = Range.normalize(localRanges);
- }
-
+ Collection<Range<Token>> localRanges = Range.normalize(ClusterMetadata.current().localWriteRanges(keyspace.getMetadata()));
for (Range<Token> repairRange : Range.normalize(repairRanges))
{
if (!Iterables.any(localRanges, localRange -> localRange.contains(repairRange)))
diff --git a/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosStartPrepareCleanup.java b/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosStartPrepareCleanup.java
index 9f30692ad4..c6191f0019 100644
--- a/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosStartPrepareCleanup.java
+++ b/src/java/org/apache/cassandra/service/paxos/cleanup/PaxosStartPrepareCleanup.java
@@ -39,7 +39,6 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.*;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
-import org.apache.cassandra.service.PendingRangeCalculatorService;
import org.apache.cassandra.service.paxos.Ballot;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.service.paxos.PaxosRepairHistory;
@@ -126,8 +125,6 @@ public class PaxosStartPrepareCleanup extends AsyncFuture<PaxosCleanupHistory> i
// (which currently drops some migration tasks on the floor).
// Note it would be fine for us to fail to complete the migration task and simply treat this response as a failure/timeout.
}
- // even if we have th latest gossip info, wait until pending range calculations are complete
- PendingRangeCalculatorService.instance.blockUntilFinished();
}
public static class Request
diff --git a/src/java/org/apache/cassandra/service/paxos/uncommitted/UncommittedTableData.java b/src/java/org/apache/cassandra/service/paxos/uncommitted/UncommittedTableData.java
index dd47e40ce6..e3fc0b9a5a 100644
--- a/src/java/org/apache/cassandra/service/paxos/uncommitted/UncommittedTableData.java
+++ b/src/java/org/apache/cassandra/service/paxos/uncommitted/UncommittedTableData.java
@@ -185,7 +185,7 @@ public class UncommittedTableData
return Range.normalize(FULL_RANGE);
String ksName = table.keyspace.getName();
- List<Range<Token>> ranges = StorageService.instance.getLocalAndPendingRanges(ksName);
+ Collection<Range<Token>> ranges = StorageService.instance.getLocalAndPendingRanges(ksName);
// don't filter anything if we're not aware of any locally replicated ranges
if (ranges.isEmpty())
diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
index 1613283300..e0a13fe27f 100644
--- a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
+++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
@@ -42,6 +42,7 @@ import org.apache.cassandra.locator.ReplicaPlans;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageProxy.LocalReadRunnable;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.service.reads.repair.ReadRepair;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
@@ -147,7 +148,7 @@ public abstract class AbstractReadExecutor
traceState.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint);
if (null == message)
- message = readCommand.createMessage(false);
+ message = readCommand.createMessage(false).withEpoch(ClusterMetadata.current().epoch);
MessagingService.instance().sendWithCallback(message, endpoint, handler);
}
@@ -181,13 +182,13 @@ public abstract class AbstractReadExecutor
/**
* @return an executor appropriate for the configured speculative read policy
*/
- public static AbstractReadExecutor getReadExecutor(SinglePartitionReadCommand command, ConsistencyLevel consistencyLevel, long queryStartNanoTime) throws UnavailableException
+ public static AbstractReadExecutor getReadExecutor(ClusterMetadata metadata, SinglePartitionReadCommand command, ConsistencyLevel consistencyLevel, long queryStartNanoTime) throws UnavailableException
{
Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(command.metadata().id);
SpeculativeRetryPolicy retry = cfs.metadata().params.speculativeRetry;
- ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forRead(keyspace, command.partitionKey().getToken(), consistencyLevel, retry);
+ ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forRead(metadata, keyspace, command.partitionKey().getToken(), consistencyLevel, retry);
// Speculative retry is disabled *OR*
// 11980: Disable speculative retry if using EACH_QUORUM in order to prevent miscounting DC responses
@@ -432,7 +433,7 @@ public abstract class AbstractReadExecutor
logger.trace("Timed out waiting on digest mismatch repair requests");
// the caught exception here will have CL.ALL from the repair command,
// not whatever CL the initial command was at (CASSANDRA-7947)
- throw new ReadTimeoutException(replicaPlan().consistencyLevel(), handler.blockFor - 1, handler.blockFor, true);
+ throw new ReadTimeoutException(replicaPlan().consistencyLevel(), handler.replicaPlan().readQuorum() - 1, handler.replicaPlan().readQuorum(), true);
}
}
diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
index 673fe8be55..844e87eb53 100644
--- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
+++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java
@@ -63,8 +63,10 @@ import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.vint.VIntCoding;
+import static org.apache.cassandra.config.CassandraRelevantProperties.LINE_SEPARATOR;
import static org.apache.cassandra.db.TypeSizes.sizeof;
public class ClusterMetadata
@@ -514,11 +516,82 @@ public class ClusterMetadata
}
}
+ public String legacyToString()
+ {
+ StringBuilder sb = new StringBuilder();
+ Set<Pair<Token, InetAddressAndPort>> normal = new HashSet<>();
+ Set<Pair<Token, InetAddressAndPort>> bootstrapping = new HashSet<>();
+ Set<InetAddressAndPort> leaving = new HashSet<>();
+
+ for (Map.Entry<NodeId, NodeState> entry : directory.states.entrySet())
+ {
+ InetAddressAndPort endpoint = directory.endpoint(entry.getKey());
+ switch (entry.getValue())
+ {
+ case BOOTSTRAPPING:
+ for (Token t : tokenMap.tokens(entry.getKey()))
+ bootstrapping.add(Pair.create(t, endpoint));
+ break;
+ case LEAVING:
+ leaving.add(endpoint);
+ break;
+ case JOINED:
+ for (Token t : tokenMap.tokens(entry.getKey()))
+ normal.add(Pair.create(t, endpoint));
+ break;
+ case MOVING:
+ // todo when adding MOVE
+ break;
+ }
+ }
+
+ if (!normal.isEmpty())
+ {
+ sb.append("Normal Tokens:");
+ sb.append(LINE_SEPARATOR.getString());
+ for (Pair<Token, InetAddressAndPort> ep : normal)
+ {
+ sb.append(ep.right);
+ sb.append(':');
+ sb.append(ep.left);
+ sb.append(LINE_SEPARATOR.getString());
+ }
+ }
+
+ if (!bootstrapping.isEmpty())
+ {
+ sb.append("Bootstrapping Tokens:" );
+ sb.append(LINE_SEPARATOR.getString());
+ for (Pair<Token, InetAddressAndPort> entry : bootstrapping)
+ {
+ sb.append(entry.right).append(':').append(entry.left);
+ sb.append(LINE_SEPARATOR.getString());
+ }
+ }
+
+ if (!leaving.isEmpty())
+ {
+ sb.append("Leaving Endpoints:");
+ sb.append(LINE_SEPARATOR.getString());
+ for (InetAddressAndPort ep : leaving)
+ {
+ sb.append(ep);
+ sb.append(LINE_SEPARATOR.getString());
+ }
+ }
+ return sb.toString();
+ }
+
@Override
public String toString()
{
return "ClusterMetadata{" +
"epoch=" + epoch +
+ ", schema=" + schema +
+ ", directory=" + directory +
+ ", tokenMap=" + tokenMap +
+ ", placements=" + placements +
+ ", lockedRanges=" + lockedRanges +
'}';
}
diff --git a/src/java/org/apache/cassandra/tcm/compatibility/AsEndpoints.java b/src/java/org/apache/cassandra/tcm/compatibility/AsEndpoints.java
deleted file mode 100644
index fd8a4d39f6..0000000000
--- a/src/java/org/apache/cassandra/tcm/compatibility/AsEndpoints.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.tcm.compatibility;
-
-import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.tcm.membership.NodeId;
-
-public interface AsEndpoints
-{
- public NodeId peerId(InetAddressAndPort endpoint);
- public InetAddressAndPort endpoint(NodeId id);
-}
diff --git a/src/java/org/apache/cassandra/tcm/compatibility/AsLocations.java b/src/java/org/apache/cassandra/tcm/compatibility/AsLocations.java
deleted file mode 100644
index 3286f1f1d9..0000000000
--- a/src/java/org/apache/cassandra/tcm/compatibility/AsLocations.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.tcm.compatibility;
-
-import java.util.Map;
-import java.util.Set;
-
-import com.google.common.collect.Multimap;
-
-import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.tcm.membership.Location;
-import org.apache.cassandra.tcm.membership.NodeId;
-
-public interface AsLocations
-{
- Location location(NodeId peer);
- Set<InetAddressAndPort> datacenterEndpoints(String datacenter);
- Multimap<String, InetAddressAndPort> allDatacenterEndpoints();
-
- Multimap<String, InetAddressAndPort> datacenterRacks(String datacenter);
- Map<String, Multimap<String, InetAddressAndPort>> allDatacenterRacks();
-}
diff --git a/src/java/org/apache/cassandra/tcm/compatibility/AsTokenMap.java b/src/java/org/apache/cassandra/tcm/compatibility/AsTokenMap.java
deleted file mode 100644
index 9960dc10b8..0000000000
--- a/src/java/org/apache/cassandra/tcm/compatibility/AsTokenMap.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.tcm.compatibility;
-
-import java.util.List;
-
-import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.tcm.membership.NodeId;
-
-public interface AsTokenMap
-{
- public IPartitioner partitioner();
- // todo: this could return NavigableSet
- public List<Token> tokens();
- public NodeId owner(Token token);
-}
diff --git a/src/java/org/apache/cassandra/tcm/listeners/PlacementsChangeListener.java b/src/java/org/apache/cassandra/tcm/listeners/PlacementsChangeListener.java
new file mode 100644
index 0000000000..0dfeff2c08
--- /dev/null
+++ b/src/java/org/apache/cassandra/tcm/listeners/PlacementsChangeListener.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.tcm.listeners;
+
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.tcm.ClusterMetadata;
+
+public class PlacementsChangeListener implements ChangeListener
+{
+ @Override
+ public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next)
+ {
+ if (shouldInvalidate(prev, next))
+ StorageService.instance.invalidateLocalRanges();
+ }
+
+ private boolean shouldInvalidate(ClusterMetadata prev, ClusterMetadata next)
+ {
+ if (!prev.placements.lastModified().equals(next.placements.lastModified()) &&
+ !prev.placements.equals(next.placements)) // <- todo should we update lastModified if the result is the same?
+ return true;
+
+ if (prev.schema.getKeyspaces().size() != next.schema.getKeyspaces().size())
+ return true;
+
+ // can't rely only on placements alone since we can move a ks from rf=1 to rf=3 and the rf=3 params might already exist in the placements:
+ for (KeyspaceMetadata ksm : prev.schema.getKeyspaces())
+ {
+ Keyspace newKs = next.schema.getKeyspace(ksm.name);
+ if (newKs == null || !ksm.params.equals(newKs.getMetadata().params))
+ return true;
+ }
+ return false;
+ }
+}
diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java
index 7adf3fd0fb..1a05a29460 100644
--- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java
+++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java
@@ -48,6 +48,7 @@ import org.apache.cassandra.tcm.listeners.ChangeListener;
import org.apache.cassandra.tcm.listeners.InitializationListener;
import org.apache.cassandra.tcm.listeners.LogListener;
import org.apache.cassandra.tcm.listeners.MetadataSnapshotListener;
+import org.apache.cassandra.tcm.listeners.PlacementsChangeListener;
import org.apache.cassandra.tcm.listeners.SchemaListener;
import org.apache.cassandra.tcm.listeners.PaxosRepairListener;
import org.apache.cassandra.tcm.transformations.ForceSnapshot;
@@ -545,6 +546,7 @@ public abstract class LocalLog implements Closeable
addListener(snapshotListener());
addListener(new InitializationListener());
addListener(new SchemaListener());
+ addListener(new PlacementsChangeListener());
addListener(new MetadataSnapshotListener());
addListener(new PaxosRepairListener());
}
diff --git a/src/java/org/apache/cassandra/tcm/membership/Directory.java b/src/java/org/apache/cassandra/tcm/membership/Directory.java
index 94a67c2864..961ff669b1 100644
--- a/src/java/org/apache/cassandra/tcm/membership/Directory.java
+++ b/src/java/org/apache/cassandra/tcm/membership/Directory.java
@@ -34,8 +34,6 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MetadataValue;
-import org.apache.cassandra.tcm.compatibility.AsEndpoints;
-import org.apache.cassandra.tcm.compatibility.AsLocations;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.net.MessagingService;
@@ -46,7 +44,7 @@ import org.apache.cassandra.utils.btree.BTreeMultimap;
import static org.apache.cassandra.db.TypeSizes.sizeof;
-public class Directory implements MetadataValue<Directory>, AsEndpoints, AsLocations
+public class Directory implements MetadataValue<Directory>
{
public static final Serializer serializer = new Serializer();
diff --git a/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java b/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java
index 8dde234c5f..d72c56b346 100644
--- a/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java
+++ b/src/java/org/apache/cassandra/tcm/ownership/TokenMap.java
@@ -37,7 +37,6 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.MetadataValue;
-import org.apache.cassandra.tcm.compatibility.AsTokenMap;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.serialization.MetadataSerializer;
import org.apache.cassandra.tcm.serialization.Version;
@@ -47,7 +46,7 @@ import org.apache.cassandra.utils.SortedBiMultiValMap;
import static org.apache.cassandra.db.TypeSizes.sizeof;
-public class TokenMap implements MetadataValue<TokenMap>, AsTokenMap
+public class TokenMap implements MetadataValue<TokenMap>
{
public static final Serializer serializer = new Serializer();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org