From 67e21ec4ba585e1d62598aea663055990361cbee Mon Sep 17 00:00:00 2001 From: Alex Petrov Date: Fri, 7 Feb 2025 19:04:05 +0100 Subject: [PATCH] Fix AccordMigrationTest, preclude possible races in topology propagation Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20316 --- modules/accord | 2 +- .../accord/AccordConfigurationService.java | 64 +++++++++++++++---- .../service/accord/AccordJournal.java | 1 - .../service/accord/AccordService.java | 5 +- .../service/accord/AccordVerbHandler.java | 17 +++-- .../AccordMigrationWriteRaceTestBase.java | 16 +++++ .../test/accord/AccordTestBase.java | 2 + .../MigrationFromAccordWriteRaceTest.java | 3 - .../fuzz/ring/ConsistentBootstrapTest.java | 1 + .../service/accord/AccordJournalBurnTest.java | 11 +++- .../simulator/cluster/ClusterActions.java | 3 +- .../simulator/paxos/PaxosSimulation.java | 4 ++ .../accord/AccordSyncPropagatorTest.java | 2 +- 13 files changed, 101 insertions(+), 30 deletions(-) diff --git a/modules/accord b/modules/accord index 1542da5d8a..a4a8255beb 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 1542da5d8acf08b8226227c91d4f81c64ed9762c +Subproject commit a4a8255bebb6804b8f55289f2fbc846a5253e66c diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java index 9d8b41bcb1..271fc3b219 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java @@ -22,7 +22,9 @@ 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.TimeUnit; +import java.util.function.BiConsumer; import java.util.stream.Collectors; import javax.annotation.Nullable; import javax.annotation.concurrent.GuardedBy; @@ -59,7 +61,6 @@ 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.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static org.apache.cassandra.service.accord.AccordTopology.tcmIdToAccord; import static org.apache.cassandra.utils.Simulate.With.MONITORS; @@ -448,21 +449,60 @@ public class AccordConfigurationService extends AbstractConfigurationService reportMetadata(metadata)); } + private final Map> pendingTopologies = new ConcurrentHashMap<>(); + @Override - protected void fetchTopologyInternal(long epoch) + 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 void fetchTopologyInternal(long epoch) + { + pendingTopologies.computeIfAbsent(epoch, (epoch_) -> { + AsyncPromise future = new AsyncPromise<>(); + fetchTopologyAsync(epoch_, + (topology, throwable) -> { + if (topology != null) + future.setSuccess(topology); + else + { + Invariants.require(future == pendingTopologies.remove(epoch_)); + 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(() -> { - if (ClusterMetadata.current().epoch.getEpoch() < epoch) - ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(epoch)); - + 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(AccordTopology.createAccordTopology(metadata), null); return; + } try { @@ -475,18 +515,18 @@ public class AccordConfigurationService extends AbstractConfigurationService= epoch) - return; - Thread.currentThread().interrupt(); - throw new UncheckedInterruptedException(e); + onResult.accept(topology, null); } catch (Throwable e) { if (currentEpoch() >= epoch) + { + onResult.accept(getTopologyForEpoch(epoch), null); return; + } + if (e instanceof InterruptedException) + Thread.currentThread().interrupt(); + onResult.accept(null, e); throw new RuntimeException(e.getCause()); } }); diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 4614ba24ba..dd9f0b0e85 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -440,7 +440,6 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier @SuppressWarnings("unchecked") @Override public void replay(CommandStores commandStores) { - journal.closeCurrentSegmentForTestingIfNonEmpty(); try (CloseableIterator> iter = journalTable.keyIterator()) { while (iter.hasNext()) diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 87d0fa95c3..5263a0e9c2 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -261,8 +261,8 @@ public class AccordService implements IAccordService, Shutdownable // when replacing another node but using the same ip the hostId will also match, this causes no TCM transactions // to be committed... // In order to bootup correctly, need to pull in the current epoch - ClusterMetadata current = ClusterMetadata.current(); - as.configurationService().listener.notifyPostCommit(current, current, false); + ClusterMetadata metadata = ClusterMetadata.current(); + as.configurationService().listener.notifyPostCommit(metadata, metadata, false); } instance = as; @@ -381,6 +381,7 @@ public class AccordService implements IAccordService, Shutdownable { Journal.TopologyUpdate update = iter.next(); local.add(update.global); + Invariants.require(lastSeen == null || update.global.epoch() > lastSeen.global.epoch()); lastSeen = update; } diff --git a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java index 48cd307ccd..61f18867da 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java +++ b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java @@ -72,17 +72,20 @@ public class AccordVerbHandler implements IVerbHandler request.process(node, fromNodeId, message.header); else { + // TODO (required): review this claim. Downstream from `withEpoch`, we do call fetch log, albeit from _CMS_, since we + // do not know the peer there. // withEpoch does not reliably ensure that TCM is up to date, if Accord has the topology it won't // wait for TCM to come up to date, so do it here in the verb handler if (!cmUpToDate) { - ClusterMetadataService.instance().fetchLogFromPeerOrCMSAsync(cm, message.from(), Epoch.create(waitForEpoch)).addCallback((success, failure) -> - node.withEpoch(waitForEpoch, (ignored, withEpochFailure) -> { - if (withEpochFailure != null) - throw new RuntimeException("Timed out waiting for epoch when processing message from " + fromNodeId + " to " + node + " message " + message, withEpochFailure); - request.process(node, fromNodeId, message.header); - }) - ); + ClusterMetadataService.instance().fetchLogFromPeerOrCMSAsync(cm, message.from(), Epoch.create(waitForEpoch)) + .addCallback((success, failure) -> { + node.withEpoch(waitForEpoch, (ignored, withEpochFailure) -> { + if (withEpochFailure != null) + throw new RuntimeException("Timed out waiting for epoch when processing message from " + fromNodeId + " to " + node + " message " + message, withEpochFailure); + request.process(node, fromNodeId, message.header); + }); + }); } else { diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationWriteRaceTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationWriteRaceTestBase.java index 6984fa92f7..9b736ec67b 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationWriteRaceTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationWriteRaceTestBase.java @@ -275,6 +275,22 @@ public abstract class AccordMigrationWriteRaceTestBase extends AccordTestBase ConsensusKeyMigrationState.reset(); }); truncateSystemTables(); + ClusterUtils.waitForCMSToQuiesce(SHARED_CLUSTER, 1); + SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> ClusterMetadata.current().epoch.getEpoch() == + ((AccordService) AccordService.instance()).configurationService().currentEpoch() && + AccordService.instance().topology().current().epoch() == + ((AccordService) AccordService.instance()).configurationService().currentEpoch(), + 60)); + SHARED_CLUSTER.forEach(() -> { + try + { + AccordService.instance().epochReady(ClusterMetadata.current().epoch).get(30, TimeUnit.SECONDS); + } + catch (Throwable e) + { + throw new RuntimeException(e); + } + }); } private ListenableFuture alterTableTransactionalModeAsync(TransactionalMode mode) 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 9a3e81d74c..f67bc07c05 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java @@ -371,6 +371,8 @@ public abstract class AccordTestBase extends TestBaseImpl .set("write_request_timeout", "10s") .set("transaction_timeout", "15s") .set("native_transport_timeout", "30s") + .set("cms_await_timeout", "1s") + .set("cms_default_max_retries", 10_000) .set("accord.ephemeral_read_enabled", "false") .set("accord.shard_durability_target_splits", "1") .set("accord.shard_durability_cycle", "60s") diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/MigrationFromAccordWriteRaceTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/MigrationFromAccordWriteRaceTest.java index 027c863c21..1a8f30f765 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/MigrationFromAccordWriteRaceTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/MigrationFromAccordWriteRaceTest.java @@ -18,9 +18,6 @@ package org.apache.cassandra.distributed.test.accord; -import org.junit.Ignore; - -@Ignore public class MigrationFromAccordWriteRaceTest extends AccordMigrationWriteRaceTestBase { protected boolean migratingAwayFromAccord() diff --git a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java index 57c12b7684..c854a5da76 100644 --- a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java +++ b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java @@ -153,6 +153,7 @@ public class ConsistentBootstrapTest extends FuzzTestBase .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0")) .withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP) .set("write_request_timeout", "10s") + .set("accord.enabled", false) .set("metadata_snapshot_frequency", 5)) .start()) { diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java index 75fdd8fea8..d1653c6228 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java @@ -32,6 +32,7 @@ import accord.burn.SimulationException; import accord.impl.TopologyFactory; import accord.impl.basic.Cluster; import accord.impl.basic.RandomDelayQueue; +import accord.local.CommandStores; import accord.local.Node; import accord.utils.DefaultRandom; import accord.utils.RandomSource; @@ -55,7 +56,6 @@ import org.apache.cassandra.tools.FieldUtil; import static accord.impl.PrefixedIntHashKey.ranges; - public class AccordJournalBurnTest extends BurnTestBase { private static final Logger logger = LoggerFactory.getLogger(AccordJournalBurnTest.class); @@ -163,7 +163,14 @@ public class AccordJournalBurnTest extends BurnTestBase { return false; } - }, new AccordAgent(), directory, cfs); + }, new AccordAgent(), directory, cfs) + { + public void replay(CommandStores commandStores) + { + closeCurrentSegmentForTestingIfNonEmpty(); + super.replay(commandStores); + } + }; journal.start(null); journal.unsafeSetStarted(); diff --git a/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java b/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java index ea7256247f..c8bc94d145 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java +++ b/test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java @@ -61,6 +61,7 @@ import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.tcm.transformations.UnsafeJoin; import static org.apache.cassandra.simulator.Action.Modifiers.NO_TIMEOUTS; +import static org.apache.cassandra.simulator.Action.Modifiers.RELIABLE; import static org.apache.cassandra.simulator.Debug.EventType.CLUSTER; import static org.apache.cassandra.simulator.cluster.ClusterActions.TopologyChange.JOIN; import static org.apache.cassandra.simulator.cluster.ClusterActions.TopologyChange.LEAVE; @@ -216,7 +217,7 @@ public class ClusterActions extends SimulatedSystems return StrictAction.of("Initialise Cluster", () -> { List actions = new ArrayList<>(); - cluster.stream().forEach(i -> actions.add(invoke("Startup " + i.broadcastAddress(), NO_TIMEOUTS, NO_TIMEOUTS, + cluster.stream().forEach(i -> actions.add(invoke("Startup " + i.broadcastAddress(), RELIABLE, RELIABLE, new InterceptedRunnableExecution((InterceptingExecutor) i.executor(), i::startup)))); List endpoints = cluster.stream().map(IInstance::broadcastAddress).collect(Collectors.toList()); diff --git a/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulation.java index 299e702a49..fbc590c942 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulation.java @@ -39,6 +39,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.coordinate.CoordinationFailed; +import accord.coordinate.Invalidated; +import accord.coordinate.Preempted; import org.apache.cassandra.concurrent.ExecutorFactory; import org.apache.cassandra.concurrent.ScheduledExecutorPlus; import org.apache.cassandra.distributed.Cluster; @@ -90,6 +92,8 @@ public abstract class PaxosSimulation implements Simulation, ClusterActionListen protected Class[] expectedExceptionsAccord() { return (Class[]) new Class[] { RequestExecutionException.class, + Invalidated.class, + Preempted.class, CancellationException.class, CoordinationFailed.class, ClosedChannelException.class, diff --git a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java index ba6a9ebbc1..bbdaf617e8 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java @@ -429,7 +429,7 @@ public class AccordSyncPropagatorTest } @Override - protected void fetchTopologyInternal(long epoch) + public void fetchTopologyForEpoch(long epoch) { // TODO }