diff --git a/modules/accord b/modules/accord index a575ab316b..a681b1bf7d 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit a575ab316b4e79f99009801312579bcbf86451ef +Subproject commit a681b1bf7d9e8d1d797af2fd54bdd180a4863406 diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index e48a945393..b32a166b8a 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -56,6 +56,8 @@ import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.Routable; import accord.primitives.RoutingKeys; +import accord.topology.ActiveEpoch; +import accord.topology.ActiveEpochs; import accord.topology.Shard; import accord.topology.Topology; import accord.utils.SortedListMap; @@ -2166,7 +2168,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace public void collect(PartitionsCollector collector) { IAccordService service = AccordService.unsafeInstance(); - List snapshot = service.node().topology().topologySnapshot(); + ActiveEpochs snapshot = service.node().topology().active(); Map>> tableIdLookup = new HashMap<>(); { @@ -2177,8 +2179,9 @@ public class AccordDebugKeyspace extends VirtualKeyspace TableId prevTableId = null; Map> startLookup = null; - for (Topology topology : snapshot) + for (ActiveEpoch epoch : snapshot) { + Topology topology = epoch.global(); for (Shard shard : topology.shards()) { Range range = shard.range; diff --git a/src/java/org/apache/cassandra/db/virtual/AccordVirtualTables.java b/src/java/org/apache/cassandra/db/virtual/AccordVirtualTables.java index 0ca2caf838..8dbba15e33 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordVirtualTables.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordVirtualTables.java @@ -32,9 +32,9 @@ import com.google.common.collect.Sets; import accord.primitives.Range; import accord.primitives.Ranges; -import accord.topology.TopologyManager.EpochsSnapshot; -import accord.topology.TopologyManager.EpochsSnapshot.Epoch; -import accord.topology.TopologyManager.EpochsSnapshot.EpochReady; +import accord.topology.ActiveEpoch; +import accord.topology.ActiveEpochs; +import accord.topology.EpochReady; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; import org.apache.cassandra.db.marshal.LongType; @@ -46,8 +46,6 @@ import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.accord.TokenRange; -import static accord.topology.TopologyManager.EpochsSnapshot.ResultStatus.SUCCESS; - public class AccordVirtualTables { public static final String EPOCHS = "accord_epochs"; @@ -79,7 +77,7 @@ public class AccordVirtualTables { super(parse(keyspace, "CREATE TABLE " + EPOCHS + " (\n" + " epoch bigint PRIMARY KEY,\n" + - " ready_metadata text,\n" + + " ready_active text,\n" + " ready_coordinate text,\n" + " ready_data text,\n" + " ready_reads text,\n" + @@ -94,16 +92,16 @@ public class AccordVirtualTables public DataSet data() { SimpleDataSet ds = new SimpleDataSet(metadata()); - EpochsSnapshot snapshot = epochsSnapshot(); - for (Epoch epoch : snapshot) + ActiveEpochs snapshot = AccordService.instance().topology().active(); + for (ActiveEpoch epoch : snapshot) { - ds.row(epoch.epoch); - EpochReady ready = epoch.ready; - ds.column("ready_metadata", ready.metadata.value); - ds.column("ready_coordinate", ready.coordinate.value); - ds.column("ready_data", ready.data.value); - ds.column("ready_reads", ready.reads.value); - ds.column("ready", ready.reads == SUCCESS); + ds.row(epoch.epoch()); + EpochReady ready = epoch.epochReady(); + ds.column("ready_active", ready.active.toString()); + ds.column("ready_coordinate", ready.coordinate.toString()); + ds.column("ready_data", ready.data.toString()); + ds.column("ready_reads", ready.reads.toString()); + ds.column("ready", ready.active.isDone() && ready.coordinate.isDone() && ready.data.isDone() && ready.reads.isDone()); } return ds; } @@ -133,21 +131,21 @@ public class AccordVirtualTables public DataSet data() { SimpleDataSet ds = new SimpleDataSet(metadata()); - EpochsSnapshot snapshot = epochsSnapshot(); - for (Epoch state : snapshot) + ActiveEpochs snapshot = AccordService.instance().topology().active(); + for (ActiveEpoch state : snapshot) { Map> addedRanges = groupByTable(state.addedRanges); Map> removedRanges = groupByTable(state.removedRanges); - Map> synced = groupByTable(state.synced); - Map> closed = groupByTable(state.closed); - Map> retired = groupByTable(state.retired); + Map> synced = groupByTable(state.quorumReady()); + Map> closed = groupByTable(state.closed()); + Map> retired = groupByTable(state.retired()); Set allTables = union(addedRanges.keySet(), removedRanges.keySet(), synced.keySet(), closed.keySet(), retired.keySet()); for (TableId table : allTables) { TableMetadata metadata = Schema.instance.getTableMetadata(table); if (metadata == null) continue; // table dropped, ignore - ds.row(state.epoch, metadata.keyspace, metadata.name); + ds.row(state.epoch(), metadata.keyspace, metadata.name); ds.column("added", format(addedRanges.get(table))); ds.column("removed", format(removedRanges.get(table))); @@ -179,11 +177,6 @@ public class AccordVirtualTables } } - private static EpochsSnapshot epochsSnapshot() - { - return AccordService.instance().topology().epochsSnapshot(); - } - private static String toStringNoTable(TokenRange tr) { // TokenRange extends Range.EndInclusive diff --git a/src/java/org/apache/cassandra/service/Rebuild.java b/src/java/org/apache/cassandra/service/Rebuild.java index 56ad1f47ba..feb782ddcb 100644 --- a/src/java/org/apache/cassandra/service/Rebuild.java +++ b/src/java/org/apache/cassandra/service/Rebuild.java @@ -33,7 +33,7 @@ import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.EpochReady; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.dht.Range; diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java deleted file mode 100644 index bca1cad65f..0000000000 --- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java +++ /dev/null @@ -1,654 +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.accord; - -import java.util.HashSet; -import java.util.Map; -import java.util.Objects; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executor; -import java.util.concurrent.TimeUnit; -import java.util.function.BiConsumer; -import java.util.stream.Collectors; -import javax.annotation.Nullable; -import javax.annotation.concurrent.GuardedBy; - -import com.google.common.annotations.VisibleForTesting; -import com.google.common.collect.Sets; - -import accord.api.Agent; -import accord.api.TopologySorter; -import accord.coordinate.tracking.NotifyTracker; -import accord.coordinate.tracking.RequestStatus; -import accord.impl.AbstractConfigurationService; -import accord.local.Node; -import accord.primitives.Ranges; -import accord.topology.Topologies; -import accord.topology.Topology; -import accord.utils.Invariants; -import accord.utils.async.AsyncResult; -import accord.utils.async.AsyncResults; -import org.agrona.collections.LongArrayList; -import org.apache.cassandra.concurrent.ScheduledExecutorPlus; -import org.apache.cassandra.concurrent.ScheduledExecutors; -import org.apache.cassandra.concurrent.Shutdownable; -import org.apache.cassandra.concurrent.Stage; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.net.MessageDelivery; -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.repair.SharedContext; -import org.apache.cassandra.tcm.ClusterMetadata; -import org.apache.cassandra.tcm.ClusterMetadataService; -import org.apache.cassandra.tcm.Epoch; -import org.apache.cassandra.tcm.membership.NodeState; -import org.apache.cassandra.utils.FBUtilities; -import org.apache.cassandra.utils.Simulate; -import org.apache.cassandra.utils.concurrent.AsyncPromise; -import org.apache.cassandra.utils.concurrent.Future; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static org.apache.cassandra.service.accord.AccordTopology.tcmIdToAccord; -import static org.apache.cassandra.utils.Simulate.With.MONITORS; - -@Simulate(with=MONITORS) -public class AccordConfigurationService extends AbstractConfigurationService implements AccordSyncPropagator.Listener, Shutdownable -{ - public static final Logger logger = LoggerFactory.getLogger(AccordConfigurationService.class); - - private final AccordSyncPropagator syncPropagator; - private final WatermarkCollector watermarkCollector; - private final AccordEndpointMapper endpointMapper; - - private enum State { INITIALIZED, STARTED, SHUTDOWN } - - @GuardedBy("this") - private State state = State.INITIALIZED; - - public enum SyncStatus { NOT_STARTED, NOTIFYING, COMPLETED } - - static class EpochState extends AbstractConfigurationService.AbstractEpochState - { - private NotifyTracker notifyTracker; - private volatile SyncStatus syncStatus = SyncStatus.NOT_STARTED; - protected final AsyncResult.Settable localSyncNotified = AsyncResults.settable(); - - public EpochState(long epoch) - { - super(epoch); - } - - void ack(Node.Id id) - { - if (notifyTracker != null && RequestStatus.Success == notifyTracker.recordSuccess(id)) - doneNotifying(); - } - - void nack(Node.Id id) - { - if (notifyTracker != null && RequestStatus.Success == notifyTracker.recordFailure(id)) - doneNotifying(); - } - - void startNotifying() - { - syncStatus = SyncStatus.NOTIFYING; - notifyTracker = new NotifyTracker(new Topologies.Single((TopologySorter) null, topology)); - } - - void doneNotifying() - { - syncStatus = SyncStatus.COMPLETED; - localSyncNotified.trySuccess(null); - notifyTracker = null; - } - - AsyncResult received() - { - return received; - } - - AsyncResult acknowledged() - { - return acknowledged; - } - - @Nullable AsyncResult reads() - { - return ready == null ? null : ready.reads; - } - - AsyncResult.Settable localSyncNotified() - { - return localSyncNotified; - } - - public void addDebugString(StringBuilder sb) - { - sb.append(" syncStatus ") - .append(syncStatus) - .append(" localSyncNotified ") - .append(localSyncNotified); - super.addDebugString(sb); - } - } - - static class EpochHistory extends AbstractConfigurationService.AbstractEpochHistory - { - @Override - protected EpochState createEpochState(long epoch) - { - return new EpochState(epoch); - } - } - - public AccordConfigurationService(Node.Id node, Agent agent, AccordEndpointMapper endpointMapper, MessageDelivery messagingService, ScheduledExecutorPlus scheduledTasks) - { - super(node, agent); - this.syncPropagator = new AccordSyncPropagator(localId, endpointMapper, messagingService, scheduledTasks, this); - this.watermarkCollector = new WatermarkCollector(); - this.endpointMapper = endpointMapper; - listeners.add(watermarkCollector); - } - - public AccordConfigurationService(Node.Id node, Agent agent) - { - this(node, agent, new EndpointMapping.Updateable(), MessagingService.instance(), ScheduledExecutors.scheduledTasks); - } - - @Override - protected EpochHistory createEpochHistory() - { - return new EpochHistory(); - } - - /** - * On restart, loads topologies. On bootstrap, discovers existing topologies and initializes the node. - */ - public synchronized void start() - { - Invariants.require(state == State.INITIALIZED, "Expected state to be INITIALIZED but was %s", state); - state = State.STARTED; - endpointMapper.removedNodes().forEach((removed, removedIn) -> { - onNodeRemoved(removedIn, new Node.Id(removed.id)); - }); - } - - @Override - public synchronized boolean isTerminated() - { - return state == State.SHUTDOWN; - } - - @Override - public synchronized void shutdown() - { - if (isTerminated()) - return; - state = State.SHUTDOWN; - } - - @Override - public Object shutdownNow() - { - shutdown(); - return null; - } - - @Override - public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException - { - return isTerminated(); - } - - public synchronized void updateMapping(ClusterMetadata metadata) - { - endpointMapper.updateMapping(metadata); - } - - private void reportMetadata(ClusterMetadata metadata) - { - Stage.MISC.submit(() -> reportMetadataInternal(metadata)); - } - - void reportMetadataInternal(ClusterMetadata metadata) - { - Topology topology = AccordTopology.createAccordTopology(metadata); - - updateMapping(metadata); - reportTopology(topology); - Set stillLiveNodes = metadata.directory.states.entrySet() - .stream() - .filter(e -> e.getValue() != NodeState.LEFT && e.getValue() != NodeState.LEAVING) - .map(e -> tcmIdToAccord(e.getKey())) - .collect(Collectors.toSet()); - if (epochs.lastAcknowledged() >= topology.epoch()) checkIfNodesRemoved(topology, stillLiveNodes); - else epochs.acknowledgeFuture(topology.epoch()).invokeIfSuccess(() -> checkIfNodesRemoved(topology, stillLiveNodes)); - } - - private void checkIfNodesRemoved(Topology topology, Set stillLiveNodes) - { - long minEpoch = epochs.minEpoch(); - if (minEpoch == 0 || topology.epoch() <= minEpoch) return; - Topology previous = getTopologyForEpoch(topology.epoch() - 1); - // for all nodes removed, or pending removal, mark them as removed so we don't wait on their replies - Set removedNodes = Sets.difference(previous.nodes(), topology.nodes()); - removedNodes = Sets.filter(removedNodes, id -> !stillLiveNodes.contains(id)); - // TODO (desired, efficiency): there should be no need to notify every epoch for every removed node - for (Node.Id removedNode : removedNodes) - { - if (topology.epoch() >= minEpoch) - onNodeRemoved(topology.epoch(), removedNode); - } - } - - public void onNodeRemoved(long epoch, Node.Id removed) - { - syncPropagator.onNodesRemoved(removed); - // TODO (desired): this is test only; make integration cleaner - for (long oldEpoch : nonCompletedEpochsBefore(epoch)) - receiveRemoteSyncCompletePreListenerNotify(removed, oldEpoch); - - listeners.forEach(l -> l.onRemoveNode(epoch, removed)); - } - - private long[] nonCompletedEpochsBefore(long max) - { - LongArrayList notComplete = new LongArrayList(); - synchronized (epochs) - { - for (long epoch = epochs.minEpoch(), maxKnown = epochs.maxEpoch(); epoch <= max && epoch <= maxKnown; epoch++) - { - EpochSnapshot snapshot = getEpochSnapshot(epoch); - if (snapshot.syncStatus != SyncStatus.COMPLETED) - notComplete.add(epoch); - } - } - return notComplete.toLongArray(); - } - - @VisibleForTesting - void maybeReportMetadata(ClusterMetadata metadata) - { - // don't report metadata until the previous one has been acknowledged - long epoch = metadata.epoch.getEpoch(); - synchronized (epochs) - { - // Accord has never been enabled for this cluster. - if (epochs.isEmpty() && !metadata.schema.hasAccordKeyspaces()) - return; - - // On first boot, we have 2 options: - // - // - we can start listening to TCM _before_ we replay topologies - // - we can start listening to TCM _after_ we replay topologies - // - // If we start listening to TCM _before_ we replay topologies from other nodes, - // we may end up in a situation where TCM reports metadata that would create an - // `epoch - 1` epoch state that is not associated with any topologies, and - // therefore should not be listened upon. - // - // If we start listening to TCM _after_ we replay topologies, we may end up in a - // situation where TCM reports metadata that is 1 (or more) epochs _ahead_ of the - // last known epoch. Previous implementations were using TCM peer catch up, which - // could have resulted in gaps. - // - // Current protocol solves both problems by _first_ replaying topologies form peers, - // then subscribing to TCM _and_, if there are still any gaps, filling them again. - // However, it still has a slight chance of creating an `epoch - 1` epoch state - // not associated with any topologies, which under "right" circumstances could - // have been waited upon with `epochReady`. This check precludes creation of this - // epoch: by the time this code can be called, remote topology replay is already - // done, so TCM listener will only report epochs that are _at least_ min epoch. - if (epochs.maxEpoch() == 0 || epochs.minEpoch() == metadata.epoch.getEpoch()) - { - getOrCreateEpochState(epoch); // touch epoch state so subsequent calls see it - reportMetadata(metadata); - return; - } - } - - getOrCreateEpochState(epoch - 1).acknowledged().invokeIfSuccess(() -> reportMetadata(metadata)); - } - - private final Map> pendingTopologies = new ConcurrentHashMap<>(); - - @Override - public void fetchTopologyForEpoch(long epoch) - { - long minEpoch = currentEpoch() + 1; - // Find and fetch all epochs in-between - for (long i = minEpoch; i <= epoch; ++i) - fetchTopologyInternal(i); - } - - private static final Object SUCCESS = new Object(); - - protected void fetchTopologyInternal(long epoch) - { - pendingTopologies.computeIfAbsent(epoch, (epoch_) -> { - AsyncPromise future = new AsyncPromise<>(); - fetchTopologyAsync(epoch_, - (success, throwable) -> { - Future removed = pendingTopologies.remove(epoch_); - Invariants.require(future == removed, "%s should be equal to %s", future, removed); - if (success != null) - future.setSuccess(null); - else - { - future.setFailure(Invariants.nonNull(throwable)); - fetchTopologyForEpoch(epoch_); - } - }); - return future; - }); - } - - private void fetchTopologyAsync(long epoch, BiConsumer onResult) - { - // It's not safe for this to block on CMS so for now pick a thread pool to handle it - Stage.ACCORD_MIGRATION.execute(() -> { - try - { - if (ClusterMetadata.current().epoch.getEpoch() < epoch) - ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(epoch)); - } - catch (Throwable t) - { - onResult.accept(null, t); - return; - } - - // In most cases, after fetching log from CMS, we will be caught up to the required epoch. - // This TCM will also notify Accord via reportMetadata, so we do not need to fetch topologies. - // If metadata has reported has skipped one or more epochs, and is _ahead_ of the requested epoch, - // we need to fetch topologies from peers to fill in the gap. - ClusterMetadata metadata = ClusterMetadata.current(); - if (metadata.epoch.getEpoch() == epoch) - { - onResult.accept(SUCCESS, null); - return; - } - - Set peers = new HashSet<>(metadata.directory.allJoinedEndpoints()); - peers.remove(FBUtilities.getBroadcastAddressAndPort()); - if (peers.isEmpty()) - { - onResult.accept(SUCCESS, null); - return; - } - - // Fetching only one epoch here since later epochs might have already been requested concurrently - FetchTopologies.fetch(SharedContext.Global.instance, peers, epoch, epoch) - .addCallback((topologyRange, t) -> { - if (t != null) - { - if (currentEpoch() >= epoch) - onResult.accept(SUCCESS, null); - else - onResult.accept(null, t); - } - else - { - topologyRange.forEach(this::reportTopology, epoch, 1); - onResult.accept(SUCCESS, null); - } - }); - }); - } - - @Override - protected Executor executor() - { - return Stage.ACCORD_MIGRATION::execute; - } - - @Override - public void reportTopology(Topology topology) - { - long tcmEpoch = ClusterMetadata.current().epoch.getEpoch(); - Invariants.require(topology.epoch() <= tcmEpoch, - "Reported topology %s not known to TCM", topology.epoch(), tcmEpoch); - super.reportTopology(topology); - } - - @Override - protected void onReadyToCoordinate(Topology topology) - { - long epoch = topology.epoch(); - EpochState epochState = getOrCreateEpochState(epoch); - synchronized (this) - { - if (epochState.syncStatus != SyncStatus.NOT_STARTED) - return; - - if (topology.nodes().contains(localId)) epochState.startNotifying(); - else epochState.doneNotifying(); - } - - syncPropagator.reportCoordinationReady(epoch, topology.nodes(), localId); - } - - @Override - public synchronized void onEndpointAck(Node.Id id, long epoch) - { - if (epochs.wasTruncated(epoch)) - return; - - EpochState epochState = getOrCreateEpochState(epoch); - synchronized (this) - { - epochState.ack(id); - } - } - - @Override - public synchronized void onEndpointNack(Node.Id id, long epoch) - { - if (epochs.wasTruncated(epoch)) - return; - - EpochState epochState = getOrCreateEpochState(epoch); - synchronized (this) - { - epochState.nack(id); - } - } - - @Override - public void reportEpochClosed(Ranges ranges, long epoch) - { - checkStarted(); - EpochHistory epochs = this.epochs; - if (epoch < minEpoch() || epochs.wasTruncated(epoch)) - return; - - Topology topology = getTopologyForEpoch(epoch); - if (topology != null) - syncPropagator.reportClosed(epoch, topology.nodes(), ranges); - } - - @VisibleForTesting - public AccordSyncPropagator syncPropagator() - { - return syncPropagator; - } - - @Override - public void reportEpochRetired(Ranges ranges, long epoch) - { - if (epochs.wasTruncated(epoch)) - return; - - checkStarted(); - // TODO (expected): ensure we aren't fetching a truncated epoch; otherwise this should be non-null - Topology topology = getTopologyForEpoch(epoch); - if (topology != null) - syncPropagator.reportRetired(epoch, topology.nodes(), ranges); - } - - @Override - public void receiveClosed(Ranges ranges, long epoch) - { - super.receiveClosed(ranges, epoch); - } - - @Override - public void receiveRetired(Ranges ranges, long epoch) - { - super.receiveRetired(ranges, epoch); - } - - @Override - public void reportEpochRemoved(long epoch) - { - logger.info("Epoch removed, truncated epochs until {}", epoch); - epochs.truncateUntil(epoch); - } - - private synchronized void checkStarted() - { - State state = this.state; - Invariants.require(state == State.STARTED, "Expected state to be STARTED but was %s", state); - } - - public WatermarkCollector watermarkCollector() - { - return watermarkCollector; - } - - public AccordEndpointMapper endpointMapper() - { - return endpointMapper; - } - - @VisibleForTesting - public static class EpochSnapshot - { - public enum ResultStatus - { - PENDING, SUCCESS, FAILURE; - - static ResultStatus of(AsyncResult result) - { - if (result == null || !result.isDone()) - return PENDING; - - return result.isSuccess() ? SUCCESS : FAILURE; - } - } - - public final long epoch; - public final SyncStatus syncStatus; - public final ResultStatus received; - public final ResultStatus acknowledged; - public final ResultStatus reads; - - private EpochSnapshot(EpochState state) - { - this.epoch = state.epoch(); - this.syncStatus = state.syncStatus; - this.received = ResultStatus.of(state.received()); - this.acknowledged = ResultStatus.of(state.acknowledged()); - this.reads = ResultStatus.of(state.reads()); - } - - public EpochSnapshot(long epoch, SyncStatus syncStatus, ResultStatus received, ResultStatus acknowledged, ResultStatus reads) - { - this.epoch = epoch; - this.syncStatus = syncStatus; - this.received = received; - this.acknowledged = acknowledged; - this.reads = reads; - } - - public boolean equals(Object o) - { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - EpochSnapshot that = (EpochSnapshot) o; - return epoch == that.epoch && syncStatus == that.syncStatus && received == that.received && acknowledged == that.acknowledged && reads == that.reads; - } - - public int hashCode() - { - return Objects.hash(epoch, syncStatus, received, acknowledged, reads); - } - - public String toString() - { - return "EpochSnapshot{" + - "epoch=" + epoch + - ", syncStatus=" + syncStatus + - ", received=" + received + - ", acknowledged=" + acknowledged + - ", reads=" + reads + - '}'; - } - - public static EpochSnapshot completed(long epoch) - { - return new EpochSnapshot(epoch, SyncStatus.COMPLETED, ResultStatus.SUCCESS, ResultStatus.SUCCESS, ResultStatus.SUCCESS); - } - } - - @VisibleForTesting - public EpochSnapshot getEpochSnapshot(long epoch) - { - EpochState state; - // If epoch truncate happens then getting the epoch again will recreate an empty one - synchronized (epochs) - { - if (epoch < epochs.minEpoch() || epoch > epochs.maxEpoch()) - return null; - - state = getOrCreateEpochState(epoch); - } - return new EpochSnapshot(state); - } - - @VisibleForTesting - public long minEpoch() - { - return epochs.minEpoch(); - } - - @VisibleForTesting - public long maxEpoch() - { - return epochs.maxEpoch(); - } - - /** - * The callback is resolved while holding the object lock, which can cause the future chain to resolve while also - * holding the lock! This behavior is exposed for tests and is unsafe due to the lock behind held while resolving - * the callback - */ - @VisibleForTesting - public Future unsafeLocalSyncNotified(long epoch) - { - AsyncPromise promise = new AsyncPromise<>(); - getOrCreateEpochState(epoch).localSyncNotified().invoke((result, failure) -> { - if (failure != null) promise.tryFailure(failure); - else promise.trySuccess(result); - }); - return promise; - } -} diff --git a/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java b/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java index 12a706a442..fb93d9d4b9 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java @@ -18,9 +18,8 @@ package org.apache.cassandra.service.accord; -import accord.api.ConfigurationService; +import accord.api.TopologyListener; import accord.local.Node; -import accord.primitives.Ranges; import accord.topology.Topology; import accord.utils.Invariants; import accord.utils.async.AsyncResult; @@ -50,7 +49,7 @@ import java.util.concurrent.TimeUnit; /** * Listens to availability status of peers and updates tcm fast path data accordingly */ -public abstract class AccordFastPathCoordinator implements ChangeListener, ConfigurationService.Listener +public abstract class AccordFastPathCoordinator implements ChangeListener, TopologyListener { private static final AsyncResult SUCCESS = AsyncResults.success(null); @@ -182,12 +181,11 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi private static class Impl extends AccordFastPathCoordinator implements IEndpointStateChangeSubscriber { - private final AccordConfigurationService configService; - - public Impl(Node.Id localId, AccordConfigurationService configService) + final AccordEndpointMapper endpointMapper; + public Impl(Node.Id localId, AccordEndpointMapper endpointMapper) { super(localId); - this.configService = configService; + this.endpointMapper = endpointMapper; } @Override @@ -201,7 +199,7 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi { Gossiper.instance.register(this); StorageService.instance.addPreShutdownHook(this::onShutdown); - configService.registerListener(this); + AccordService.unsafeInstance().topology().addListener(this); } @Override @@ -222,21 +220,21 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi @Override public void onAlive(InetAddressAndPort endpoint, EndpointState state) { - Node.Id node = configService.endpointMapper().mappedIdOrNull(endpoint); + Node.Id node = endpointMapper.mappedIdOrNull(endpoint); if (node != null) onAlive(node); } @Override public void onDead(InetAddressAndPort endpoint, EndpointState state) { - Node.Id node = configService.endpointMapper().mappedIdOrNull(endpoint); + Node.Id node = endpointMapper.mappedIdOrNull(endpoint); if (node != null) onDead(node); } } - public static AccordFastPathCoordinator create(Node.Id localId, AccordConfigurationService configService) + public static AccordFastPathCoordinator create(Node.Id localId, AccordEndpointMapper endpointMapper) { - return new Impl(localId, configService); + return new Impl(localId, endpointMapper); } synchronized void maybeUpdateFastPath(Node.Id node, Status status) @@ -335,13 +333,8 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi } @Override - public AsyncResult onTopologyUpdate(Topology topology) + public void onReceived(Topology topology) { updatePeers(topology); - return SUCCESS; } - - @Override public void onRemoteSyncComplete(Node.Id node, long epoch) {} - @Override public void onEpochClosed(Ranges ranges, long epoch) {} - @Override public void onEpochRetired(Ranges ranges, long epoch) {} } diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index fb2bd93d97..bcf58650a9 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -38,7 +38,8 @@ import javax.annotation.concurrent.GuardedBy; import com.google.common.annotations.VisibleForTesting; import com.google.common.primitives.Ints; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.ActiveEpochs; +import accord.topology.EpochReady; import accord.primitives.Txn; import org.apache.cassandra.metrics.AccordReplicaMetrics; import org.apache.cassandra.metrics.AccordSystemMetrics; @@ -53,7 +54,6 @@ import accord.api.ProtocolModifiers; import accord.coordinate.CoordinateMaxConflict; import accord.coordinate.CoordinateTransaction; import accord.coordinate.KeyBarriers; -import accord.impl.AbstractConfigurationService; import accord.impl.DefaultLocalListeners; import accord.impl.DefaultRemoteListeners; import accord.impl.RequestCallbacks; @@ -77,6 +77,7 @@ import accord.primitives.TxnId; import accord.topology.Shard; import accord.topology.Topology; import accord.topology.TopologyManager; +import accord.topology.TopologyRange; import accord.utils.DefaultRandom; import accord.utils.Invariants; import accord.utils.async.AsyncChain; @@ -133,7 +134,6 @@ import static accord.local.durability.DurabilityService.SyncRemote.All; import static accord.messages.SimpleReply.Ok; import static accord.primitives.Txn.Kind.ExclusiveSyncPoint; import static accord.primitives.Txn.Kind.Write; -import static accord.topology.TopologyManager.TopologyRange; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.cassandra.config.DatabaseDescriptor.getAccordCommandStoreShardCount; @@ -219,7 +219,8 @@ public class AccordService implements IAccordService, Shutdownable private final Node node; private final Shutdownable nodeShutdown; private final AccordMessageSink messageSink; - private final AccordConfigurationService configService; + private final AccordEndpointMapper endpointMapper; + private final AccordTopologyService topologyService; private final AccordFastPathCoordinator fastPathCoordinator; private final AccordScheduler scheduler; private final AccordDataStore dataStore; @@ -258,7 +259,7 @@ public class AccordService implements IAccordService, Shutdownable { if (!isSetup()) return ignore -> {}; AccordService i = (AccordService) instance(); - return i.configService().watermarkCollector().handler; + return i.topologyService().watermarkCollector().handler; } public static IVerbHandler requestHandlerOrNoop() @@ -300,8 +301,6 @@ public class AccordService implements IAccordService, Shutdownable as.finishInitialization(); - as.configService.start(); - as.configService.unsafeMarkTruncated(); as.fastPathCoordinator.start(); ClusterMetadataService.instance().log().addListener(as.fastPathCoordinator); @@ -311,7 +310,6 @@ public class AccordService implements IAccordService, Shutdownable as.node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS); as.state = State.STARTED; // Only enable durability scheduling _after_ we have fully replayed journal - as.configService.registerListener(as.node.durability()); as.node.durability().start(); instance = as; @@ -320,7 +318,7 @@ public class AccordService implements IAccordService, Shutdownable AccordSystemMetrics.touch(); AccordViolationHandler.setup(); - WatermarkCollector.fetchAndReportWatermarksAsync(as.configService); + WatermarkCollector.fetchAndReportWatermarksAsync(as.topology()); return as; } @@ -383,12 +381,13 @@ public class AccordService implements IAccordService, Shutdownable this.scheduler = new AccordScheduler(); this.dataStore = new AccordDataStore(); this.journal = new AccordJournal(DatabaseDescriptor.getAccord().journal); - this.configService = new AccordConfigurationService(localId, agent); - this.fastPathCoordinator = AccordFastPathCoordinator.create(localId, configService); - this.messageSink = new AccordMessageSink(agent, configService.endpointMapper(), callbacks); + this.endpointMapper = new EndpointMapping.Updateable(); + this.topologyService = new AccordTopologyService(localId, endpointMapper); + this.fastPathCoordinator = AccordFastPathCoordinator.create(localId, endpointMapper); + this.messageSink = new AccordMessageSink(agent, endpointMapper, callbacks); this.node = new Node(localId, messageSink, - configService, + topologyService, time, new AtomicUniqueTimeWithStaleReservation(time), () -> dataStore, new KeyspaceSplitter(new EvenSplit<>(getAccordCommandStoreShardCount(), getPartitioner().accordSplitter())), @@ -396,18 +395,18 @@ public class AccordService implements IAccordService, Shutdownable new DefaultRandom(), scheduler, CompositeTopologySorter.create(SizeOfIntersectionSorter.SUPPLIER, - new AccordTopologySorter.Supplier(configService.endpointMapper(), DatabaseDescriptor.getNodeProximity())), + new AccordTopologySorter.Supplier(endpointMapper, DatabaseDescriptor.getNodeProximity())), DefaultRemoteListeners::new, ignore -> callbacks, DefaultProgressLogs::new, DefaultLocalListeners.Factory::new, AccordCommandStores.factory(), - new AccordInteropFactory(configService.endpointMapper()), + new AccordInteropFactory(endpointMapper), journal.durableBeforePersister(), journal); this.nodeShutdown = toShutdownable(node); - this.requestHandler = new AccordVerbHandler<>(node, configService.endpointMapper()); - this.responseHandler = new AccordResponseVerbHandler<>(callbacks, configService.endpointMapper()); + this.requestHandler = new AccordVerbHandler<>(node, endpointMapper); + this.responseHandler = new AccordResponseVerbHandler<>(callbacks, endpointMapper); } @Override @@ -419,7 +418,7 @@ public class AccordService implements IAccordService, Shutdownable node.load(); ClusterMetadata metadata = ClusterMetadata.current(); - configService.updateMapping(metadata); + endpointMapper.updateMapping(metadata); List images = journal.replayTopologies(); if (!images.isEmpty()) @@ -435,7 +434,7 @@ public class AccordService implements IAccordService, Shutdownable // Replay local epochs for (TopologyUpdate image : images) - configService.reportTopology(image.global); + node.topology().reportTopology(image.global); } } } @@ -448,17 +447,20 @@ public class AccordService implements IAccordService, Shutdownable @VisibleForTesting public void finishInitialization() { - configService.updateMapping(ClusterMetadata.current()); + endpointMapper.updateMapping(ClusterMetadata.current()); + TopologyManager topology = node.topology(); long highestKnown = -1; - if (configService.currentTopology() != null) - highestKnown = configService.currentEpoch(); + if (!topology.active().isEmpty()) + { + highestKnown = topology.active().epoch(); + } try { TopologyRange remote = fetchTopologies(highestKnown + 1); if (remote != null) // TODO (required): if remote.min > highestKnown + 1, should we decide if we need to truncate our local topologies? Probably not until startup has finished. { - remote.forEach(configService::reportTopology, remote.min, Integer.MAX_VALUE); + remote.forEach(node.topology()::reportTopology, remote.min, Integer.MAX_VALUE); if (remote.current > highestKnown) highestKnown = remote.current; } @@ -470,7 +472,7 @@ public class AccordService implements IAccordService, Shutdownable public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) { if (state != State.SHUTDOWN) - configService.maybeReportMetadata(next); + maybeReportMetadata(next); } }); @@ -481,7 +483,7 @@ public class AccordService implements IAccordService, Shutdownable for (ClusterMetadata item : preinit.getItems()) { if (item.epoch.getEpoch() > highestKnown) - configService.maybeReportMetadata(item); + maybeReportMetadata(item); } } catch (InterruptedException e) @@ -494,6 +496,45 @@ public class AccordService implements IAccordService, Shutdownable throw new RuntimeException(e); } } + + private void maybeReportMetadata(ClusterMetadata metadata) + { + endpointMapper.updateMapping(metadata); + + ActiveEpochs epochs = topology().active(); + if (!metadata.schema.hasAccordKeyspaces() && epochs.isEmpty()) + return; + + if (epochs.epoch() >= metadata.epoch.getEpoch()) + return; + + // On first boot, we have 2 options: + // + // - we can start listening to TCM _before_ we replay topologies + // - we can start listening to TCM _after_ we replay topologies + // + // If we start listening to TCM _before_ we replay topologies from other nodes, + // we may end up in a situation where TCM reports metadata that would create an + // `epoch - 1` epoch state that is not associated with any topologies, and + // therefore should not be listened upon. + // + // If we start listening to TCM _after_ we replay topologies, we may end up in a + // situation where TCM reports metadata that is 1 (or more) epochs _ahead_ of the + // last known epoch. Previous implementations were using TCM peer catch up, which + // could have resulted in gaps. + // + // Current protocol solves both problems by _first_ replaying topologies form peers, + // then subscribing to TCM _and_, if there are still any gaps, filling them again. + // However, it still has a slight chance of creating an `epoch - 1` epoch state + // not associated with any topologies, which under "right" circumstances could + // have been waited upon with `epochReady`. This check precludes creation of this + // epoch: by the time this code can be called, remote topology replay is already + // done, so TCM listener will only report epochs that are _at least_ min epoch. + Topology topology = AccordTopology.createAccordTopology(metadata); + topology().reportTopology(topology); + } + + /** * Queries peers to discover min epoch, and then fetches all topologies between min and current epochs */ @@ -723,10 +764,9 @@ public class AccordService implements IAccordService, Shutdownable @Override public long currentEpoch() { - return configService.currentEpoch(); + return topology().epoch(); } - @Override public TopologyManager topology() { @@ -859,7 +899,7 @@ public class AccordService implements IAccordService, Shutdownable private List shutdownableSubsystems() { - return Arrays.asList(scheduler, nodeShutdown, journal, configService); + return Arrays.asList(scheduler, nodeShutdown, journal, topologyService); } @VisibleForTesting @@ -906,7 +946,7 @@ public class AccordService implements IAccordService, Shutdownable @Override public Future epochReady(Epoch epoch, Function> get) { - return toFuture(configService.epochReady(epoch.getEpoch(), get)); + return toFuture(topology().epochReady(epoch.getEpoch(), get)); } @Override @@ -921,18 +961,18 @@ public class AccordService implements IAccordService, Shutdownable @Override public void receive(Message message) { - receive(MessagingService.instance(), configService, message); + receive(MessagingService.instance(), node.topology(), message); } @VisibleForTesting - public static void receive(MessageDelivery sink, AbstractConfigurationService configService, Message message) + public static void receive(MessageDelivery sink, TopologyManager topologyManager, Message message) { AccordSyncPropagator.Notification notification = message.payload; - notification.syncComplete.forEach(id -> configService.receiveRemoteSyncComplete(id, notification.epoch)); + notification.readyToCoordinate.forEach(id -> topologyManager.onReadyToCoordinate(id, notification.epoch)); if (!notification.closed.isEmpty()) - configService.receiveClosed(notification.closed, notification.epoch); + topologyManager.onEpochClosed(notification.closed, notification.epoch); if (!notification.retired.isEmpty()) - configService.receiveRetired(notification.retired, notification.epoch); + topologyManager.onEpochRetired(notification.retired, notification.epoch); sink.respond(Ok, message); } @@ -974,9 +1014,15 @@ public class AccordService implements IAccordService, Shutdownable } @VisibleForTesting - public AccordConfigurationService configService() + public AccordEndpointMapper endpointMapper() { - return configService; + return endpointMapper; + } + + @Override + public AccordTopologyService topologyService() + { + return topologyService; } @Override diff --git a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java index 8a1c296b61..ef67c5b496 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java @@ -19,22 +19,24 @@ package org.apache.cassandra.service.accord; import java.io.IOException; -import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.api.TopologyListener; import accord.local.Node; import accord.messages.SimpleReply; import accord.primitives.Ranges; +import accord.topology.Topology; import accord.utils.Invariants; -import accord.utils.SortedList; +import accord.utils.SortedArrays.SortedArrayList; import accord.utils.SortedListSet; import accord.utils.UnhandledEnum; import org.agrona.collections.Int2ObjectHashMap; @@ -63,7 +65,7 @@ import org.apache.cassandra.utils.NoSpamLogger; * Notifies remote replicas that the local replica has synchronised coordination * information for this epoch. */ -public class AccordSyncPropagator +public class AccordSyncPropagator implements TopologyListener { private static final Logger logger = LoggerFactory.getLogger(AccordSyncPropagator.class); private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES); @@ -74,10 +76,9 @@ public class AccordSyncPropagator AccordService.instance().receive(message); }; - interface Listener + interface TestListener { void onEndpointAck(Node.Id id, long epoch); - void onEndpointNack(Node.Id id, long epoch); } private interface ReportPending @@ -136,10 +137,10 @@ public class AccordSyncPropagator boolean ack(Notification notification) { - if (!notification.syncComplete.isEmpty()) + if (!notification.readyToCoordinate.isEmpty()) { - if (notification.syncComplete.containsAll(syncComplete)) syncComplete = ImmutableSet.of(); - else syncComplete = ImmutableSet.copyOf(Iterables.filter(syncComplete, v -> !notification.syncComplete.contains(v))); + if (notification.readyToCoordinate.containsAll(syncComplete)) syncComplete = ImmutableSet.of(); + else syncComplete = ImmutableSet.copyOf(Iterables.filter(syncComplete, v -> !notification.readyToCoordinate.contains(v))); } closed = closed.without(notification.closed); retired = retired.without(notification.retired); @@ -186,21 +187,24 @@ public class AccordSyncPropagator } private final PendingNodes pending = new PendingNodes(); - private final Node.Id localId; + private final Node.Id self; private final AccordEndpointMapper endpointMapper; private final MessageDelivery messagingService; private final ScheduledExecutorPlus scheduler; - private final Listener listener; + private TestListener listener; private final ConcurrentHashMap retryingNotifications = new ConcurrentHashMap<>(); - public AccordSyncPropagator(Node.Id localId, AccordEndpointMapper endpointMapper, - MessageDelivery messagingService, ScheduledExecutorPlus scheduler, - Listener listener) + public AccordSyncPropagator(Node.Id self, AccordEndpointMapper endpointMapper, + MessageDelivery messagingService, ScheduledExecutorPlus scheduler) { - this.localId = localId; + this.self = self; this.endpointMapper = endpointMapper; this.messagingService = messagingService; this.scheduler = scheduler; + } + + void setTestListener(TestListener listener) + { this.listener = listener; } @@ -222,52 +226,59 @@ public class AccordSyncPropagator public String toString() { return "AccordSyncPropagator{" + - "localId=" + localId + + "localId=" + self + ", pending=" + pending + '}'; } - public void onNodesRemoved(Node.Id removed) + public void onNodesRemoved(SortedArrayList removed) { - long[] toAck; - synchronized (AccordSyncPropagator.this) { - PendingEpochs pendingEpochs = pending.remove(removed.id); - if (pendingEpochs == null) return; - toAck = new long[pendingEpochs.size()]; - Long2ObjectHashMap.KeyIterator it = pendingEpochs.keySet().iterator(); - for (int i = 0; it.hasNext(); i++) - { - long epoch = it.nextLong(); - toAck[i] = epoch; - } - Arrays.sort(toAck); - } - - for (int i = 0; i < toAck.length; i++) - { - long epoch = toAck[i]; - listener.onEndpointAck(removed, epoch); + for (Node.Id id : removed) + pending.remove(id.id); } } - public void reportCoordinationReady(long epoch, SortedList notify, Node.Id syncCompleteId) + @Override + public void onReadyToCoordinate(Topology topology) { - SortedListSet remaining = SortedListSet.allOf(notify); - if (remaining.remove(localId)) - listener.onEndpointAck(localId, epoch); - report(epoch, remaining, PendingEpoch::syncComplete, syncCompleteId); + onReadyToCoordinate(topology.epoch(), topology.nodes()); } - public void reportClosed(long epoch, Collection notify, Ranges closed) + @VisibleForTesting + void onReadyToCoordinate(long epoch, SortedArrayList nodes) { - report(epoch, notify, PendingEpoch::closed, closed); + SortedListSet remaining = SortedListSet.allOf(nodes); + if (remaining.remove(self) && listener != null) + listener.onEndpointAck(self, epoch); + report(epoch, remaining, PendingEpoch::syncComplete, self); } - public void reportRetired(long epoch, Collection notify, Ranges retired) + @Override + public void onEpochClosed(Ranges ranges, long epoch, Topology topology) { - report(epoch, notify, PendingEpoch::retired, retired); + if (topology != null) + onEpochClosed(ranges, epoch, topology.nodes()); + } + + @VisibleForTesting + void onEpochClosed(Ranges ranges, long epoch, Collection nodes) + { + report(epoch, nodes, PendingEpoch::closed, ranges); + } + + @Override + public void onEpochRetired(Ranges ranges, long epoch, Topology topology) + { + if (topology != null) + onEpochRetired(ranges, epoch, topology.nodes()); + } + + @VisibleForTesting + void onEpochRetired(Ranges ranges, long epoch, Collection nodes) + { + report(epoch, nodes, PendingEpoch::retired, ranges); } private void report(long epoch, Collection notify, ReportPending report, T param) @@ -291,7 +302,7 @@ public class AccordSyncPropagator private void scheduleRetry(Node.Id to, Notification notification) { - Notification retry = new Notification(notification.epoch, notification.syncComplete, notification.closed, notification.retired, notification.attempts + 1); + Notification retry = new Notification(notification.epoch, notification.readyToCoordinate, notification.closed, notification.retired, notification.attempts + 1); RetryKey key = new RetryKey(to, notification.epoch); retryingNotifications.compute(key, (k, cur) -> { if (cur == null) @@ -332,7 +343,6 @@ public class AccordSyncPropagator case UNKNOWN: // endpoint is not a member of the latest epoch pending.ack(to, notification); - listener.onEndpointNack(to, notification.epoch); return true; case HEALTHY: @@ -349,7 +359,8 @@ public class AccordSyncPropagator } long epoch = notification.epoch; - listener.onEndpointAck(to, epoch); + if (listener != null && notification.readyToCoordinate.contains(self)) + listener.onEndpointAck(to, epoch); } @Override @@ -377,7 +388,7 @@ public class AccordSyncPropagator public void serialize(Notification notification, DataOutputPlus out) throws IOException { out.writeLong(notification.epoch); - CollectionSerializers.serializeCollection(notification.syncComplete, out, TopologySerializers.nodeId); + CollectionSerializers.serializeCollection(notification.readyToCoordinate, out, TopologySerializers.nodeId); KeySerializers.ranges.serialize(notification.closed, out); KeySerializers.ranges.serialize(notification.retired, out); } @@ -395,26 +406,26 @@ public class AccordSyncPropagator public long serializedSize(Notification notification) { return TypeSizes.LONG_SIZE - + CollectionSerializers.serializedCollectionSize(notification.syncComplete, TopologySerializers.nodeId) + + CollectionSerializers.serializedCollectionSize(notification.readyToCoordinate, TopologySerializers.nodeId) + KeySerializers.ranges.serializedSize(notification.closed) + KeySerializers.ranges.serializedSize(notification.retired); } }; final long epoch; - final Collection syncComplete; + final Collection readyToCoordinate; final Ranges closed, retired; final int attempts; - public Notification(long epoch, Collection syncComplete, Ranges closed, Ranges retired) + public Notification(long epoch, Collection readyToCoordinate, Ranges closed, Ranges retired) { - this(epoch, syncComplete, closed, retired, 0); + this(epoch, readyToCoordinate, closed, retired, 0); } - public Notification(long epoch, Collection syncComplete, Ranges closed, Ranges retired, int attempts) + public Notification(long epoch, Collection readyToCoordinate, Ranges closed, Ranges retired, int attempts) { this.epoch = epoch; - this.syncComplete = syncComplete; + this.readyToCoordinate = readyToCoordinate; this.closed = closed; this.retired = retired; this.attempts = attempts; @@ -424,8 +435,8 @@ public class AccordSyncPropagator { Invariants.require(add.epoch == this.epoch); Collection syncComplete = ImmutableSet.builder() - .addAll(this.syncComplete) - .addAll(add.syncComplete) + .addAll(this.readyToCoordinate) + .addAll(add.readyToCoordinate) .build(); return new Notification(epoch, syncComplete, closed.with(add.closed), retired.with(add.retired), Math.max(add.attempts, this.attempts)); } @@ -435,7 +446,7 @@ public class AccordSyncPropagator { return "Notification{" + "epoch=" + epoch + - ", syncComplete=" + syncComplete + + ", syncComplete=" + readyToCoordinate + ", closed=" + closed + ", retired=" + retired + '}'; @@ -468,5 +479,11 @@ public class AccordSyncPropagator RetryKey that = (RetryKey) obj; return that.epoch == this.epoch && that.to.equals(this.to); } + + @Override + public String toString() + { + return epoch + "@" + to; + } } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopology.java b/src/java/org/apache/cassandra/service/accord/AccordTopology.java index db18ba4c9d..15fa854f92 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTopology.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTopology.java @@ -34,10 +34,10 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.Iterables; import com.google.common.collect.Sets; -import accord.api.ConfigurationService.EpochReady; import accord.local.Node; import accord.local.Node.Id; import accord.primitives.Ranges; +import accord.topology.EpochReady; import accord.topology.Shard; import accord.topology.Topology; import accord.utils.Invariants; diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopologyService.java b/src/java/org/apache/cassandra/service/accord/AccordTopologyService.java new file mode 100644 index 0000000000..1a4852bf83 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/AccordTopologyService.java @@ -0,0 +1,204 @@ +/* + * 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.accord; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; +import javax.annotation.concurrent.GuardedBy; + +import com.google.common.annotations.VisibleForTesting; + +import accord.api.TopologyListener; +import accord.api.TopologyService; +import accord.local.Node; +import accord.topology.Topology; +import accord.topology.TopologyRetiredException; +import accord.utils.Invariants; +import accord.utils.SortedArrays.SortedArrayList; +import accord.utils.async.AsyncResult; +import accord.utils.async.AsyncResults.SettableByCallback; +import org.apache.cassandra.concurrent.ScheduledExecutorPlus; +import org.apache.cassandra.concurrent.ScheduledExecutors; +import org.apache.cassandra.concurrent.Shutdownable; +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessageDelivery; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.repair.SharedContext; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.Epoch; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Simulate; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.cassandra.utils.Simulate.With.MONITORS; + +@Simulate(with=MONITORS) +public class AccordTopologyService implements TopologyService, Shutdownable, TopologyListener +{ + public static final Logger logger = LoggerFactory.getLogger(AccordTopologyService.class); + + // TODO (expected): move syncPropagator and watermarkCollector out of this class (and merge them) + private final AccordSyncPropagator syncPropagator; + private final WatermarkCollector watermarkCollector; + + private SortedArrayList previouslyRemovedIds = SortedArrayList.ofSorted(); + + private enum State { INITIALIZED, STARTED, SHUTDOWN } + + @GuardedBy("this") + private State state = State.INITIALIZED; + + public AccordTopologyService(Node.Id node, AccordEndpointMapper endpointMapper, MessageDelivery messagingService, ScheduledExecutorPlus scheduledTasks) + { + this.syncPropagator = new AccordSyncPropagator(node, endpointMapper, messagingService, scheduledTasks); + this.watermarkCollector = new WatermarkCollector(); + } + + public AccordTopologyService(Node.Id node, AccordEndpointMapper endpointMapper) + { + this(node, endpointMapper, MessagingService.instance(), ScheduledExecutors.scheduledTasks); + } + + /** + * On restart, loads topologies. On bootstrap, discovers existing topologies and initializes the node. + */ + public void onStartup(Node node) + { + SortedArrayList removed = node.topology().current().removedIds(); + synchronized (this) + { + Invariants.require(state == State.INITIALIZED, "Expected state to be INITIALIZED but was %s", state); + state = State.STARTED; + previouslyRemovedIds = removed; + } + node.topology().addListener(watermarkCollector); + node.topology().addListener(syncPropagator); + syncPropagator.onNodesRemoved(removed); + } + + @Override + public synchronized boolean isTerminated() + { + return state == State.SHUTDOWN; + } + + @Override + public synchronized void shutdown() + { + if (isTerminated()) + return; + state = State.SHUTDOWN; + } + + @Override + public Object shutdownNow() + { + shutdown(); + return null; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException + { + return isTerminated(); + } + + @Override + public void onReceived(Topology topology) + { + SortedArrayList newlyRemoved; + synchronized (this) + { + newlyRemoved = topology.removedIds().without(previouslyRemovedIds); + previouslyRemovedIds = topology.removedIds().with(previouslyRemovedIds); + } + syncPropagator.onNodesRemoved(newlyRemoved); + } + + @Override + public AsyncResult fetchTopologyForEpoch(long epoch) + { + SettableByCallback result = new SettableByCallback<>(); + fetchTopologyAsync(epoch, result); + return result; + } + + private void fetchTopologyAsync(long epoch, BiConsumer onResult) + { + // It's not safe for this to block on CMS so for now pick a thread pool to handle it + Stage.ACCORD_MIGRATION.execute(() -> { + ClusterMetadata metadata = ClusterMetadata.current(); + try + { + if (metadata.epoch.getEpoch() < epoch) + ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(epoch)); + } + catch (Throwable t) + { + onResult.accept(null, t); + return; + } + + // In most cases, after fetching log from CMS, we will be caught up to the required epoch. + // This TCM will also notify Accord via reportMetadata, so we do not need to fetch topologies. + // If metadata has reported has skipped one or more epochs, and is _ahead_ of the requested epoch, + // we need to fetch topologies from peers to fill in the gap. + if (metadata.epoch.getEpoch() == epoch) + { + Topology topology = AccordTopology.createAccordTopology(metadata); + onResult.accept(topology, null); + return; + } + + Set peers = new HashSet<>(metadata.directory.allJoinedEndpoints()); + peers.remove(FBUtilities.getBroadcastAddressAndPort()); + if (peers.isEmpty()) + { + onResult.accept(null, new TopologyRetiredException("No joined nodes to query; latest epoch was not the one requested", null)); + return; + } + + // Fetching only one epoch here since later epochs might have already been requested concurrently + FetchTopologies.fetch(SharedContext.Global.instance, peers, epoch, epoch) + .addCallback((success, fail) -> { + if (fail != null) onResult.accept(null, fail); + else if (success.hasEpoch(epoch)) onResult.accept(success.get(epoch), null); + else if (success.min > epoch) onResult.accept(null, new TopologyRetiredException("Could not fetch epoch " + epoch + " from peers; too far ahead", null)); + else onResult.accept(null, new RuntimeException("Could not yet retrieve epoch " + epoch)); + }); + }); + } + + @VisibleForTesting + public AccordSyncPropagator syncPropagator() + { + return syncPropagator; + } + + public WatermarkCollector watermarkCollector() + { + return watermarkCollector; + } +} diff --git a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java index d90b449b47..a850a537fd 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java +++ b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java @@ -68,7 +68,7 @@ public class AccordVerbHandler implements IVerbHandler } long waitForEpoch = request.waitForEpoch(); - if (node.topology().hasAtLeastEpoch(waitForEpoch)) + if (node.topology().active().hasAtLeastEpoch(waitForEpoch)) { request.process(node, fromNodeId, message.header); } diff --git a/src/java/org/apache/cassandra/service/accord/FetchTopologies.java b/src/java/org/apache/cassandra/service/accord/FetchTopologies.java index 3aca171e38..80a1098f39 100644 --- a/src/java/org/apache/cassandra/service/accord/FetchTopologies.java +++ b/src/java/org/apache/cassandra/service/accord/FetchTopologies.java @@ -28,6 +28,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.topology.Topology; +import accord.topology.TopologyRange; + import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.exceptions.RequestFailure; import org.apache.cassandra.io.UnversionedSerializer; @@ -43,7 +45,6 @@ import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.service.accord.serializers.TopologySerializers; import org.apache.cassandra.utils.concurrent.Future; -import static accord.topology.TopologyManager.TopologyRange; import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryFetchTopology; /** @@ -153,7 +154,7 @@ public class FetchTopologies return; } - TopologyRange topologies = AccordService.instance().topology().between(message.payload.minEpoch, message.payload.maxEpoch); + TopologyRange topologies = AccordService.instance().topology().active().between(message.payload.minEpoch, message.payload.maxEpoch); logger.debug("Responding with {} failure to {}", topologies, message.payload); MessagingService.instance().respond(topologies, message); }; diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index 37bf99739b..8c74725ab6 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -28,7 +28,6 @@ import java.util.function.Function; import javax.annotation.Nonnull; import javax.annotation.Nullable; -import accord.api.ConfigurationService.EpochReady; import accord.utils.async.AsyncResult; import org.apache.cassandra.tcm.ClusterMetadata; import org.slf4j.Logger; @@ -48,6 +47,7 @@ import accord.primitives.Keys; import accord.primitives.Ranges; import accord.primitives.Timestamp; import accord.primitives.Txn; +import accord.topology.EpochReady; import accord.topology.TopologyManager; import accord.utils.Invariants; import accord.utils.async.AsyncChain; @@ -180,7 +180,9 @@ public interface IAccordService void awaitDone(TableId id, long epoch); - AccordConfigurationService configService(); + AccordEndpointMapper endpointMapper(); + + AccordTopologyService topologyService(); Params journalConfiguration(); @@ -350,9 +352,15 @@ public interface IAccordService } @Override - public AccordConfigurationService configService() + public AccordEndpointMapper endpointMapper() { - return null; + throw new UnsupportedOperationException(); + } + + @Override + public AccordTopologyService topologyService() + { + throw new UnsupportedOperationException(); } @Override @@ -420,9 +428,15 @@ public interface IAccordService } @Override - public AccordConfigurationService configService() + public AccordEndpointMapper endpointMapper() { - return delegate.configService(); + return delegate.endpointMapper(); + } + + @Override + public AccordTopologyService topologyService() + { + return delegate.topologyService(); } @Nonnull diff --git a/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java b/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java index 6993504a5c..a4ed02eaa1 100644 --- a/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java +++ b/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java @@ -28,19 +28,21 @@ import java.util.Objects; import java.util.Set; import java.util.function.BiConsumer; +import javax.annotation.Nullable; + import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Predicates; import com.google.common.collect.Iterators; import com.google.common.primitives.Ints; -import accord.api.ConfigurationService; +import accord.api.TopologyListener; import accord.local.Node; import accord.primitives.Range; import accord.primitives.Ranges; import accord.topology.Topology; +import accord.topology.TopologyManager; import accord.utils.Invariants; import accord.utils.ReducingRangeMap; -import accord.utils.async.AsyncResult; import org.agrona.collections.Long2LongHashMap; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.UnversionedSerializer; @@ -62,7 +64,7 @@ import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retry /** * Collects watermarks of closed and retired epochs per range, and synced epochs per node. */ -public class WatermarkCollector implements ConfigurationService.Listener +public class WatermarkCollector implements TopologyListener { private static final Comparator> sortByEpochThenRange = (a, b) -> { int c = Long.compareUnsigned(a.getValue(), b.getValue()); @@ -81,30 +83,25 @@ public class WatermarkCollector implements ConfigurationService.Listener synced = new Long2LongHashMap(-1); } - @Override public AsyncResult onTopologyUpdate(Topology topology) - { - return null; - } - @Override - public synchronized void onRemoteSyncComplete(Node.Id node, long epoch) + public synchronized void onRemoteReadyToCoordinate(Node.Id node, long epoch) { synced.compute(node.id, (k, prev) -> prev == -1 ? epoch : Long.max(prev, epoch)); } @Override - public synchronized void onEpochClosed(Ranges ranges, long epoch) + public synchronized void onEpochClosed(Ranges ranges, long epoch, @Nullable Topology topology) { closed = ReducingRangeMap.merge(closed, ReducingRangeMap.create(ranges, epoch), Long::max); } @Override - public synchronized void onEpochRetired(Ranges ranges, long epoch) + public synchronized void onEpochRetired(Ranges ranges, long epoch, @Nullable Topology topology) { retired = ReducingRangeMap.merge(retired, ReducingRangeMap.create(ranges, epoch), Long::max); } - public final IVerbHandler handler = new IVerbHandler() + public final IVerbHandler handler = new IVerbHandler<>() { public void doVerb(Message message) { @@ -123,7 +120,7 @@ public class WatermarkCollector implements ConfigurationService.Listener }; @VisibleForTesting - static void fetchAndReportWatermarksAsync(AccordConfigurationService configService) + static void fetchAndReportWatermarksAsync(TopologyManager topologyManager) { SharedContext context = SharedContext.Global.instance; Set peers = new HashSet<>(); @@ -142,14 +139,13 @@ public class WatermarkCollector implements ConfigurationService.Listener return; Snapshot snapshot = m.payload; - long minEpoch = configService.minEpoch(); - forEachEpoch(configService::receiveClosed, snapshot.closed); - forEachEpoch(configService::receiveRetired, snapshot.retired); + long minEpoch = topologyManager.minEpoch(); + forEachEpoch(topologyManager::onEpochClosed, snapshot.closed); + forEachEpoch(topologyManager::onEpochRetired, snapshot.retired); for (Map.Entry e : snapshot.synced.entrySet()) { Node.Id node = new Node.Id(Ints.saturatedCast(e.getKey())); - for (long epoch = minEpoch; epoch <= e.getValue(); epoch++) - configService.receiveRemoteSyncComplete(node, epoch); + topologyManager.onReadyToCoordinate(node, e.getValue()); } }); } diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index eb69136d6e..e310353729 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -90,6 +90,7 @@ import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.fetch import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.recover; import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryBootstrap; import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryDurability; +import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryFetchTopology; import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryJoinBootstrap; import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retrySyncPoint; import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.slowTxnPreaccept; @@ -303,7 +304,7 @@ public class AccordAgent implements Agent, OwnershipEventListener } RoutingKey homeKey = command.route().homeKey(); - Shard shard = node.topology().forEpochIfKnown(homeKey, command.txnId().epoch()); + Shard shard = node.topology().active().forEpochIfKnown(homeKey, command.txnId().epoch()); startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), ONE_SECOND, random); long delayMicros = Math.max(1, startTime - nowMicros); @@ -383,6 +384,12 @@ public class AccordAgent implements Agent, OwnershipEventListener return retrySyncPoint.computeWait(attempt, units); } + @Override + public long retryTopologyDelay(Node node, int attempt, TimeUnit units) + { + return retryFetchTopology.computeWait(attempt, units); + } + @Override public long retryDurabilityDelay(Node node, int attempt, TimeUnit units) { diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java index eddd77abb6..eaae5c350c 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -46,6 +46,7 @@ import accord.primitives.Timestamp; import accord.primitives.TimestampWithUniqueHlc; import accord.primitives.Txn; import accord.primitives.TxnId; +import accord.topology.ActiveEpochs; import accord.topology.Shard; import accord.topology.Topologies; import accord.topology.Topology; @@ -156,9 +157,10 @@ public class AccordInteropExecution implements ReadCoordinator this.endpointMapper = endpointMapper; // TODO (required): compare this to latest logic in Accord, make sure it makes sense - this.executes = node.topology().forEpoch(route, executeAt.epoch(), SHARE); + ActiveEpochs epochs = node.topology().active(); + this.executes = epochs.forEpoch(route, executeAt.epoch(), SHARE); this.allTopologies = txnId.epoch() != executeAt.epoch() - ? node.topology().preciseEpochs(route, txnId.epoch(), executeAt.epoch(), SHARE) + ? epochs.preciseEpochs(route, txnId.epoch(), executeAt.epoch(), SHARE) : executes; this.executeTopology = executes.getEpoch(executeAt.epoch()); this.coordinateTopology = allTopologies.getEpoch(txnId.epoch()); diff --git a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java index ffdca7a2ad..26b7e546f8 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java @@ -32,7 +32,6 @@ import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.UnversionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.service.accord.AccordConfigurationService; import org.apache.cassandra.service.accord.AccordJournalValueSerializers; import org.apache.cassandra.service.accord.JournalKey; import org.apache.cassandra.service.accord.serializers.KeySerializers; @@ -77,7 +76,6 @@ public interface AccordTopologyUpdate epochs[i] = in.readLong(); ranges[i] = KeySerializers.ranges.deserialize(in); } - Invariants.require(ranges.length == epochs.length); return new CommandStores.RangesForEpoch(epochs, ranges); } @@ -162,10 +160,7 @@ public interface AccordTopologyUpdate out.writeBoolean(image.update != null); if (image.update != null) TopologyUpdateSerializer.instance.serialize(image.update, out); - if (image.syncStatus == null) - out.writeByte(Byte.MAX_VALUE); - else - out.writeByte(image.syncStatus.ordinal()); + out.writeByte(0); // defunct enum byte KeySerializers.ranges.serialize(image.closed, out); KeySerializers.ranges.serialize(image.retired, out); @@ -192,9 +187,7 @@ public interface AccordTopologyUpdate update = TopologyUpdateSerializer.instance.deserialize(in); TopologyImage image = new TopologyImage(epoch, kind, update); - byte syncStateByte = in.readByte(); - if (syncStateByte != Byte.MAX_VALUE) - image.syncStatus = AccordConfigurationService.SyncStatus.values()[syncStateByte]; + in.readByte(); // defunct enum byte image.closed = KeySerializers.ranges.deserialize(in); image.retired = KeySerializers.ranges.deserialize(in); @@ -252,7 +245,6 @@ public interface AccordTopologyUpdate private final long epoch; private final Kind kind; private Journal.TopologyUpdate update; - private AccordConfigurationService.SyncStatus syncStatus = null; private Ranges closed = Ranges.EMPTY; private Ranges retired = Ranges.EMPTY; @@ -317,9 +309,6 @@ public interface AccordTopologyUpdate Invariants.require(accumulator.update == null || accumulator.update.equals(update)); accumulator.update = update; - // We're iterating in _reverse_ order - if (accumulator.syncStatus == null) - accumulator.syncStatus = syncStatus; accumulator.closed = accumulator.closed.with(closed); accumulator.retired = accumulator.retired.with(retired); } @@ -330,13 +319,13 @@ public interface AccordTopologyUpdate if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; TopologyImage that = (TopologyImage) o; - return epoch == that.epoch && Objects.equals(update, that.update) && syncStatus == that.syncStatus && closed.equals(that.closed) && retired.equals(that.retired); + return epoch == that.epoch && Objects.equals(update, that.update) && closed.equals(that.closed) && retired.equals(that.retired); } @Override public int hashCode() { - return Objects.hash(update, syncStatus, closed, retired, epoch); + return Objects.hash(update, closed, retired, epoch); } } diff --git a/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java b/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java index a6d37598bf..c2e9bce8aa 100644 --- a/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java +++ b/src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java @@ -90,7 +90,7 @@ public class AccordRepair if (endpoints != null) { including = new ArrayList<>(endpoints.size()); - AccordEndpointMapper mapper = AccordService.instance().configService().endpointMapper(); + AccordEndpointMapper mapper = AccordService.instance().endpointMapper(); for (InetAddressAndPort ep : endpoints) { Node.Id id = mapper.mappedIdOrNull(ep); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java index f3b4d1107e..c20d4a5e91 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java @@ -42,7 +42,6 @@ import static accord.primitives.KeyDeps.SerializerSupport.txnIdsToKeys; import static accord.primitives.RangeDeps.SerializerSupport.ranges; import static accord.primitives.RangeDeps.SerializerSupport.rangesToTxnIds; import static accord.primitives.RangeDeps.SerializerSupport.txnIdsToRanges; -import static org.apache.cassandra.service.accord.serializers.KeySerializers.keys; import static org.apache.cassandra.service.accord.serializers.SerializePacked.deserializePackedInts; import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializePackedInts; import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializedPackedIntsSize; diff --git a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java index 70664213a2..e41d1a2f43 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java +++ b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java @@ -30,7 +30,7 @@ import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.EpochReady; import accord.local.Node; import accord.utils.async.AsyncResult; import com.googlecode.concurrenttrees.common.Iterables; @@ -371,7 +371,7 @@ public class BootstrapAndJoin extends MultiStepOperation { IAccordService service = AccordService.instance(); { - Future ready = AccordService.toFuture(service.epochReadyFor(metadata, EpochReady::metadata)); + Future ready = AccordService.toFuture(service.epochReadyFor(metadata, EpochReady::active)); logger.info("Waiting for Accord metadata to be ready"); ready.syncThrowUncheckedOnInterrupt(); ready.rethrowIfFailed(); diff --git a/src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java b/src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java index 8a121fd2c2..8cd981335a 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java +++ b/src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java @@ -26,7 +26,7 @@ import java.util.concurrent.ExecutionException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.EpochReady; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.schema.Keyspaces; diff --git a/src/java/org/apache/cassandra/tcm/sequences/Move.java b/src/java/org/apache/cassandra/tcm/sequences/Move.java index 25058e20f2..e67763f6b1 100644 --- a/src/java/org/apache/cassandra/tcm/sequences/Move.java +++ b/src/java/org/apache/cassandra/tcm/sequences/Move.java @@ -31,7 +31,7 @@ import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.EpochReady; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.TypeSizes; diff --git a/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java b/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java index 6a518975b4..02fd3d9483 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java +++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java @@ -24,6 +24,7 @@ import java.util.Set; import com.google.common.collect.ImmutableSet; +import accord.utils.Invariants; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Token; @@ -105,8 +106,8 @@ public class PrepareJoin implements Transformation boolean joinTokenRing, boolean streamData) { - this.nodeId = nodeId; - this.tokens = tokens; + this.nodeId = Invariants.nonNull(nodeId); + this.tokens = Invariants.nonNull(tokens); this.placementProvider = placementProvider; this.joinTokenRing = joinTokenRing; this.streamData = streamData; diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java index 3cdc16ce23..a362938aba 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -49,7 +49,7 @@ import com.google.common.base.Strings; import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.Futures; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.EpochReady; import org.agrona.collections.IntArrayList; import org.apache.cassandra.tcm.compatibility.TokenRingUtils; import org.apache.cassandra.utils.FBUtilities; diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java index 19ba9dd279..131bdfca3f 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java @@ -22,7 +22,9 @@ import java.net.InetAddress; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.function.Consumer; import org.junit.Assert; @@ -30,6 +32,7 @@ import org.junit.Test; import accord.primitives.RoutingKeys; import accord.primitives.Timestamp; +import accord.topology.EpochReady; import accord.topology.TopologyManager; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.UntypedResultSet; @@ -70,7 +73,8 @@ public class AccordBootstrapTest extends AccordBootstrapTestBase IInvokableInstance newInstance = cluster.bootstrap(config); newInstance.startup(cluster); spinUntilTrue(() -> cluster.stream().anyMatch(instance -> instance.callOnInstance(() -> StreamListener.listener.hasFailedStream))); - newInstance.shutdown(false); + try { newInstance.shutdown(false).get(5L, TimeUnit.MINUTES); } + catch (InterruptedException | ExecutionException | TimeoutException e) { throw new RuntimeException(e); } cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> StreamListener.listener.failStream = false)); newInstance.startup(cluster); // todo: re-add once we fix write survey/join ring = false mode @@ -174,6 +178,7 @@ public class AccordBootstrapTest extends AccordBootstrapTestBase .with(NETWORK, GOSSIP)) .start()) { + cluster.setUncaughtExceptionsFilter(throwable -> throwable.getClass().getSimpleName().equals("UncheckedInterruptedException")); cluster.schemaChange("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor':2}"); cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key(k, c)) WITH transactional_mode='full'"); @@ -228,7 +233,7 @@ public class AccordBootstrapTest extends AccordBootstrapTestBase for (long epoch = topologyManager.minEpoch() ; epoch <= topologyManager.epoch() ; ++epoch) { CountDownLatch latch = new CountDownLatch(1); - topologyManager.epochReady(epoch).data.invokeIfSuccess(latch::countDown); + topologyManager.epochReady(epoch, EpochReady::data).invokeIfSuccess(latch::countDown); while (true) { try diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTestBase.java index 57836285e4..8e1d28d070 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTestBase.java @@ -23,7 +23,8 @@ import java.util.function.Function; import org.junit.Assert; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.EpochReady; +import accord.topology.TopologyManager; import accord.utils.async.AsyncResult; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.DecoratedKey; @@ -35,7 +36,6 @@ import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableFuncti import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; -import org.apache.cassandra.service.accord.AccordConfigurationService; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.accord.api.PartitionKey; import org.apache.cassandra.tcm.ClusterMetadata; @@ -45,8 +45,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.assertj.core.api.Assertions; -import static org.apache.cassandra.service.accord.AccordConfigurationService.EpochSnapshot.ResultStatus.SUCCESS; -import static org.apache.cassandra.service.accord.AccordConfigurationService.SyncStatus.COMPLETED; +import static org.apache.cassandra.service.accord.AccordService.toFuture; public class AccordBootstrapTestBase extends TestBaseImpl { @@ -89,9 +88,8 @@ public class AccordBootstrapTestBase extends TestBaseImpl { boolean completed = service().epochReady(Epoch.create(epoch), await).await(60, TimeUnit.SECONDS); Assertions.assertThat(completed) - .describedAs("Epoch %s did not become ready within timeout on %s -> %s", - epoch, FBUtilities.getBroadcastAddressAndPort(), - service().configService().getEpochSnapshot(epoch)) + .describedAs("Epoch %s did not become ready within timeout on %s", + epoch, FBUtilities.getBroadcastAddressAndPort()) .isTrue(); } catch (InterruptedException e) @@ -104,10 +102,10 @@ public class AccordBootstrapTestBase extends TestBaseImpl { try { - AccordConfigurationService configService = service().configService(); - boolean completed = configService.unsafeLocalSyncNotified(epoch).await(60, TimeUnit.SECONDS); - Assert.assertTrue(String.format("Local sync notification for epoch %s did not become ready within timeout on %s\n%s", - epoch, FBUtilities.getBroadcastAddressAndPort(), service().configService().getDebugStr()), completed); + TopologyManager tm = service().topology(); + boolean completed = toFuture(tm.epochReady(epoch, EpochReady::coordinate)).await(60, TimeUnit.SECONDS); + Assert.assertTrue(String.format("Local sync notification for epoch %s did not become ready within timeout on %s", + epoch, FBUtilities.getBroadcastAddressAndPort()), completed); } catch (InterruptedException e) { @@ -131,7 +129,7 @@ public class AccordBootstrapTestBase extends TestBaseImpl static long awaitMaxEpochMetadataReady(Cluster cluster) { - return awaitMaxEpoch(cluster, EpochReady::metadata, false); + return awaitMaxEpoch(cluster, EpochReady::active, false); } static long awaitMaxEpoch(Cluster cluster, SerializableFunction> await, boolean expectReadyToRead) @@ -147,16 +145,16 @@ public class AccordBootstrapTestBase extends TestBaseImpl Assert.assertEquals(maxEpoch, ClusterMetadata.current().epoch.getEpoch()); AccordService service = (AccordService) AccordService.instance(); awaitEpoch(maxEpoch, aw); - AccordConfigurationService configService = service.configService(); + TopologyManager tm = service.topology(); awaitLocalSyncNotification(maxEpoch); - for (long epoch = configService.minEpoch(); epoch <= maxEpoch; epoch++) + for (long epoch = tm.minEpoch(); epoch <= maxEpoch; epoch++) { - Assert.assertEquals(COMPLETED, configService.getEpochSnapshot(maxEpoch).syncStatus); - Assert.assertEquals(SUCCESS, configService.getEpochSnapshot(maxEpoch).acknowledged); - Assert.assertEquals(SUCCESS, configService.getEpochSnapshot(maxEpoch).received); + EpochReady epochReady = tm.active().epochReady(epoch); + Assert.assertTrue(epochReady.active().isDone()); + Assert.assertTrue(epochReady.coordinate.isDone()); if (expectReadyToRead) - Assert.assertEquals(SUCCESS, configService.getEpochSnapshot(maxEpoch).reads); + Assert.assertTrue(epochReady.reads.isDone()); } }, node.transfer(await)); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java index a774889711..2279026306 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -709,7 +709,7 @@ public abstract class AccordCQLTestBase extends AccordTestBase Unseekables routables = AccordTestUtils.createTxn(sb.toString()).keys().toParticipants(); long epoch = AccordService.instance().topology().epoch(); - Topologies topology = AccordService.instance().topology().withUnsyncedEpochs(routables, epoch, epoch); + Topologies topology = AccordService.instance().topology().active().withUnsyncedEpochs(routables, epoch, epoch); // we don't detect out-of-bounds read/write yet, so use this to validate we reach different shards Assertions.assertThat(topology.totalShards()).isEqualTo(2); }); diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMoveTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMoveTest.java index 5669e261a7..930d3c67e7 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMoveTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMoveTest.java @@ -63,7 +63,6 @@ public class AccordMoveTest extends AccordBootstrapTestBase cluster.schemaChange("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor':2}"); cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key(k, c)) WITH transactional_mode='full'"); - long initialMax = maxEpoch(cluster); long[] tokens = new long[3]; for (int i=0; i<3; i++) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordRecoverFromAvailabilityLossTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordRecoverFromAvailabilityLossTest.java index c6dfc9c382..fd223de906 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordRecoverFromAvailabilityLossTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordRecoverFromAvailabilityLossTest.java @@ -31,6 +31,7 @@ import org.junit.Test; import accord.primitives.RoutingKeys; import accord.primitives.Timestamp; +import accord.topology.EpochReady; import accord.topology.TopologyManager; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.UntypedResultSet; @@ -106,7 +107,7 @@ public class AccordRecoverFromAvailabilityLossTest extends AccordBootstrapTestBa FBUtilities.waitOnFuture(cluster.get(removeIdx).shutdown(false)); { - List> results = new ArrayList<>(); + List> results = new ArrayList<>(); for (int key = 50; key < 100; key++) { String query = "BEGIN TRANSACTION\n" + @@ -155,7 +156,7 @@ public class AccordRecoverFromAvailabilityLossTest extends AccordBootstrapTestBa for (long epoch = topologyManager.minEpoch() ; epoch <= topologyManager.epoch() ; ++epoch) { CountDownLatch latch = new CountDownLatch(1); - topologyManager.epochReady(epoch).data.invokeIfSuccess(latch::countDown); + topologyManager.epochReady(epoch, EpochReady::data).invokeIfSuccess(latch::countDown); while (true) { try diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java index cf024f43f8..1a8a8a7223 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java @@ -23,6 +23,7 @@ import java.util.Comparator; import java.util.HashSet; import java.util.Set; +import accord.topology.TopologyManager; import org.apache.cassandra.service.accord.AccordFastPath; import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.tcm.Epoch; @@ -40,7 +41,6 @@ import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.gms.FailureDetector; import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.service.accord.AccordConfigurationService; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.tcm.ClusterMetadata; import org.slf4j.Logger; @@ -109,8 +109,8 @@ public class AccordSimpleFastPathTest extends TestBaseImpl Assert.assertEquals(idSet(), accordFastPath.unavailableIds()); long epoch = cm.epoch.getEpoch(); - AccordConfigurationService configService = ((AccordService) AccordService.instance()).configService(); - Topology topology = configService.getTopologyForEpoch(epoch); + TopologyManager tm = AccordService.instance().topology(); + Topology topology = tm.active().globalForEpoch(epoch); Assert.assertFalse(topology.shards().isEmpty()); topology.shards().forEach(shard -> Assert.assertEquals(idSet(1, 2, 3), shard.nodes.without(shard.notInFastPath))); return cm.epoch.getEpoch(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java index cb91e9a160..47db6443d0 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java @@ -177,8 +177,7 @@ public abstract class AccordTestBase extends TestBaseImpl return true; AccordService accord = ((AccordService) AccordService.instance()); - return metadata.epoch.getEpoch() == accord.configService().currentEpoch() && - metadata.epoch.getEpoch() == accord.topology().current().epoch(); + return metadata.epoch.getEpoch() == accord.topology().current().epoch(); }, 60)); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java index 84c2bb2a0f..0e8a01c228 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java @@ -137,7 +137,7 @@ public class AccordJournalReplayTest extends TestBaseImpl return txnId.toString(); }); - cluster.get(1).shutdown(false); + cluster.get(1).shutdown(false).get(); cluster.get(1).startup(); cluster.get(1).runOnInstance(() -> { AccordService service = (AccordService) AccordService.instance(); diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java index a76d2ca6a1..64e47e9802 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java @@ -33,12 +33,11 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import accord.api.ConfigurationService; import accord.local.Node; import accord.primitives.Ranges; +import accord.topology.EpochReady; import accord.topology.TopologyManager; import org.apache.cassandra.exceptions.InvalidRequestException; -import org.apache.cassandra.service.accord.AccordConfigurationService; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.simulator.Action; @@ -195,7 +194,6 @@ public class EpochStressTest extends SimulationTestBase AccordService accord = (AccordService) AccordService.instance(); Node node = accord.node(); - AccordConfigurationService configService = (AccordConfigurationService) node.configService(); TopologyManager tm = node.topology(); long minEpoch = tm.minEpoch(); @@ -212,26 +210,9 @@ public class EpochStressTest extends SimulationTestBase for (long epoch = minEpoch; epoch <= maxEpoch; epoch++) { long finalEpoch = epoch; - ConfigurationService.EpochReady ready = tm.epochReady(epoch); + EpochReady ready = tm.active().epochReady(epoch); while (!isDone(ready)) sleep.accept(() -> "Epoch " + finalEpoch + "'s EpochReady is not done; " + ready); - - AccordConfigurationService.EpochSnapshot snapshot = configService.getEpochSnapshot(epoch); - while (!isDone(snapshot)) - { - AccordConfigurationService.SyncStatus status = snapshot.syncStatus; - sleep.accept(() -> "Epoch " + finalEpoch + "'s SyncStatus is not done; " + status); - snapshot = configService.getEpochSnapshot(epoch); - } - - Ranges expected = tm.globalForEpoch(epoch).ranges().mergeTouching(); - Ranges synced = tm.syncComplete(epoch).mergeTouching(); - while (!isDone(synced, expected)) - { - Ranges finalSynced = synced; - sleep.accept(() -> "Epoch " + finalEpoch + "'s syncComplete is not done; missing " + expected.without(finalSynced)); - synced = tm.syncComplete(epoch).mergeTouching(); - } } logger.info("All epochs completed in {}", Duration.ofNanos(Clock.Global.nanoTime() - startNano)); } @@ -241,14 +222,9 @@ public class EpochStressTest extends SimulationTestBase return synced.equals(expected); } - private static boolean isDone(AccordConfigurationService.EpochSnapshot snapshot) + private static boolean isDone(EpochReady ready) { - return snapshot.syncStatus == AccordConfigurationService.SyncStatus.COMPLETED; - } - - private static boolean isDone(ConfigurationService.EpochReady ready) - { - return ready.metadata.isDone() + return ready.active.isDone() && ready.coordinate.isDone() && ready.data.isDone() && ready.reads.isDone(); diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java index d53906cc3c..e4e6de559f 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java @@ -26,10 +26,15 @@ import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; -import accord.api.ConfigurationService; +import accord.Utils; +import accord.api.MessageSink; import accord.api.TopologySorter; +import accord.impl.mock.MockCluster; import accord.local.Node; import accord.primitives.Ranges; +import accord.topology.ActiveEpoch; +import accord.topology.ActiveEpochs; +import accord.topology.EpochReady; import accord.topology.Shard; import accord.topology.Topologies; import accord.topology.Topology; @@ -54,8 +59,8 @@ public class AccordVirtualTablesTest extends CQLTester public static final Node.Id N1 = new Node.Id(1); public static final SortedArrays.SortedArrayList ALL = SortedArrays.SortedArrayList.ofSorted(N1); public static final Set FP = Collections.singleton(N1); - public static final String SUCCESS = "success"; - public static final String PENDING = "pending"; + public static final String SUCCESS = "Success"; + public static final String PENDING = "SettableResult{status=pending}"; public static final List FULL_RANGE = List.of("(-Inf, +Inf]"); public static TableId T1; @@ -92,12 +97,14 @@ public class AccordVirtualTablesTest extends CQLTester { TopologyManager tm = empty(); long e1 = 1; - tm.onTopologyUpdate(topology(e1, T1), () -> ConfigurationService.EpochReady.done(e1), e -> {}); + ActiveEpoch a1 = active(topology(e1, T1), EpochReady.done(e1), Topology.EMPTY); + tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a1 }, -1)); assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS), row(e1, true, SUCCESS, SUCCESS, SUCCESS, SUCCESS)); long e2 = 2; - tm.onTopologyUpdate(topology(e2, T1), () -> pendingReady(e1), e -> {}); + ActiveEpoch a2 = active(topology(e2, T1), pendingReady(e2), Topology.EMPTY); + tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a2, a1 }, -1)); assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS), row(e2, false, PENDING, PENDING, PENDING, PENDING), row(e1, true, SUCCESS, SUCCESS, SUCCESS, SUCCESS)); @@ -108,7 +115,8 @@ public class AccordVirtualTablesTest extends CQLTester { TopologyManager tm = empty(); long e1 = 1; - tm.onTopologyUpdate(topology(e1, T1), () -> ConfigurationService.EpochReady.done(e1), e -> {}); + ActiveEpoch a1 = active(topology(e1, T1), EpochReady.done(e1), Topology.EMPTY); + tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a1 }, -1)); // the range was added in the first epoch, so its fully synced assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS), @@ -116,12 +124,13 @@ public class AccordVirtualTablesTest extends CQLTester // range is no longer "added" so doesn't show up as synced! long e2 = 2; - tm.onTopologyUpdate(topology(e2, T1), () -> ConfigurationService.EpochReady.done(e2), e -> {}); + ActiveEpoch a2 = active(topology(e2, T1), EpochReady.done(e2), a1.global()); + tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a2, a1 }, -1)); assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS), row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE)); // sync the range - tm.onEpochSyncComplete(N1, e2); + tm.onReadyToCoordinate(N1, e2); assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS), row(e2, T1_META.keyspace, T1_META.name, List.of(), List.of(), List.of(), List.of(), FULL_RANGE), row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE)); @@ -139,9 +148,9 @@ public class AccordVirtualTablesTest extends CQLTester row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, FULL_RANGE, List.of(), FULL_RANGE, FULL_RANGE)); } - private static ConfigurationService.EpochReady pendingReady(long epoch) + private static EpochReady pendingReady(long epoch) { - return new ConfigurationService.EpochReady(epoch, AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable()); + return new EpochReady(epoch, AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable()); } private static Topology topology(long epoch, TableId tableId) @@ -152,26 +161,34 @@ public class AccordVirtualTablesTest extends CQLTester private static TopologyManager empty() { - TopologySorter sorter = (TopologySorter.StaticSorter) (node1, node2, shards) -> 0; TopologySorter.Supplier supplier = new TopologySorter.Supplier() { @Override public TopologySorter get(Topology topologies) { - return sorter; + return SORTER; } @Override public TopologySorter get(Topologies topologies) { - return sorter; + return SORTER; } }; - TopologyManager tm = new TopologyManager(supplier, null, N1, null, null); + + Node node = Utils.createNode(N1, Topology.EMPTY, new MessageSink.NoOpSink(), new MockCluster.Clock(0)); + TopologyManager tm = new TopologyManager(supplier, node, ignore -> { throw new UnsupportedOperationException(); }, null, null); var mock = Mockito.mock(IAccordService.class); Mockito.when(mock.topology()).thenReturn(tm); AccordService.unsafeSetNewAccordService(mock); return tm; } + + private static final TopologySorter SORTER = (TopologySorter.StaticSorter) (node1, node2, shards) -> 0; + + private static ActiveEpoch active(Topology topology, EpochReady ready, Topology prev) + { + return ActiveEpoch.unsafeNew(N1, topology, ready, SORTER, prev.ranges()); + } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java index b467174c42..9e6d406570 100644 --- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java @@ -31,7 +31,7 @@ import java.util.function.Supplier; import java.util.stream.Collectors; import javax.annotation.Nullable; -import accord.api.ConfigurationService.EpochReady; +import accord.topology.EpochReady; import accord.api.Journal; import accord.api.RoutingKey; import accord.local.CommandStores; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java deleted file mode 100644 index 04fa068f06..0000000000 --- a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java +++ /dev/null @@ -1,295 +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.accord; - -import java.net.UnknownHostException; -import java.nio.file.Files; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Optional; -import java.util.UUID; - -import accord.api.Journal.TopologyUpdate; -import org.junit.BeforeClass; -import org.junit.Test; - -import accord.impl.AbstractConfigurationServiceTest; -import accord.local.Node.Id; -import accord.topology.Topology; -import accord.utils.SortedArrays.SortedArrayList; -import accord.utils.async.AsyncResult; -import org.agrona.collections.Int2ObjectHashMap; -import org.apache.cassandra.SchemaLoader; -import org.apache.cassandra.ServerTestUtils; -import org.apache.cassandra.concurrent.ScheduledExecutors; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.db.marshal.Int32Type; -import org.apache.cassandra.dht.Murmur3Partitioner; -import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.io.util.File; -import org.apache.cassandra.journal.TestParams; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.locator.Replica; -import org.apache.cassandra.net.ConnectionType; -import org.apache.cassandra.net.Message; -import org.apache.cassandra.net.MessageDelivery; -import org.apache.cassandra.net.RequestCallback; -import org.apache.cassandra.schema.DistributedSchema; -import org.apache.cassandra.schema.KeyspaceMetadata; -import org.apache.cassandra.schema.KeyspaceParams; -import org.apache.cassandra.schema.Schema; -import org.apache.cassandra.schema.TableId; -import org.apache.cassandra.schema.TableMetadata; -import org.apache.cassandra.schema.Tables; -import org.apache.cassandra.service.accord.EndpointMapping.Updateable; -import org.apache.cassandra.service.accord.api.AccordAgent; -import org.apache.cassandra.tcm.ClusterMetadata; -import org.apache.cassandra.tcm.ValidatingClusterMetadataService; -import org.apache.cassandra.tcm.membership.Location; -import org.apache.cassandra.tcm.membership.NodeAddresses; -import org.apache.cassandra.tcm.membership.NodeId; -import org.apache.cassandra.tcm.membership.NodeVersion; -import org.apache.cassandra.tcm.ownership.DataPlacement; -import org.apache.cassandra.tcm.serialization.Version; -import org.apache.cassandra.utils.concurrent.Future; - -import static accord.impl.AbstractConfigurationServiceTest.TestListener; -import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse; - -public class AccordConfigurationServiceTest -{ - private static final Id ID1 = new Id(1); - private static final Id ID2 = new Id(2); - private static final Id ID3 = new Id(3); - private static final SortedArrayList ID_LIST = new SortedArrayList<>(new Id[] { ID1, ID2, ID3 }); - private static final String KEYSPACE_NAME = "test_ks"; - private static final TableId TBL_ID = TableId.fromUUID(new UUID(0, 1)); - - private static EndpointMapping mappingForEpoch(long epoch) - { - try - { - EndpointMapping.Builder builder = EndpointMapping.builder(epoch); - builder.add(InetAddressAndPort.getByName("127.0.0.1"), ID1); - builder.add(InetAddressAndPort.getByName("127.0.0.2"), ID2); - builder.add(InetAddressAndPort.getByName("127.0.0.3"), ID3); - return builder.build(); - } - catch (UnknownHostException e) - { - throw new RuntimeException(e); - } - } - - private static class Messaging implements MessageDelivery - { - static class Request - { - final Message message; - final InetAddressAndPort to; - final RequestCallback callback; - - public Request(Message message, InetAddressAndPort to, RequestCallback callback) - { - this.message = message; - this.to = to; - this.callback = callback; - } - } - - final List requests = new ArrayList<>(); - - @Override - public void send(Message message, InetAddressAndPort to) - { - requests.add(new Request(message, to, null)); - } - - @Override - public void sendWithCallback(Message message, InetAddressAndPort to, RequestCallback cb) - { - requests.add(new Request(message, to, cb)); - } - - @Override - public void sendWithCallback(Message message, InetAddressAndPort to, RequestCallback cb, ConnectionType specifyConnection) - { - throw new UnsupportedOperationException(); - } - - @Override - public Future> sendWithResult(Message message, InetAddressAndPort to) - { - throw new UnsupportedOperationException(); - } - - @Override - public void respond(V response, Message message) - { - throw new UnsupportedOperationException(); - } - } - - @BeforeClass - public static void beforeClass() throws Throwable - { - ServerTestUtils.daemonInitialization(); - DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance); - - SchemaLoader.prepareServer(); - SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1), - parse("CREATE TABLE tbl (k int, c int, v int, primary key (k, c)) WITH transactional_mode='full'", "ks")); - } - - @Test - public void loadTest() throws Throwable - { - ValidatingClusterMetadataService cms = ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION); - - AccordJournal journal = null; - try - { - journal = initJournal(); - AccordConfigurationService service = new AccordConfigurationService(ID1, new AccordAgent(), new Updateable(), new Messaging(), ScheduledExecutors.scheduledTasks); - AccordJournal journal_ = journal; - TestListener listener = new TestListener(service, true) - { - @Override - public AsyncResult onTopologyUpdate(Topology topology) - { - // Fake journal save - journal_.saveTopology(new TopologyUpdate(new Int2ObjectHashMap<>(), topology), () -> {}); - return super.onTopologyUpdate(topology); - } - }; - service.registerListener(listener); - service.start(); - - Topology topology1 = createTopology(cms); - ((Updateable)service.endpointMapper()).updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1)); - service.reportTopology(topology1); - service.receiveRemoteSyncComplete(ID1, 1); - service.receiveRemoteSyncComplete(ID2, 1); - service.receiveRemoteSyncComplete(ID3, 1); - - Topology topology2 = createTopology(cms); - service.reportTopology(topology2); - service.receiveRemoteSyncComplete(ID1, 2); - - Topology topology3 = createTopology(cms); - service.reportTopology(topology3); - - AccordConfigurationService loaded = new AccordConfigurationService(ID1, new AccordAgent(), new Updateable(), new Messaging(), ScheduledExecutors.scheduledTasks); - ((EndpointMapping.Updateable)loaded.endpointMapper()).updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1)); - listener = new AbstractConfigurationServiceTest.TestListener(loaded, true); - loaded.registerListener(listener); - journal_.closeCurrentSegmentForTestingIfNonEmpty(); - List list = journal.replayTopologies(); - // Simulate journal replay - for (TopologyUpdate update : list) - loaded.reportTopology(update.global); - loaded.start(); - - listener.assertTopologiesFor(1L, 2L, 3L); - listener.assertTopologyForEpoch(1, topology1); - listener.assertTopologyForEpoch(2, topology2); - listener.assertTopologyForEpoch(3, topology3); - } - finally - { - if (journal != null) - journal.shutdown(); - } - } - - private static AccordJournal initJournal() throws Throwable - { - File directory = new File(Files.createTempDirectory("config_service_test")); - directory.deleteRecursiveOnExit(); - Keyspace ks = Schema.instance.getKeyspaceInstance("system_accord"); - ColumnFamilyStore cfs = ks.getColumnFamilyStore(AccordKeyspace.JOURNAL); - AccordJournal journal = new AccordJournal(new TestParams(), directory, cfs); - journal.start(null); - journal.unsafeSetStarted(); - return journal; - } - private static Topology createTopology(ValidatingClusterMetadataService cms) - { - ClusterMetadata previous = cms.metadata(); - ClusterMetadata.Transformer next = previous.transformer(); - maybeCreateTable(previous, next); - - ClusterMetadata metadata = next.build().metadata; - cms.setMetadata(metadata); - return AccordTopology.createAccordTopology(metadata); - } - - private static void maybeCreateTable(ClusterMetadata previous, ClusterMetadata.Transformer next) - { - Optional ks = previous.schema.getKeyspaces().get(KEYSPACE_NAME); - if (ks.isPresent()) return; - // lets create it - TableMetadata table = TableMetadata.builder(KEYSPACE_NAME, "tbl") - .id(TBL_ID) - .kind(TableMetadata.Kind.REGULAR) - .partitioner(Murmur3Partitioner.instance) - .addPartitionKeyColumn("pk", Int32Type.instance) - .build(); - KeyspaceMetadata keyspace = KeyspaceMetadata.create(KEYSPACE_NAME, KeyspaceParams.simple(ID_LIST.size())) - .withSwapped(Tables.builder().add(table).build()); - - next.with(new DistributedSchema(previous.schema.getKeyspaces().with(keyspace))); - - for (Id node : ID_LIST) - { - // not forcing the cms node id to match as they do when this logic was first added... - next.register(new NodeAddresses(getAddress(node)), - new Location("dc1", "rack1"), - NodeVersion.CURRENT); - - next.proposeToken(new NodeId(node.id), Collections.singleton(new Murmur3Partitioner.LongToken(node.id))); - } - - DataPlacement.Builder replication = DataPlacement.builder(); - Range fullRange = new Range<>(Murmur3Partitioner.MINIMUM, Murmur3Partitioner.MINIMUM); - for (int i = 0; i < ID_LIST.size(); i++) - { - InetAddressAndPort address = getAddress(ID_LIST.get(i)); - Replica replica = new Replica(address, fullRange, true); - replication.withReadReplica(next.epoch(), replica).withWriteReplica(next.epoch(), replica); - } - next.with(previous.placements.unbuild().with(keyspace.params.replication, replication.build()).build()); - } - - private static InetAddressAndPort getAddress(Id node) - { - try - { - return InetAddressAndPort.getByAddress(new byte[]{127, 0, 0, (byte) node.id}); - } - catch (UnknownHostException e) - { - throw new RuntimeException(e); - } - } -} diff --git a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java index 75b59b3f94..fdb992947b 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java @@ -41,24 +41,28 @@ import com.google.common.collect.Sets; import org.junit.BeforeClass; import org.junit.Test; +import accord.Utils; import accord.api.Agent; -import accord.impl.AbstractTestConfigurationService; +import accord.api.MessageSink; +import accord.api.TopologyListener; +import accord.api.TopologySorter; import accord.impl.TestAgent; import accord.impl.basic.Pending; import accord.impl.basic.PendingQueue; import accord.impl.basic.MonitoredPendingQueue; import accord.impl.basic.RandomDelayQueue; import accord.impl.basic.SimulatedDelayedExecutorService; +import accord.impl.mock.MockCluster; import accord.local.Node; import accord.primitives.Range; import accord.primitives.Ranges; import accord.topology.Topology; +import accord.topology.TopologyManager; import accord.utils.AccordGens; import accord.utils.Gen; import accord.utils.Gens; import accord.utils.RandomSource; import accord.utils.SortedArrays.SortedArrayList; -import accord.utils.SortedList; import accord.utils.SortedListSet; import org.apache.cassandra.concurrent.AdaptingScheduledExecutorPlus; import org.apache.cassandra.concurrent.ScheduledExecutorPlus; @@ -72,7 +76,6 @@ import org.apache.cassandra.net.ConnectionType; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessageDelivery; import org.apache.cassandra.net.RequestCallback; -import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.tcm.ValidatingClusterMetadataService; import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.utils.AccordGenerators; @@ -82,6 +85,7 @@ import org.assertj.core.api.Assertions; import static accord.utils.Property.qt; import static org.apache.cassandra.simulator.RandomSource.Choices.choose; +import static org.apache.cassandra.utils.AccordGenerators.partitioner; public class AccordSyncPropagatorTest { @@ -95,14 +99,14 @@ public class AccordSyncPropagatorTest @Test public void burnTest() { - Gen rangesGen = AccordGenerators.ranges().filter(r -> !r.isEmpty()); + Gen> rangesGenGen = partitioner().map(p -> AccordGenerators.ranges(p).filter(r -> !r.isEmpty())); Gen> nodesGen = Gens.lists(AccordGens.nodes()).unique().ofSizeBetween(1, 40); - qt().withExamples(100).check(rs -> { + qt().withExamples(10).check(rs -> { // when gossip and cluster metadata don't know an endpoint, retries are avoided (node removed) // so when instances are created here they are added to gossip to trick the membership check... Gossiper.instance.clearUnsafe(); - SortedList nodes = SortedArrayList.copyUnsorted(nodesGen.next(rs), Node.Id[]::new); + SortedArrayList nodes = SortedArrayList.copyUnsorted(nodesGen.next(rs), Node.Id[]::new); SortedListSet nodesAsSet = SortedListSet.allOf(nodes); List failures = new ArrayList<>(); @@ -112,22 +116,31 @@ public class AccordSyncPropagatorTest SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, agent, null); ScheduledExecutorPlus scheduler = new AdaptingScheduledExecutorPlus(globalExecutor); - Cluster cluster = new Cluster(nodes, rs, scheduler); - - long epochOffset = rs.nextLong(1, 1024); - int numEpochs = rs.nextInt(1, 10); + final long epochOffset = rs.nextLong(1, 1024); Map allRanges = new HashMap<>(); + Cluster cluster = new Cluster(nodes, epochOffset, allRanges, rs, scheduler); + + Gen rangesGen = rangesGenGen.next(rs); + int numEpochs = rs.nextInt(1, 10); Pending.Global.setNoActiveOrigin(); + Set prefixes = new HashSet<>(); for (int i = 0; i < numEpochs; i++) { long epoch = epochOffset + i; - Ranges ranges = rangesGen.next(rs); - allRanges.put(epoch, ranges); + Ranges ranges; + { + Ranges tmp = rangesGen.next(rs); + while (tmp.stream().anyMatch(r -> prefixes.contains(r.prefix()))) + tmp = rangesGen.next(rs); + ranges = tmp; + } + ranges.stream().forEach(r -> prefixes.add(r.prefix())); + allRanges.put(epoch, ranges.mergeTouching()); scheduler.schedule(() -> { for (Node.Id nodeId : nodes) { - cluster.node(nodeId).configurationService.receiveRemoteSyncComplete(nodeId, epoch); - cluster.node(nodeId).propagator.reportCoordinationReady(epoch, nodes, nodeId); + cluster.node(nodeId).topology.onReadyToCoordinate(nodeId, epoch); + cluster.node(nodeId).propagator.onReadyToCoordinate(epoch, nodes); } for (int j = 0, attempts = rs.nextInt(1, 4); j < attempts; j++) @@ -137,8 +150,12 @@ public class AccordSyncPropagatorTest Cluster.Instance inst = cluster.node(choose(rs, nodesAsSet)); scheduler.schedule(() -> { Ranges subrange = Ranges.of(range); - inst.propagator.reportClosed(epoch, nodes, subrange); - scheduler.schedule(() -> inst.propagator.reportRetired(epoch, nodes, subrange), 1, TimeUnit.MINUTES); + inst.topology.onEpochClosed(subrange, epoch); + inst.propagator.onEpochClosed(subrange, epoch, nodes); + scheduler.schedule(() -> { + inst.topology.onEpochRetired(subrange, epoch); + inst.propagator.onEpochRetired(subrange, epoch, nodes); + }, 1, TimeUnit.MINUTES); }, rs.nextInt(30, 300), TimeUnit.SECONDS); } } @@ -166,12 +183,16 @@ public class AccordSyncPropagatorTest for (Cluster.Instance inst : cluster.instances.values()) { - Cluster.ConfigService cs = inst.configurationService; + Cluster.Listener cs = inst.listener; assertSetsEqual(cs.completedEpochs, allRanges.keySet(), "completedEpochs %s", inst.id); assertSetsEqual(cs.syncCompletes.keySet(), allRanges.keySet(), "syncCompletes %s", inst.id); for (Map.Entry> e : cs.syncCompletes.entrySet()) assertSetsEqual(e.getValue(), nodesAsSet, "syncCompletes values on %s", inst.id); + for (Map.Entry e : cs.closed.entrySet()) + e.setValue(e.getValue().mergeTouching()); + for (Map.Entry e : cs.redundant.entrySet()) + e.setValue(e.getValue().mergeTouching()); assertMapEquals(cs.closed, allRanges, "Unexpected state for closed on %s", inst.id); assertMapEquals(cs.redundant, allRanges, "Unexpected state for redundant on %s", inst.id); } @@ -195,7 +216,7 @@ public class AccordSyncPropagatorTest V value = e.getValue(); V other = expected.get(e.getKey()); if (!Objects.equals(value, other)) - errors.add(String.format("Missmatch at key %s: expected %s but given %s", e.getKey(), other, value)); + errors.add(String.format("Mismatch at key %s: expected %s but given %s", e.getKey(), other, value)); } if (!errors.isEmpty()) throw new AssertionError(String.join("\n", errors)); @@ -214,6 +235,8 @@ public class AccordSyncPropagatorTest private final ScheduledExecutorPlus scheduler; private Cluster(List nodes, + long minEpoch, + Map allRanges, RandomSource rs, ScheduledExecutorPlus scheduler) { @@ -225,10 +248,16 @@ public class AccordSyncPropagatorTest { InetAddressAndPort address = addressFromInt(id.id); nodeToAddress.put(id, address); - ConfigService cs = new ConfigService(id, nodes); + Node node = Utils.createNode(id, Topology.EMPTY, new MessageSink.NoOpSink(), new MockCluster.Clock(0)); + TopologyManager topology = new TopologyManager((TopologySorter.StaticSorter)(a,b,c)->0, node, ignore -> { throw new UnsupportedOperationException(); }, null, null); + Listener cs = new Listener(id, minEpoch, nodes, allRanges); Sink sink = new Sink(id); FailureWrapper fw = new FailureWrapper(Cluster.this, id); - instances.put(id, new Instance(id, cs, sink, new AccordSyncPropagator(id, fw, sink, scheduler, cs))); + AccordSyncPropagator propagator = new AccordSyncPropagator(id, fw, sink, scheduler); + topology.addListener(propagator); + topology.addListener(cs); + propagator.setTestListener(cs); + instances.put(id, new Instance(id, topology, cs, sink, propagator)); Gossiper.instance.endpointStateMap.put(address, new EndpointState(HeartBeatState.empty())); } this.nodeToAddress = nodeToAddress.build(); @@ -326,7 +355,7 @@ public class AccordSyncPropagatorTest throw new IllegalStateException("Unknown action: " + action); } callbacks.put(message.id(), cb); - scheduler.schedule(() -> AccordService.receive(this, node(to).configurationService, (Message) message.withFrom(mappedEndpointOrNull(from))), 500, TimeUnit.MILLISECONDS); + scheduler.schedule(() -> AccordService.receive(this, node(to).topology, (Message) message.withFrom(mappedEndpointOrNull(from))), 500, TimeUnit.MILLISECONDS); scheduler.schedule(() -> { RequestCallback removed = callbacks.remove(message.id()); if (removed != null) @@ -413,56 +442,58 @@ public class AccordSyncPropagatorTest } } - private class ConfigService extends AbstractTestConfigurationService implements AccordSyncPropagator.Listener + private static class Listener implements AccordSyncPropagator.TestListener, TopologyListener { + private final Node.Id self; + private final long minEpoch; private final List nodes; private final Map> syncCompletes = new HashMap<>(); private final Map> endpointAcks = new HashMap<>(); private final NavigableSet completedEpochs = Collections.synchronizedNavigableSet(new TreeSet<>()); + private final Map allRanges; private final Map closed = new HashMap<>(); private final Map redundant = new HashMap<>(); - private ConfigService(Node.Id node, List nodes) + private Listener(Node.Id node, long minEpoch, List nodes, Map allRanges) { - super(node, new AccordAgent()); + this.self = node; + this.minEpoch = minEpoch; this.nodes = nodes; + this.allRanges = allRanges; } @Override - protected void receiveRemoteSyncCompletePreListenerNotify(Node.Id node, long epoch) + public void onRemoteReadyToCoordinate(Node.Id node, long epoch) { syncCompletes.computeIfAbsent(epoch, ignore -> new HashSet<>()).add(node); } @Override - public void fetchTopologyForEpoch(long epoch) + public void onEpochClosed(Ranges ranges, long epoch, @Nullable Topology topology) { - // TODO + merge(closed, ranges, epoch); } @Override - protected void onReadyToCoordinate(Topology topology) + public void onEpochRetired(Ranges ranges, long epoch, @Nullable Topology topology) { - instances.get(localId).propagator.reportCoordinationReady(topology.epoch(), topology.nodes(), localId); + merge(redundant, ranges, epoch); } - @Override - public void reportEpochClosed(Ranges ranges, long epoch) - { - Topology topology = getTopologyForEpoch(epoch); - instances.get(localId).propagator.reportClosed(epoch, topology.nodes(), ranges); - } - - @Override - public void reportEpochRetired(Ranges ranges, long epoch) - { - Topology topology = getTopologyForEpoch(epoch); - instances.get(localId).propagator.reportRetired(epoch, topology.nodes(), ranges); - } - - @Override - public void reportEpochRemoved(long epoch) + private void merge(Map map, Ranges ranges, long epoch) { + while (epoch >= minEpoch) + { + Ranges cur = map.get(epoch); + Ranges upd = cur == null ? ranges : cur.with(ranges); + if (upd == cur) + break; + if (cur != null) + ranges = ranges.without(cur); + ranges = ranges.without(allRanges.get(epoch)); + map.put(epoch, upd); + --epoch; + } } @Override @@ -472,42 +503,25 @@ public class AccordSyncPropagatorTest if (acks.add(id) && acks.containsAll(nodes)) completedEpochs.add(epoch); } - - @Override - public void onEndpointNack(Node.Id id, long epoch) - { - onEndpointAck(id, epoch); - } - - @Override - public synchronized void receiveClosed(Ranges ranges, long epoch) - { - super.receiveClosed(ranges, epoch); - closed.merge(epoch, ranges, Ranges::with); - } - - @Override - public synchronized void receiveRetired(Ranges ranges, long epoch) - { - super.receiveRetired(ranges, epoch); - redundant.merge(epoch, ranges, Ranges::with); - } } public class Instance { private final Node.Id id; - private final ConfigService configurationService; + private final TopologyManager topology; + private final Listener listener; private final Sink messagingService; private final AccordSyncPropagator propagator; private Instance(Node.Id id, - ConfigService configurationService, + TopologyManager topology, + Listener listener, Sink messagingService, AccordSyncPropagator propagator) { this.id = id; - this.configurationService = configurationService; + this.topology = topology; + this.listener = listener; this.messagingService = messagingService; this.propagator = propagator; } diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java index 406d65aad8..63051f72f8 100644 --- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java +++ b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java @@ -39,8 +39,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; import java.util.function.Predicate; -import java.util.stream.Collectors; -import java.util.stream.LongStream; +import java.util.function.Supplier; import javax.annotation.Nullable; @@ -49,14 +48,17 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import accord.api.ConfigurationService; -import accord.api.ConfigurationService.EpochReady; +import accord.Utils; +import accord.api.MessageSink; +import accord.impl.TestAgent; +import accord.impl.mock.MockCluster; +import accord.local.ShardDistributor; import accord.impl.DefaultTimeouts; import accord.impl.SizeOfIntersectionSorter; -import accord.impl.TestAgent; import accord.local.Node; import accord.local.TimeService; import accord.primitives.Ranges; +import accord.topology.EpochReady; import accord.topology.Topology; import accord.topology.TopologyManager; import accord.utils.Gen; @@ -66,6 +68,7 @@ import accord.utils.RandomSource; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; import accord.utils.async.AsyncResult; +import accord.utils.async.AsyncResults; import accord.utils.async.Cancellable; import org.apache.cassandra.concurrent.ScheduledExecutorPlus; import org.apache.cassandra.concurrent.SimulatedExecutorFactory; @@ -86,8 +89,7 @@ import org.apache.cassandra.schema.ReplicationParams; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.TableParams; import org.apache.cassandra.schema.Tables; -import org.apache.cassandra.service.accord.AccordConfigurationService.EpochSnapshot; -import org.apache.cassandra.service.accord.api.AccordAgent; +import org.apache.cassandra.service.accord.api.TokenKey; import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadataService; @@ -112,6 +114,8 @@ import org.assertj.core.description.Description; import static accord.utils.Property.commands; import static accord.utils.Property.stateful; +import static org.apache.cassandra.config.DatabaseDescriptor.getAccordCommandStoreShardCount; +import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner; public class EpochSyncTest { @@ -128,7 +132,7 @@ public class EpochSyncTest @Test public void test() { - stateful().withSeed(1).withExamples(50).withSteps(500).check(commands(() -> Cluster::new) + stateful().withSeed(152472217520379L).withExamples(50).withSteps(500).check(commands(() -> Cluster::new) .destroyState(cluster -> { finishPendingWork(cluster); cluster.processAll(); @@ -195,7 +199,7 @@ public class EpochSyncTest private static class Cluster { - private static final int rf = 2; + private static final int rf = 3; private static final ReplicationParams replication_params = ReplicationParams.simple(rf); private final RandomSource rs; @@ -290,7 +294,8 @@ public class EpochSyncTest private boolean hasPendingWork() { return !status(s -> s == Cluster.Status.Registered).isEmpty() - || !cms.metadata().inProgressSequences.isEmpty(); + || !cms.metadata().inProgressSequences.isEmpty() + || globalExecutor.hasWork(); } private boolean hasNoPendingWork() @@ -334,7 +339,7 @@ public class EpochSyncTest return metadata.placements.get(replication_params).reads.byEndpoint().keySet().contains(address.broadcastAddress); } - public enum EpochTracker { topologyManager, accordSyncPropagator, configurationService} + public enum EpochTracker { topologyManager, accordSyncPropagator } Set globalSynced(long epoch) { @@ -376,38 +381,34 @@ public class EpochSyncTest { Instance inst = instances.get(id); if (removed.contains(id)) continue; // ignore removed nodes - AccordConfigurationService conf = inst.config; TopologyManager tm = inst.topology; for (long epoch = Math.max(tm.firstNonEmpty(), inst.epoch.getEpoch()); epoch <= cms.metadata().epoch.getEpoch(); epoch++) { // validate config - EpochSnapshot snapshot = conf.getEpochSnapshot(epoch); if (isDone) { - Assertions.assertThat(snapshot).describedAs("node%s does not have epoch %d", id, epoch).isNotNull(); - Assertions.assertThat(snapshot.syncStatus).isEqualTo(AccordConfigurationService.SyncStatus.COMPLETED); + Assertions.assertThat(inst.topologyService.syncPropagator().hasPending(epoch)) + .describedAs("node%s has pending notifications in AccordSyncPropagator for epoch %d", id, epoch) + .isFalse(); // validate topology manager - Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue(); - Ranges ranges = tm.globalForEpoch(epoch).ranges().mergeTouching(); - Ranges actual = tm.syncComplete(epoch).mergeTouching(); + Ranges ranges = tm.active().globalForEpoch(epoch).ranges().mergeTouching(); + Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching(); Assertions.assertThat(actual) .describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.without(actual)) .isEqualTo(ranges); } else { - if (snapshot == null || snapshot.syncStatus != AccordConfigurationService.SyncStatus.COMPLETED) continue; - if (!allSynced(epoch)) continue; - Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue(); - Topology topology = tm.globalForEpoch(epoch); + Assertions.assertThat(tm.active().hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue(); + Topology topology = tm.active().globalForEpoch(epoch); Ranges ranges = topology.ranges().mergeTouching(); - Ranges actual = tm.syncComplete(epoch).mergeTouching(); + Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching(); // TopologyManager defines syncComplete for an epoch as (epoch - 1).syncComplete. This means that an epoch has reached quorum, but will still miss ranges as previous epochs have not - if (!ranges.equals(actual) && tm.minEpoch() != epoch && !ranges.equals(tm.syncComplete(epoch - 1).mergeTouching())) + if (!ranges.equals(actual) && tm.minEpoch() != epoch && !ranges.equals(tm.unsafeQuorumReady(epoch - 1).mergeTouching())) continue; long epoch_ = epoch; Assertions.assertThat(actual) @@ -415,11 +416,8 @@ public class EpochSyncTest { public String value() { - return String.format("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s; previous epochs %s", - id, epoch_, ranges.without(actual), topology.nodes(), - LongStream.range(inst.epoch.getEpoch(), epoch_ + 1) - .mapToObj(e -> String.format("%d -> %s(synced=%s): %s", e, conf.getEpochSnapshot(e).syncStatus, globalSynced(e), tm.syncComplete(e))) - .collect(Collectors.joining("\n"))); + return String.format("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s", + id, epoch_, ranges.without(actual), topology.nodes()); } }) @@ -548,7 +546,7 @@ public class EpochSyncTest return Action.DELIVER; }, SimulatedMessageDelivery.randomDelay(rs.fork()), - (to, msg) -> instances.get(nodeId(to)).reciver.recieve(msg), + (to, msg) -> instances.get(nodeId(to)).receiver.recieve(msg), (action, to, msg) -> logger.trace("{} message {}", action, msg), scheduler::schedule, failures::add); @@ -594,6 +592,7 @@ public class EpochSyncTest Instance inst = Objects.requireNonNull(instances.get(pick), "Unknown id " + pick); Invariants.require(!removed.contains(pick), "Can not remove node twice; node " + pick); removed.add(pick); + Assertions.assertThat(inst.topologyService.syncPropagator().hasPending()).isFalse(); inst.status = Status.Leaving; PrepareLeave prepareLeave = new PrepareLeave(new NodeId(pick.id), false, new UniformRangePlacement(), LeaveStreams.Kind.REMOVENODE); notify(process(prepareLeave).metadata); @@ -615,7 +614,8 @@ public class EpochSyncTest { Instance inst = instances.get(id); inst.maybeTransition(current, t); - inst.config.maybeReportMetadata(current); + Topology topology = AccordTopology.createAccordTopology(current); + inst.topology.reportTopology(topology); } } @@ -650,72 +650,44 @@ public class EpochSyncTest { private final Node.Id id; private final long token; - private final AccordConfigurationService config; + private final AccordTopologyService topologyService; private final SimulatedMessageDelivery messaging; - private final SimulatedMessageDelivery.SimulatedMessageReceiver reciver; + private final SimulatedMessageDelivery.SimulatedMessageReceiver receiver; private final TopologyManager topology; private final Epoch epoch; private Status status = Status.Init; - Instance(Node.Id node, long token, Epoch epoch, SimulatedMessageDelivery messagingService, AccordEndpointMapper mapper) + Instance(Node.Id id, long token, Epoch epoch, SimulatedMessageDelivery messagingService, AccordEndpointMapper mapper) { - this.id = node; + this.id = id; this.token = token; this.epoch = epoch; + MockCluster.Clock clock = new MockCluster.Clock(0); + Node node = Utils.createNode(id, ignore -> AsyncResults.settable(), new MessageSink.NoOpSink(), clock, new TestAgent(clock), new TokenKey.KeyspaceSplitter(new ShardDistributor.EvenSplit<>(getAccordCommandStoreShardCount(), getPartitioner().accordSplitter()))); // TODO (review): Should there be a real scheduler here? Is it possible to adapt the Scheduler interface to scheduler used in this test? TimeService time = TimeService.ofNonMonotonic(globalExecutor::currentTimeMillis, TimeUnit.MILLISECONDS); - this.topology = new TopologyManager(SizeOfIntersectionSorter.SUPPLIER, new TestAgent.RethrowAgent(), id, time, new DefaultTimeouts(time)); - config = new AccordConfigurationService(node, new AccordAgent(), mapper, messagingService, scheduler); - config.registerListener(new ConfigurationService.Listener() + this.topologyService = new AccordTopologyService(id, mapper, messagingService, scheduler); + this.topology = new TopologyManager(SizeOfIntersectionSorter.SUPPLIER, node, topologyService, time, new DefaultTimeouts(time)) { @Override - public AsyncResult onTopologyUpdate(Topology topology) + protected EpochReady bootstrap(Supplier bootstrap) { + bootstrap.get(); AsyncResult metadata = schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable) () -> null).beginAsResult(); AsyncResult coordination = metadata.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable) () -> null).beginAsResult()); AsyncResult data = coordination.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable) () -> null).beginAsResult()); AsyncResult reads = data.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable) () -> null).beginAsResult()); - EpochReady ready = new EpochReady(topology.epoch(), metadata, coordination, data, reads); - - topology().onTopologyUpdate(topology, () -> ready, e -> {}); - ready.coordinate.invokeIfSuccess(() -> topology().onEpochSyncComplete(id, topology.epoch())); - if (topology().minEpoch() == topology.epoch() && topology().epoch() != topology.epoch()) - return ready.coordinate; - config.acknowledgeEpoch(ready); - return ready.coordinate; + return new EpochReady(topology.epoch(), metadata, coordination, data, reads); } - - @Override - public void onRemoteSyncComplete(Node.Id node, long epoch) - { - topology.onEpochSyncComplete(node, epoch); - } - - @Override - public void onRemoveNode(long epoch, Node.Id removed) - { - // TODO - //topology.onRemoveNode(epoch, removed); - } - - @Override - public void onEpochClosed(Ranges ranges, long epoch) - { - topology.onEpochClosed(ranges, epoch); - } - - @Override - public void onEpochRetired(Ranges ranges, long epoch) - { - topology.onEpochRetired(ranges, epoch); - } - }); + }; + this.topology.addListener(topologyService.syncPropagator()); + this.topology.addListener(topologyService.watermarkCollector()); Map> handlers = new EnumMap<>(Verb.class); //noinspection unchecked - handlers.put(Verb.ACCORD_SYNC_NOTIFY_REQ, msg -> AccordService.receive(messagingService, config, (Message) (Message) msg)); + handlers.put(Verb.ACCORD_SYNC_NOTIFY_REQ, msg -> AccordService.receive(messagingService, topology, (Message) (Message) msg)); this.messaging = messagingService; - this.reciver = messagingService.receiver(new SimulatedMessageDelivery.SimpleVerbHandler(handlers)); + this.receiver = messagingService.receiver(new SimulatedMessageDelivery.SimpleVerbHandler(handlers)); } @Override @@ -736,7 +708,6 @@ public class EpochSyncTest case Init: Invariants.require(!t.nodes().contains(id), "Node was in Init state but present in the Topology!"); Invariants.require(current.directory.peerId(address(id)) != null, "Node exists but not in TCM"); - start(); status = Status.Registered; break; case Registered: @@ -762,26 +733,13 @@ public class EpochSyncTest } } - private void start() - { - config.start(); - } - - TopologyManager topology() - { - return topology; - } - Set synced(long epoch) { if (epoch < this.epoch.getEpoch()) throw new IllegalArgumentException("Asked for epoch before this instance existed"); EnumSet done = EnumSet.noneOf(EpochTracker.class); - EpochSnapshot snapshot = config.getEpochSnapshot(epoch); - if (snapshot != null && snapshot.syncStatus == AccordConfigurationService.SyncStatus.COMPLETED) - done.add(EpochTracker.configurationService); - if (topology.hasReachedQuorum(epoch)) + if (topology.unsafeIsQuorumReady(epoch)) done.add(EpochTracker.topologyManager); - if (!config.syncPropagator().hasPending(epoch)) + if (!topologyService.syncPropagator().hasPending(epoch)) done.add(EpochTracker.accordSyncPropagator); return done; } diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 01c52656c5..786b38f7ed 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -674,6 +674,7 @@ public class CommandsForKeySerializerTest @Override public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil blockedUntil, TimeUnit units) { return 0; } @Override public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil retrying, TimeUnit units) { return 0; } @Override public long retrySyncPointDelay(Node node, int attempt, TimeUnit units) { return 0; } + @Override public long retryTopologyDelay(Node node, int attempt, TimeUnit units) { return 0; } @Override public long retryDurabilityDelay(Node node, int attempt, TimeUnit units) { return 0; } @Override public long expireEpochWait(TimeUnit units) { return 0; } @Override public long expiresAt(ReplyContext replyContext, TimeUnit unit) { return 0; } diff --git a/test/unit/org/apache/cassandra/utils/AccordGenerators.java b/test/unit/org/apache/cassandra/utils/AccordGenerators.java index 0fdd6f8dd8..3b4fad9a0f 100644 --- a/test/unit/org/apache/cassandra/utils/AccordGenerators.java +++ b/test/unit/org/apache/cassandra/utils/AccordGenerators.java @@ -63,7 +63,7 @@ import accord.primitives.TxnId; import accord.primitives.Writes; import accord.topology.Shard; import accord.topology.Topology; -import accord.topology.TopologyManager; +import accord.topology.TopologyRange; import accord.utils.AccordGens; import accord.utils.Gen; import accord.utils.Gens; @@ -807,7 +807,7 @@ public class AccordGenerators }; } - public static Gen topologyRangeGen() + public static Gen topologyRangeGen() { Gen.LongGen epochGen = AccordGens.epochs(); return rs -> { @@ -823,7 +823,7 @@ public class AccordGenerators if (minEpoch == Timestamp.MAX_EPOCH) { // not possible to have a list of values, so to simplfiy just return empty - return new TopologyManager.TopologyRange(Timestamp.MAX_EPOCH, Timestamp.MAX_EPOCH, -1, Collections.emptyList()); + return new TopologyRange(Timestamp.MAX_EPOCH, Timestamp.MAX_EPOCH, -1, Collections.emptyList()); } long epochsRemaining = Timestamp.MAX_EPOCH - minEpoch; int size = rs.nextInt(1, Math.toIntExact(Math.min(100, epochsRemaining))); @@ -841,7 +841,7 @@ public class AccordGenerators firstNonEmpty = t.epoch(); topologies.add(t); } - return new TopologyManager.TopologyRange(topologies.get(0).epoch(), topologies.get(topologies.size() - 1).epoch(), firstNonEmpty, topologies); + return new TopologyRange(topologies.get(0).epoch(), topologies.get(topologies.size() - 1).epoch(), firstNonEmpty, topologies); }; }