Follow-up to CASSANDRA-20906: Fix Simulator

modules/accord

@@ -1 +1 @@
-Subproject commit 78b84b08e13530722cb785a3b748fd2075c1c449
+Subproject commit 520818a004a89217cf86efa6c8fa2968401968ec

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<AccordTask<?>, Boolean>, SaveExecutor, Shutdownable, AsyncExecutor
+public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTask<?>, Boolean>, SaveExecutor, Shutdownable, AbstractAsyncExecutor
 {
     private static final Logger logger = LoggerFactory.getLogger(AccordExecutor.class);
     public interface AccordExecutorFactory

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.impl.AbstractAsyncExecutor;
-import accord.utils.async.AsyncChain;
-import accord.utils.async.AsyncChains;
 import org.apache.cassandra.service.accord.api.AccordAgent;
 
-public class ImmediateAsyncExecutor implements AsyncExecutor, BiConsumer<Object, Throwable>
+public class ImmediateAsyncExecutor implements AbstractAsyncExecutor
 {
     public static final ImmediateAsyncExecutor INSTANCE = new ImmediateAsyncExecutor();
 
-    @Override
-    public <T> AsyncChain<T> chain(Callable<T> 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<Object,
             AccordAgent.handleUncaughtException(t);
         }
     }
-
-    @Override
-    public void accept(Object o, Throwable throwable)
-    {
-        if (throwable != null)
-            AccordAgent.handleUncaughtException(throwable);
-    }
 }

src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java

@@ -258,7 +258,7 @@ public class AccordInteropExecution implements ReadCoordinator
                              }
 
                              Group group = Group.one(command);
-                             results.add(AsyncExecutor.chain(Stage.ACCORD_MIGRATION.executor(), () -> {
+                             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<Data> 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

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())

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<Data> 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(), () -> {
+        return AsyncChains.chain(Verb.READ_REPAIR_REQ.stage.executor(), () -> {
-                                          ReadRepairVerbHandler.instance.applyMutation(mutation);
+            ReadRepairVerbHandler.instance.applyMutation(mutation);
-                                          return Data.NOOP_DATA;
+            return Data.NOOP_DATA;
-                                      });
+        });
     }
 
     @Override

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<V> extends io.netty.util.concurrent.Future<V>, ListenableFuture<V>, Awaitable, AsyncResult<V>
 {
     /**

test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java

@@ -1344,6 +1344,11 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
             return;
 
         forEach.accept(cur);
+        {
+            Shared overrideShared = cur.getAnnotation(Shared.class);
+            if (overrideShared != null)
+                shared = new SharedParams(overrideShared);
+        }
 
         switch (shared.ancestors)
         {

test/simulator/main/org/apache/cassandra/simulator/systems/InterceptorOfGlobalMethods.java

@@ -43,7 +43,7 @@ import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES;
 import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
 
 @SuppressWarnings("unused")
-@Shared(scope = SIMULATION, inner = INTERFACES)
+@Shared(scope = SIMULATION, ancestors = INTERFACES)
 public interface InterceptorOfGlobalMethods extends InterceptorOfSystemMethods, Closeable
 {
     Semaphore newSemaphore(int count);

test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java

@@ -32,7 +32,6 @@ import com.google.common.collect.Maps;
 import org.junit.Before;
 import org.junit.BeforeClass;
 
-import accord.api.AsyncExecutor;
 import accord.api.RoutingKey;
 import accord.impl.SizeOfIntersectionSorter;
 import accord.local.Node;
@@ -57,6 +56,7 @@ import accord.topology.Topologies;
 import accord.utils.Gen;
 import accord.utils.Gens;
 import accord.utils.Invariants;
+import accord.utils.async.AsyncChains;
 import accord.utils.async.AsyncResult;
 import org.apache.cassandra.ServerTestUtils;
 import org.apache.cassandra.config.CassandraRelevantProperties;
@@ -286,7 +286,7 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester
             assertDeps(success.txnId, success.deps, cloneKeyConflicts, cloneRangeConflicts);
             return success;
         });
-        var delay = preAcceptAsync.flatMap(ignore -> 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());