From aa5c3aba1d5a2fd3812f8062502fe85cfa0c4757 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Tue, 28 Oct 2025 18:52:06 +0000 Subject: [PATCH] Improve Topology Management Merge ConfigurationService with TopologyManager to remove cyclic dependency and duplicated work. Also: - Improve visibility of work blocking topology processing - Ensure we cannot double-count peers when deciding an epoch's distributed readiness - Remove the possibility of distributed stalls, by processing new topologies as soon as a contiguous sequence is known locally, regardless of whether anyprior local epoch is ready to coordinate or has recorded this fact to peers. The notification of local readiness continues to be processed only once all prior epochs are ready, but we can begin using and readying later epochs immediately. patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20998 --- modules/accord | 2 +- .../db/virtual/AccordDebugKeyspace.java | 7 +- .../db/virtual/AccordVirtualTables.java | 45 +- .../org/apache/cassandra/service/Rebuild.java | 2 +- .../accord/AccordConfigurationService.java | 654 ------------------ .../accord/AccordFastPathCoordinator.java | 29 +- .../service/accord/AccordService.java | 118 +++- .../service/accord/AccordSyncPropagator.java | 129 ++-- .../service/accord/AccordTopology.java | 2 +- .../service/accord/AccordTopologyService.java | 204 ++++++ .../service/accord/AccordVerbHandler.java | 2 +- .../service/accord/FetchTopologies.java | 5 +- .../service/accord/IAccordService.java | 26 +- .../service/accord/WatermarkCollector.java | 32 +- .../service/accord/api/AccordAgent.java | 9 +- .../interop/AccordInteropExecution.java | 6 +- .../accord/journal/AccordTopologyUpdate.java | 19 +- .../service/accord/repair/AccordRepair.java | 2 +- .../accord/serializers/DepsSerializers.java | 1 - .../tcm/sequences/BootstrapAndJoin.java | 4 +- .../tcm/sequences/DropAccordTable.java | 2 +- .../apache/cassandra/tcm/sequences/Move.java | 2 +- .../tcm/transformations/PrepareJoin.java | 5 +- .../distributed/shared/ClusterUtils.java | 2 +- .../test/accord/AccordBootstrapTest.java | 9 +- .../test/accord/AccordBootstrapTestBase.java | 34 +- .../test/accord/AccordCQLTestBase.java | 2 +- .../test/accord/AccordMoveTest.java | 1 - ...AccordRecoverFromAvailabilityLossTest.java | 5 +- .../test/accord/AccordSimpleFastPathTest.java | 6 +- .../test/accord/AccordTestBase.java | 3 +- .../journal/AccordJournalReplayTest.java | 2 +- .../simulator/test/EpochStressTest.java | 32 +- .../db/virtual/AccordVirtualTablesTest.java | 45 +- .../index/accord/RouteIndexTest.java | 2 +- .../AccordConfigurationServiceTest.java | 295 -------- .../accord/AccordSyncPropagatorTest.java | 150 ++-- .../service/accord/EpochSyncTest.java | 140 ++-- .../CommandsForKeySerializerTest.java | 1 + .../cassandra/utils/AccordGenerators.java | 8 +- 40 files changed, 664 insertions(+), 1380 deletions(-) delete mode 100644 src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java create mode 100644 src/java/org/apache/cassandra/service/accord/AccordTopologyService.java delete mode 100644 test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java 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); }; }