From 12000406c34748b6ce21d97bbcf5020c313c5672 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Thu, 18 Sep 2025 13:49:28 +0100 Subject: [PATCH] Follow-up to CASSANDRA-20906: Fix Simulator --- modules/accord | 2 +- .../service/accord/AccordExecutor.java | 4 +-- .../accord/ImmediateAsyncExecutor.java | 30 ++----------------- .../interop/AccordInteropExecution.java | 6 ++-- .../accord/interop/AccordInteropRead.java | 4 +-- .../interop/AccordInteropReadRepair.java | 9 +++--- .../cassandra/utils/concurrent/Future.java | 2 +- .../distributed/impl/AbstractCluster.java | 5 ++++ .../systems/InterceptorOfGlobalMethods.java | 2 +- .../SimulatedAccordCommandStoreTestBase.java | 4 +-- 10 files changed, 24 insertions(+), 44 deletions(-) diff --git a/modules/accord b/modules/accord index 78b84b08e1..520818a004 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 78b84b08e13530722cb785a3b748fd2075c1c449 +Subproject commit 520818a004a89217cf86efa6c8fa2968401968ec diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index 3f91882051..637b158177 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -40,8 +40,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.api.Agent; -import accord.api.AsyncExecutor; import accord.api.RoutingKey; +import accord.impl.AbstractAsyncExecutor; import accord.local.Command; import accord.local.PreLoadContext; import accord.local.SequentialAsyncExecutor; @@ -95,7 +95,7 @@ import static org.apache.cassandra.service.accord.AccordTask.State.WAITING_TO_RU * NOTE: We assume that NO BLOCKING TASKS are submitted to this executor AND WAITED ON by another task executing on this executor. * (as we do not immediately schedule additional threads for submitted tasks, but schedule new threads only if necessary when the submitting execution completes) */ -public abstract class AccordExecutor implements CacheSize, LoadExecutor, Boolean>, SaveExecutor, Shutdownable, AsyncExecutor +public abstract class AccordExecutor implements CacheSize, LoadExecutor, Boolean>, SaveExecutor, Shutdownable, AbstractAsyncExecutor { private static final Logger logger = LoggerFactory.getLogger(AccordExecutor.class); public interface AccordExecutorFactory diff --git a/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java b/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java index 7626ee9b97..0f1e82a0d3 100644 --- a/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/ImmediateAsyncExecutor.java @@ -18,34 +18,15 @@ package org.apache.cassandra.service.accord; -import java.util.concurrent.Callable; -import java.util.function.BiConsumer; - import javax.annotation.Nonnull; -import accord.api.AsyncExecutor; -import accord.utils.async.AsyncChain; -import accord.utils.async.AsyncChains; +import accord.impl.AbstractAsyncExecutor; import org.apache.cassandra.service.accord.api.AccordAgent; -public class ImmediateAsyncExecutor implements AsyncExecutor, BiConsumer +public class ImmediateAsyncExecutor implements AbstractAsyncExecutor { public static final ImmediateAsyncExecutor INSTANCE = new ImmediateAsyncExecutor(); - @Override - public AsyncChain chain(Callable call) - { - try - { - return AsyncChains.success(call.call()); - } - catch (Throwable t) - { - AccordAgent.handleUncaughtException(t); - return AsyncChains.failure(t); - } - } - @Override public void execute(@Nonnull Runnable command) { @@ -58,11 +39,4 @@ public class ImmediateAsyncExecutor implements AsyncExecutor, BiConsumer { + results.add(AsyncChains.chain(Stage.ACCORD_MIGRATION.executor(), () -> { TxnData result = new TxnData(); // Enforcing limits is redundant since we only have a group of size 1, but checking anyways // documents the requirement here @@ -294,7 +294,7 @@ public class AccordInteropExecution implements ReadCoordinator // TODO (required): To make migration work we need to validate that the range is all on Accord - results.add(AsyncExecutor.chain(Stage.ACCORD_MIGRATION.executor(), () -> { + results.add(AsyncChains.chain(Stage.ACCORD_MIGRATION.executor(), () -> { TxnData result = new TxnData(); try (PartitionIterator iterator = StorageProxy.getRangeSlice(command, consistencyLevel, this, requestTime)) { @@ -396,7 +396,7 @@ public class AccordInteropExecution implements ReadCoordinator private AsyncChain executeUnrecoverableRepairUpdate() { - return AsyncExecutor.chain(Stage.ACCORD_MIGRATION.executor(), () -> { + return AsyncChains.chain(Stage.ACCORD_MIGRATION.executor(), () -> { UnrecoverableRepairUpdate repairUpdate = (UnrecoverableRepairUpdate)txn.update(); // TODO (expected): We should send the read in the same message as the commit. This requires refactor ReadData.Kind so that it doesn't specify the ordinal encoding // and can be extended similar to MessageType which allows additional types not from Accord to be added diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java index 51f9360829..cb53c27a31 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropRead.java @@ -283,7 +283,7 @@ public class AccordInteropRead extends ReadData return AsyncChains.success(new LocalReadData(new ArrayList<>(), readCommand)); ReadCommand submit = readCommand.withTransactionalSettings(TxnNamedRead.readsWithoutReconciliation(txnRead.cassandraConsistencyLevel()), nowInSeconds); - return AsyncExecutor.chain(Stage.READ.executor(), () -> new LocalReadData(key, ReadCommandVerbHandler.instance.doRead(submit, false), command)); + return AsyncChains.chain(Stage.READ.executor(), () -> new LocalReadData(key, ReadCommandVerbHandler.instance.doRead(submit, false), command)); } // This path can have a subrange we have never seen before provided by short read protection or read repair so we need to @@ -298,7 +298,7 @@ public class AccordInteropRead extends ReadData continue; ReadCommand submit = TxnNamedRead.commandForSubrange((PartitionRangeReadCommand) command, intersection, txnRead.cassandraConsistencyLevel(), nowInSeconds); TokenKey routingKey = ((TokenRange)r).start(); - chains.add(AsyncExecutor.chain(Stage.READ.executor(), () -> new LocalReadData(routingKey, ReadCommandVerbHandler.instance.doRead(submit, false), command))); + chains.add(AsyncChains.chain(Stage.READ.executor(), () -> new LocalReadData(routingKey, ReadCommandVerbHandler.instance.doRead(submit, false), command))); } if (chains.isEmpty()) diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java index d080386ba0..1a661215cc 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java @@ -35,6 +35,7 @@ import accord.primitives.Timestamp; import accord.primitives.TxnId; import accord.topology.Topologies; import accord.utils.async.AsyncChain; +import accord.utils.async.AsyncChains; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.ReadRepairVerbHandler; import org.apache.cassandra.db.TypeSizes; @@ -145,10 +146,10 @@ public class AccordInteropReadRepair extends ReadData protected AsyncChain beginRead(SafeCommandStore safeStore, Timestamp executeAt, PartialTxn txn, Participants execute) { // TODO (required): subtract unavailable ranges, either from read or from response (or on coordinator) - return AsyncExecutor.chain(Verb.READ_REPAIR_REQ.stage.executor(), () -> { - ReadRepairVerbHandler.instance.applyMutation(mutation); - return Data.NOOP_DATA; - }); + return AsyncChains.chain(Verb.READ_REPAIR_REQ.stage.executor(), () -> { + ReadRepairVerbHandler.instance.applyMutation(mutation); + return Data.NOOP_DATA; + }); } @Override diff --git a/src/java/org/apache/cassandra/utils/concurrent/Future.java b/src/java/org/apache/cassandra/utils/concurrent/Future.java index b8bf62ecb5..629c2f1093 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Future.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Future.java @@ -42,7 +42,7 @@ import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; * A Future that integrates several different (but equivalent) APIs used within Cassandra into a single concept, * integrating also with our {@link Awaitable} abstraction, to overall improve coherency and clarity in the codebase. */ -@Shared(scope = SIMULATION, ancestors = INTERFACES) +@Shared(scope = SIMULATION, ancestors = INTERFACES, members = INTERFACES) public interface Future extends io.netty.util.concurrent.Future, ListenableFuture, Awaitable, AsyncResult { /** diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index 6cd9c8e60a..805d15665a 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -1344,6 +1344,11 @@ public abstract class AbstractCluster implements ICluster AsyncExecutor.chain(instance.unorderedScheduled, () -> { + var delay = preAcceptAsync.flatMap(ignore -> AsyncChains.chain(instance.unorderedScheduled, () -> { Ballot ballot = Ballot.fromValues(instance.storeService.epoch(), instance.storeService.now(), nodeId); return new BeginRecovery(nodeId, new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, instance.topology), txnId, null, false, txn, route, ballot); }).beginAsResult());