aweisberg commented on code in PR #4708:
URL: https://github.com/apache/cassandra/pull/4708#discussion_r3553561153
##########
src/java/org/apache/cassandra/locator/SatelliteReplicationStrategy.java:
##########
@@ -513,4 +631,1286 @@ public boolean
hasSameSettings(AbstractReplicationStrategy other)
primaryDC.equals(otherStrategy.primaryDC) &&
disabledDCs.equals(otherStrategy.disabledDCs);
}
+
+ private AbstractReplicationStrategy strategy(ClusterMetadata metadata)
+ {
+ return
metadata.schema.getKeyspaceMetadata(keyspaceName).replicationStrategy;
+ }
+
+ static abstract class CoordinationPlanner<E extends Endpoints<E>, L
extends ReplicaLayout<E>, P extends ReplicaPlan<E, P>>
+ {
+ final ClusterMetadata metadata;
+ final Keyspace keyspace;
+ final ConsistencyLevel cl;
+ final SatelliteReplicationStrategy strategy;
+ final String primary;
+ final String satellite;
+ final List<String> dcs;
+ final L fullLayout;
+ final L liveLayout;
+ final Map<String, L> fullLayouts;
+ final Map<String, L> liveLayouts;
+ final Map<String, E> fullEndpoints;
+ final Map<String, E> liveEndpoints;
+
+ public CoordinationPlanner(ClusterMetadata metadata, Keyspace
keyspace, ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String
primary, L fullLayout)
+ {
+ this.metadata = metadata;
+ this.keyspace = keyspace;
+ this.cl = cl;
+ this.strategy = strategy;
+ this.primary = primary;
+ this.satellite = strategy.getSatelliteForDC(primary);
+
+ this.dcs = createDcList();
+ Preconditions.checkState(!dcs.isEmpty(), "No DCs available for
request (primary=%s, all disabled?)", primary);
+
+ Set<String> dcSet = new HashSet<>(dcs);
+ this.fullLayout = filterLayout(fullLayout, rp ->
dcSet.contains(metadata.locator.location(rp.endpoint()).datacenter));
+ this.liveLayout = filterLayout(this.fullLayout,
FailureDetector.isReplicaAlive);
+
+ this.fullLayouts = Maps.newHashMapWithExpectedSize(dcs.size());
+ this.liveLayouts = Maps.newHashMapWithExpectedSize(dcs.size());
+
+ this.fullEndpoints = Maps.newHashMapWithExpectedSize(dcs.size());
+ this.liveEndpoints = Maps.newHashMapWithExpectedSize(dcs.size());
+
+ populateDcLayouts();
+ }
+
+ AbstractReplicationStrategy strategy(ClusterMetadata metadata)
+ {
+ return
metadata.schema.getKeyspaceMetadata(keyspace.getName()).replicationStrategy;
+ }
+
+ List<String> createDcList()
+ {
+ List<String> results = new ArrayList<>();
+ if (!strategy.disabledDCs.contains(primary))
+ results.add(primary);
+
+ if (satellite != null && !strategy.disabledDCs.contains(satellite))
+ results.add(satellite);
+
+ for (String dc : strategy.fullDCs.keySet())
+ {
+ if (dc.equals(primary) || dc.equals(satellite) ||
strategy.disabledDCs.contains(dc))
+ continue;
+ results.add(dc);
+ }
+
+ Preconditions.checkState(!results.isEmpty());
+ return results;
+ }
+
+ void populateDcLayouts()
+ {
+ for (String dc : dcs)
+ {
+ Predicate<Replica> dcFilter = rp ->
metadata.locator.location(rp.endpoint()).datacenter.equals(dc);
+ L full = filterLayout(fullLayout, dcFilter);
+ L live = filterLayout(liveLayout, dcFilter);
+
+ fullLayouts.put(dc, full);
+ liveLayouts.put(dc, live);
+
+ fullEndpoints.put(dc, full.natural());
+ liveEndpoints.put(dc, live.natural());
+ }
+
+ if (ADDL_CHECKS_ENABLED)
+ Preconditions.checkState(fullLayouts.size() == dcs.size(),
"populateDcLayouts: fullLayouts.size()=%s != dcs.size()=%s",
fullLayouts.size(), dcs.size());
+ }
+
+ ResponseTrackerBuilder<E, L, P> createResponseTrackerBuilder()
+ {
+ switch (cl)
+ {
+ case ONE:
+ case LOCAL_ONE:
+ case NODE_LOCAL:
+ case ANY:
+ case TWO:
+ case THREE:
+ case ALL:
+ return new ResponseTrackerBuilder.Simple<>(this);
+
+ case QUORUM:
+ case LOCAL_QUORUM:
+ // LOCAL_QUORUM an QUORUM mean the same thing in SRS
+ return new ResponseTrackerBuilder.Quorum<>(this);
+
+ case EACH_QUORUM:
+ return new ResponseTrackerBuilder.EachQuorum<>(this);
+
+ default:
+ throw new IllegalStateException("Unsupported consistency
level: " + cl);
+ }
+ }
+
+ ResponseTracker createResponseTracker()
+ {
+ ResponseTrackerBuilder<E, L, P> builder =
createResponseTrackerBuilder();
+ return builder.build();
+ }
+
+ abstract L filterLayout(L layout, Predicate<Replica> predicate);
+ abstract ResponseTracker createDcResponseTracker(String dc);
+
+ abstract P recomputePlan(ClusterMetadata metadata);
+ abstract P createReplicaPlan();
+
+ interface IndexReadExceptionCheck<E extends Endpoints<E>>
+ {
+ void maybeThrowIndexReadException(int initial, int filtered,
Map<String, E> candidates, Map<InetAddressAndPort, Index.Status>
indexStatusMap);
+ }
+
+ static class ForWrite extends CoordinationPlanner<EndpointsForToken,
ReplicaLayout.ForTokenWrite, ReplicaPlan.ForWrite>
+ {
+ final ReplicaPlans.Selector selector;
+
+ public ForWrite(ClusterMetadata metadata, Keyspace keyspace,
ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String primary,
ReplicaLayout.ForTokenWrite fullLayout, ReplicaPlans.Selector selector)
+ {
+ super(metadata, keyspace, cl, strategy, primary, fullLayout);
+ this.selector = selector;
+ if (ADDL_CHECKS_ENABLED)
+ {
+
Preconditions.checkState(!strategy.disabledDCs.contains(primary));
+ }
+ }
+
+ @Override
+ ResponseTracker createDcResponseTracker(String dc)
+ {
+ // this basically hardcodes ReplicaPlans.writeAll
+ Predicate<InetAddressAndPort> dcFilter = ep ->
metadata.locator.location(ep).datacenter.equals(dc);
+
+ ReplicaLayout.ForTokenWrite dcLayout = fullLayouts.get(dc);
+ Preconditions.checkNotNull(dcLayout);
+
+ int blockFor = strategy.calculateQuorum(dc);
+ int totalContacts = dcLayout.all().size();
+
+ // Only count pending replicas that are actually in the
selected contacts
+ int pendingInDc = dcLayout.pending().size();
+
+ if (pendingInDc == 0)
+ {
+ return new SimpleResponseTracker(blockFor, totalContacts,
dcFilter);
+ }
+ else
+ {
+ int committedContacts = totalContacts - pendingInDc;
+ int totalBlockFor = blockFor + pendingInDc;
+ Predicate<InetAddressAndPort> isPending = ep ->
dcLayout.pending.endpoints().contains(ep);
+ return new WriteResponseTracker(blockFor, totalBlockFor,
+ committedContacts,
pendingInDc,
+ isPending, dcFilter);
+ }
+ }
+
+ @Override
+ ReplicaPlan.ForWrite recomputePlan(ClusterMetadata metadata)
+ {
+ Token token = fullLayout.token();
+ return strategy(metadata).planForWrite(metadata, keyspace, cl,
+ cm ->
ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, token),
+ selector).replicas();
+ }
+
+ @Override
+ ReplicaLayout.ForTokenWrite
filterLayout(ReplicaLayout.ForTokenWrite layout, Predicate<Replica> predicate)
+ {
+ return layout.filter(predicate);
+ }
+
+ private ReplicaLayout.ForTokenWrite
mergeLayouts(Collection<ReplicaLayout.ForTokenWrite> layouts)
+ {
+ ReplicaCollection.Builder<EndpointsForToken> naturalBuilder =
null;
+ ReplicaCollection.Builder<EndpointsForToken> pendingBuilder =
null;
+
+ for (ReplicaLayout.ForTokenWrite layout : layouts)
+ {
+ EndpointsForToken natural = layout.natural();
+ EndpointsForToken pending = layout.pending();
+ if (naturalBuilder == null)
+ {
+ naturalBuilder = natural.newBuilder(natural.size());
+ pendingBuilder = pending.newBuilder(pending.size());
+ }
+
+ naturalBuilder.addAll(natural);
+ pendingBuilder.addAll(pending);
+ }
+
+ return new ReplicaLayout.ForTokenWrite(strategy,
naturalBuilder.build(), pendingBuilder.build());
+ }
+
+ @Override
+ ReplicaPlan.ForWrite createReplicaPlan()
+ {
+
+ ReplicaLayout.ForTokenWrite fullLayout =
mergeLayouts(fullLayouts.values());
+ ReplicaLayout.ForTokenWrite liveLayout =
mergeLayouts(liveLayouts.values());
+ EndpointsForToken contacts = selector.select(cl, fullLayout,
liveLayout);
+
+ return new ReplicaPlan.ForWrite(keyspace,
+ strategy,
+ cl,
+ fullLayout.pending(),
+ fullLayout.all(),
+ liveLayout.all(),
+ contacts,
+ this::recomputePlan,
+ metadata.epoch);
+ }
+ }
+
+ static abstract class ForRead<E extends Endpoints<E>, L extends
ReplicaLayout<E>, P extends ReplicaPlan.ForRead<E, P>> extends
CoordinationPlanner<E, L, P>
+ {
+ final @Nullable Index.QueryPlan indexQueryPlan;
+ final boolean alwaysSpeculate;
+ final boolean throwOnInsufficientLiveReplicas;
+
+ public ForRead(ClusterMetadata metadata, Keyspace keyspace,
ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String primary, L
fullLayout, @Nullable Index.QueryPlan indexQueryPlan, boolean alwaysSpeculate,
boolean throwOnInsufficientLiveReplicas)
+ {
+ super(metadata, keyspace, cl, strategy, primary, fullLayout);
+ this.indexQueryPlan = indexQueryPlan;
+ this.alwaysSpeculate = alwaysSpeculate;
+ this.throwOnInsufficientLiveReplicas =
throwOnInsufficientLiveReplicas;
+ }
+
+ @Override
+ ResponseTracker createDcResponseTracker(String dc)
+ {
+ Predicate<InetAddressAndPort> dcFilter = ep ->
metadata.locator.location(ep).datacenter.equals(dc);
+
+ L dcLayout = fullLayouts.get(dc);
+ int blockFor = strategy.calculateQuorum(dc);
+ int totalContacts = dcLayout.all().size();
+
+ return new SimpleResponseTracker(blockFor, totalContacts,
dcFilter);
+ }
+
+ Map<String, E> candidates(IndexReadExceptionCheck<E> check)
+ {
+ if (indexQueryPlan == null)
+ return liveEndpoints;
+
+ Map<String, E> candidates = new HashMap<>();
+ Map<InetAddressAndPort, Index.Status> indexStatusMap = new
HashMap<>();
+ int initial = 0;
+ int filtered = 0;
+ for (Map.Entry<String, E> entry : liveEndpoints.entrySet())
+ {
+ E liveEndpoints = entry.getValue();
+ E queryableEndpoints =
IndexStatusManager.instance.filterForQuery(liveEndpoints, keyspace.getName(),
indexQueryPlan, indexStatusMap);
+ initial += liveEndpoints.size();
+ filtered += queryableEndpoints.size();
+ candidates.put(entry.getKey(), queryableEndpoints);
+ }
+
+ if (initial != filtered)
+ check.maybeThrowIndexReadException(initial, filtered,
candidates, indexStatusMap);
+
+ return candidates;
+ }
+
+ @Override
+ ResponseTracker createResponseTracker()
+ {
+ ResponseTracker tracker = super.createResponseTracker();
+
+ // read trackers don't wait on pending nodes
+ Preconditions.checkState(tracker.pendingContacts() == 0);
+ return tracker;
+ }
+
+
+ abstract P replicaPlanBuilder(E candidates, E contacts, E
liveAndDown, int quorumSize);
+
+ ReadReplicaPlanBuilder<E, L, P> replicaPlanBuilder()
+ {
+ switch (cl)
+ {
+ case ONE:
+ case LOCAL_ONE:
+ case NODE_LOCAL:
+ case ANY:
+ case TWO:
+ case THREE:
+ case ALL:
+ return new ReadReplicaPlanBuilder.Simple<>(this);
+
+ case QUORUM:
+ case LOCAL_QUORUM:
+ case EACH_QUORUM:
+ return new ReadReplicaPlanBuilder.DcQuorum<>(this);
+
+ default:
+ throw new IllegalStateException("Unsupported
consistency level: " + cl);
+ }
+ }
+
+ @Override
+ P createReplicaPlan()
+ {
+ ReadReplicaPlanBuilder<E, L, P> builder = replicaPlanBuilder();
+ return builder.build();
+ }
+ }
+
+ static class ForTokenRead extends ForRead<EndpointsForToken,
ReplicaLayout.ForTokenRead, ReplicaPlan.ForTokenRead>
+ {
+ static final Function<ReplicaPlan<?, ?>, ReplicaPlan.ForWrite>
READ_REPAIR_PLAN = (ReplicaPlan<?, ?> self) -> {throw new
IllegalStateException("Cannot create read repair plans for
SatelliteDatacenterReplicationStrategy");};
+ final Token token;
+ final TableId tableId;
+ final SpeculativeRetryPolicy retry;
+ final ReadCoordinator coordinator;
+
+ public ForTokenRead(ClusterMetadata metadata, Keyspace keyspace,
Token token, ConsistencyLevel cl, SatelliteReplicationStrategy strategy, String
primary, ReplicaLayout.ForTokenRead fullLayout, @Nullable Index.QueryPlan
indexQueryPlan, SpeculativeRetryPolicy retry, boolean
throwOnInsufficientLiveReplicas, TableId tableId, ReadCoordinator coordinator)
+ {
+ super(metadata, keyspace, cl, strategy, primary, fullLayout,
indexQueryPlan, retry == AlwaysSpeculativeRetryPolicy.INSTANCE,
throwOnInsufficientLiveReplicas);
+ this.token = token;
+ this.tableId = tableId;
+ this.retry = retry;
+ this.coordinator = coordinator;
+ }
+
+ @Override
+ ReplicaLayout.ForTokenRead filterLayout(ReplicaLayout.ForTokenRead
layout, Predicate<Replica> predicate)
+ {
+ return layout.filter(predicate);
+ }
+
+ @Override
+ ReplicaPlan.ForTokenRead recomputePlan(ClusterMetadata metadata)
+ {
+ return strategy(metadata).planForTokenRead(metadata, keyspace,
tableId, token,
+ indexQueryPlan, cl,
retry, coordinator).replicas();
+ }
+
+ @Override
+ ReplicaPlan.ForTokenRead replicaPlanBuilder(EndpointsForToken
candidates, EndpointsForToken contacts, EndpointsForToken liveAndDown, int
quorumSize)
+ {
+ return new ReplicaPlan.ForTokenRead(keyspace, strategy, cl,
+ candidates, contacts,
liveAndDown,
+ this::recomputePlan,
+ READ_REPAIR_PLAN,
+ metadata.epoch,
+ quorumSize);
+ }
+ }
+
+ static class ForRangeRead extends ForRead<EndpointsForRange,
ReplicaLayout.ForRangeRead, ReplicaPlan.ForRangeRead>
+ {
+ static final BiFunction<ReplicaPlan<?, ?>, Token,
ReplicaPlan.ForWrite> READ_REPAIR_PLAN = (ReplicaPlan<?, ?> self, Token token)
-> {throw new IllegalStateException("Cannot create read repair plans for
SatelliteDatacenterReplicationStrategy");};
+ final AbstractBounds<PartitionPosition> range;
+ final int vnodeCount;
+ final TableId tableId;
+
+ public ForRangeRead(ClusterMetadata metadata, Keyspace keyspace,
AbstractBounds<PartitionPosition> range, int vnodeCount, ConsistencyLevel cl,
SatelliteReplicationStrategy strategy, String primary,
ReplicaLayout.ForRangeRead fullLayout, @Nullable Index.QueryPlan
indexQueryPlan, boolean throwOnInsufficientLiveReplicas, TableId tableId)
+ {
+ super(metadata, keyspace, cl, strategy, primary, fullLayout,
indexQueryPlan, false, throwOnInsufficientLiveReplicas);
+ this.range = range;
+ this.vnodeCount = vnodeCount;
+ this.tableId = tableId;
+ }
+
+ @Override
+ ReplicaLayout.ForRangeRead filterLayout(ReplicaLayout.ForRangeRead
layout, Predicate<Replica> predicate)
+ {
+ return layout.filter(predicate);
+ }
+
+ @Override
+ ReplicaPlan.ForRangeRead recomputePlan(ClusterMetadata metadata)
+ {
+ return strategy(metadata).planForRangeRead(metadata, keyspace,
tableId,
+ indexQueryPlan, cl,
range, vnodeCount).replicas();
+ }
+
+ @Override
+ ReplicaPlan.ForRangeRead replicaPlanBuilder(EndpointsForRange
candidates, EndpointsForRange contacts, EndpointsForRange liveAndDown, int
quorumSize)
+ {
+ return new ReplicaPlan.ForRangeRead(keyspace, strategy, cl,
range,
+ candidates, contacts,
liveAndDown, vnodeCount,
+ this::recomputePlan,
+ READ_REPAIR_PLAN,
+ metadata.epoch,
+ quorumSize);
+ }
+ }
+ }
+
+ static abstract class ResponseTrackerBuilder<E extends Endpoints<E>, L
extends ReplicaLayout<E>, P extends ReplicaPlan<E, P>>
+ {
+ final CoordinationPlanner<E, L, P> planner;
+
+ public ResponseTrackerBuilder(CoordinationPlanner<E, L, P> planner)
+ {
+ this.planner = planner;
+ }
+
+ Map<String, ResponseTracker> responseTrackersForPrimary()
+ {
+ Map<String, ResponseTracker> result =
Maps.newHashMapWithExpectedSize(planner.dcs.size());
+
+ for (String dc : planner.dcs)
+ result.put(dc, planner.createDcResponseTracker(dc));
+
+ return result;
+ }
+
+ abstract ResponseTracker aggregateTrackers(Map<String,
ResponseTracker> dcTrackers);
+
+ ResponseTracker build()
+ {
+ Map<String, ResponseTracker> dcTrackers =
responseTrackersForPrimary();
+ return aggregateTrackers(dcTrackers);
+ }
+
+ static class Simple<E extends Endpoints<E>, L extends
ReplicaLayout<E>, P extends ReplicaPlan<E, P>> extends
ResponseTrackerBuilder<E, L, P>
+ {
+ public Simple(CoordinationPlanner<E, L, P> planner)
+ {
+ super(planner);
+ }
+
+ @Override
+ ResponseTracker aggregateTrackers(Map<String, ResponseTracker>
dcTrackers)
+ {
+ int totalContacts = 0;
+ int totalPending = 0;
+
+
+ List<ResponseTracker> trackers = Lists.newArrayList();
+ Set<String> blockingDatacenters = Sets.newHashSet();
+
+ for (Map.Entry<String, ResponseTracker> entry :
dcTrackers.entrySet())
+ {
+ String dc = entry.getKey();
+ ResponseTracker tracker = entry.getValue();
+ if (planner.cl == ConsistencyLevel.ALL ||
dc.equals(planner.primary) || dc.equals(planner.satellite))
+ {
+ totalContacts += tracker.totalContacts();
Review Comment:
This comment is because including pending replicas is wrong if it's supposed
to be natural replicas. I dug into this way back and it seemed like a legit
mismatch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
