bdeggleston commented on code in PR #4708:
URL: https://github.com/apache/cassandra/pull/4708#discussion_r3443865152
##########
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();
+ totalPending += tracker.pendingContacts();
+ trackers.add(tracker);
+ blockingDatacenters.add(dc);
+ }
+ }
+
+ Predicate<InetAddressAndPort> filter = i -> {
+ String dc =
planner.metadata.locator.location(i).datacenter;
+ return dc != null && blockingDatacenters.contains(dc);
+ };
+
+ Preconditions.checkState(!trackers.isEmpty(), "No blocking DCs
for CL %s (primary=%s, satellite=%s)", planner.cl, planner.primary,
planner.satellite);
+
+
+ if (planner.cl == ConsistencyLevel.ALL)
+ return new SimpleResponseTracker(totalContacts,
totalContacts, filter);
+
+ int blockFor = planner.cl.blockFor(planner.strategy);
+ int totalBlockFor = blockFor + totalPending;
+
+ Preconditions.checkState(blockFor <= totalContacts, "Simple
tracker: blockFor=%s > totalContacts=%s for CL %s", blockFor, totalContacts,
planner.cl);
+
+
+ if (totalPending == 0)
+ return new SimpleResponseTracker(blockFor, totalContacts,
filter);
+
+ Predicate<InetAddressAndPort> pending = i -> {
+ for (ResponseTracker tracker : trackers)
+ {
+ if (tracker.isPending(i))
+ return true;
+ }
+ return false;
+ };
+ return new WriteResponseTracker(blockFor, totalBlockFor,
totalContacts, totalPending, pending, filter);
+ }
+ }
+
+ static class Quorum<E extends Endpoints<E>, L extends
ReplicaLayout<E>, P extends ReplicaPlan<E, P>> extends
ResponseTrackerBuilder<E, L, P>
+ {
+ public Quorum(CoordinationPlanner<E, L, P> planner)
+ {
+ super(planner);
+ }
+
+ @Override
+ ResponseTracker aggregateTrackers(Map<String, ResponseTracker>
dcTrackers)
+ {
+ return new
CompositeTracker(CompositeTracker.quorum(dcTrackers.size()), new
ArrayList<>(dcTrackers.values()));
+ }
+ }
+
+ static class EachQuorum<E extends Endpoints<E>, L extends
ReplicaLayout<E>, P extends ReplicaPlan<E, P>> extends
ResponseTrackerBuilder<E, L, P>
+ {
+ public EachQuorum(CoordinationPlanner<E, L, P> planner)
+ {
+ super(planner);
+ }
+
+ @Override
+ ResponseTracker aggregateTrackers(Map<String, ResponseTracker>
dcTrackers)
+ {
+ return new
CompositeTracker(CompositeTracker.all(dcTrackers.size()), new
ArrayList<>(dcTrackers.values()));
+ }
+ }
+ }
+
+ static abstract class ReadReplicaPlanBuilder<E extends Endpoints<E>, L
extends ReplicaLayout<E>, P extends ReplicaPlan.ForRead<E, P>>
+ implements CoordinationPlanner.IndexReadExceptionCheck<E>
+ {
+
+ private static <E extends Endpoints<E>> Iterable<Replica>
removedSupplier(Map<String, E> initial, Map<String, E> filtered)
+ {
+ return () -> initial.keySet().stream().map(dc -> {
+ Set<InetAddressAndPort> f = filtered.containsKey(dc) ?
filtered.get(dc).endpoints() : Collections.emptySet();
+ return initial.get(dc).without(f);
+ }).flatMap(AbstractReplicaCollection::stream).iterator();
+ }
+
+ final CoordinationPlanner.ForRead<E, L, P> planner;
+
+ public ReadReplicaPlanBuilder(CoordinationPlanner.ForRead<E, L, P>
planner)
+ {
+ this.planner = planner;
+ }
+
+ abstract boolean canSelectReplicasFromDc(String dc, int liveNodes);
+ abstract int minBlockFor();
+ abstract int eligibleCandidates(Map<String, E> candidates);
+
+ abstract boolean haveSufficientContacts(int contactedReplicas, int
contactedDcs);
+ abstract boolean haveSufficientContactsForDc(String dc, int
contactedReplicas, int contactedDcs, int dcContacts);
+ abstract boolean haveSufficientLiveNodes(int contactedReplicas, int
contactedDcs);
+
+ private E allLiveAndDown()
+ {
+ E fullEndpoints = planner.fullLayout.all();
+ ReplicaCollection.Builder<E> fullBuilder =
fullEndpoints.newBuilder(fullEndpoints.size());
+
+ for (String dc : planner.dcs)
+ {
+ E endpoints = planner.fullEndpoints.get(dc);
+ if (endpoints == null)
+ continue;
+
+ fullBuilder.addAll(endpoints);
+ }
+
+ return fullBuilder.build();
+ }
+
+ private E allCandidates(Map<String, E> candidates)
+ {
+ E liveEndpoints = planner.liveLayout.all();
+ ReplicaCollection.Builder<E> candidateBuilder =
liveEndpoints.newBuilder(liveEndpoints.size());
+
+ for (String dc : planner.dcs)
+ {
+ // TODO: make sure the full replica is added if this is the dc
it's from
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || !canSelectReplicasFromDc(dc,
dcCandidates.size()))
+ continue;
+
+ candidateBuilder.addAll(dcCandidates);
+ }
+
+ return candidateBuilder.build();
+ }
+
+ private E contacts(Map<String, E> candidates)
+ {
+ E liveEndpoints = planner.liveLayout.all();
+
+ ReplicaCollection.Builder<E> contactBuilder =
liveEndpoints.newBuilder(liveEndpoints.size());
+
+ // find full replica
+ Replica fullReplica = null;
+ String fullDc = null;
+ for (String dc : planner.dcs)
+ {
+ if (fullReplica != null)
+ break;
+
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || !canSelectReplicasFromDc(dc,
dcCandidates.size()))
+ continue;
+
+ for (Replica replica : dcCandidates)
+ {
+ if (replica.isFull())
+ {
+ fullReplica = replica;
+ fullDc = dc;
+ break;
+ }
+ }
+ }
+
+ if (fullReplica == null)
+ {
+ // No full replicas available - throw error similar to
assureSufficientLiveReplicas
+ throw UnavailableException.create(ConsistencyLevel.ONE,
minBlockFor(), 1, eligibleCandidates(candidates), 0);
+ }
+
+ int totalContacts = 0;
+ int totalContactedDcs = 0;
+
+ // add dc with full replicas first
+ {
+ E dcCandidates = candidates.get(fullDc);
+ Preconditions.checkNotNull(dcCandidates);
+
+ contactBuilder.add(fullReplica);
+ totalContacts++;
+ int dcContacts = 1;
+
+ if (!haveSufficientContacts(totalContacts, totalContactedDcs))
+ {
+ for (Replica replica : dcCandidates)
+ {
+ if (replica.equals(fullReplica))
+ continue;
+
+ if (haveSufficientContactsForDc(fullDc, totalContacts,
totalContactedDcs, dcContacts))
+ break;
+
+ contactBuilder.add(replica);
+ totalContacts++;
+ dcContacts++;
+ }
+ }
+ totalContactedDcs++;
+ }
+
+ for (String dc : planner.dcs)
+ {
+ if (dc.equals(fullDc))
+ continue;
+
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || dcCandidates.isEmpty())
+ continue;
+
+ if (haveSufficientContacts(totalContacts, totalContactedDcs))
+ break;
+
+ int dcContacts = 0;
+ for (Replica replica : dcCandidates)
+ {
+ if (haveSufficientContactsForDc(dc, totalContacts,
totalContactedDcs, dcContacts))
+ break;
+
+ contactBuilder.add(replica);
+ totalContacts++;
+ dcContacts++;
+ }
+
+ totalContactedDcs++;
+ }
+
+ if (planner.throwOnInsufficientLiveReplicas &&
!haveSufficientLiveNodes(totalContacts, totalContactedDcs))
+ {
+ int fullReplicas = 0;
+ for (String dc : planner.dcs)
+ {
+ E dcCandidates = candidates.get(dc);
+ if (dcCandidates == null || !canSelectReplicasFromDc(dc,
dcCandidates.size()))
+ continue;
+ for (Replica replica : dcCandidates)
+ if (replica.isFull())
+ fullReplicas++;
+ }
+
+ throw UnavailableException.create(planner.cl, minBlockFor(),
1, eligibleCandidates(candidates), fullReplicas);
+ }
+
+ E result = contactBuilder.build();
+
+ if (ADDL_CHECKS_ENABLED)
+ {
+ // check for duplicate contacts
+ Set<InetAddressAndPort> seen = new HashSet<>();
+ for (Replica r : result)
+ Preconditions.checkState(seen.add(r.endpoint()),
"Duplicate contact: %s", r.endpoint());
+ }
+
+ return result;
+ }
+
+ public P build()
+ {
+ Map<String, E> candidates = planner.candidates(this);
+
+ E allEndpoints = allLiveAndDown();
+ E allCandidates = allCandidates(candidates);
+ E allContacts = contacts(candidates);
+
+ if (ADDL_CHECKS_ENABLED)
+ {
+ // contacts must be a subset of candidates
+ Set<InetAddressAndPort> candidateSet =
allCandidates.endpoints();
+ for (Replica contact : allContacts)
+
Preconditions.checkState(candidateSet.contains(contact.endpoint()), "Contact %s
not in candidates", contact.endpoint());
+ }
+
+ return planner.replicaPlanBuilder(allCandidates, allContacts,
allEndpoints, minBlockFor());
+ }
+
+ static class Simple<E extends Endpoints<E>, L extends
ReplicaLayout<E>, P extends ReplicaPlan.ForRead<E, P>> extends
ReadReplicaPlanBuilder<E, L, P>
+ {
+ private final int blockFor;
+ private final int targetContacts;
+
+ public Simple(CoordinationPlanner.ForRead<E, L, P> planner)
+ {
+ super(planner);
+
+ if (planner.cl == ConsistencyLevel.ALL)
+ {
+ int required = 0;
+ for (E replicas : planner.fullEndpoints.values())
Review Comment:
It's a naming thing, witnesses are included.
Full in this context means "not filtered for liveness". I'll rename to
`liveAndDown` for consistency with other parts of the repo
--
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]
