Accord Query Tracing

Also improve:
 - Metric listeners

patch by Benedict; reviewed by David Capwell for CASSANDRA-20773

modules/accord

@@ -1 +1 @@
-Subproject commit e66a30cd944e902ad2749b5182e1f0d56903303f
+Subproject commit f40826db9571af286a689deada1613ffc872e5f8

src/java/org/apache/cassandra/concurrent/ExecutorLocals.java

@@ -92,8 +92,9 @@ public class ExecutorLocals implements WithResources, Closeable
     /**
      * Overwrite current locals, and return the previous ones
      */
-    public Closeable get()
+    public ExecutorLocals get()
     {
+        // TODO (desired): add compareAndSet to save one thread local round trip
         ExecutorLocals old = current();
         if (old != this)
             locals.set(this);

src/java/org/apache/cassandra/metrics/AccordMetrics.java

@@ -20,18 +20,31 @@ package org.apache.cassandra.metrics;
 
 import java.lang.reflect.Field;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
-import accord.api.EventListener;
+import javax.annotation.Nullable;
+
+import accord.api.CoordinatorEventListener;
+import accord.api.LocalEventListener;
+import accord.api.Result;
+import accord.coordinate.ExecutePath;
+import accord.impl.progresslog.DefaultProgressLog;
 import accord.local.Command;
+import accord.local.Node;
+import accord.local.SafeCommandStore;
+import accord.primitives.Ballot;
 import accord.primitives.Deps;
 import accord.primitives.PartialDeps;
 import accord.primitives.Timestamp;
 import accord.primitives.TxnId;
 import com.codahale.metrics.Counting;
+import com.codahale.metrics.Gauge;
 import com.codahale.metrics.Histogram;
 import com.codahale.metrics.Meter;
 import com.codahale.metrics.Timer;
+import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.service.accord.api.AccordTimeService;
+import org.apache.cassandra.tracing.Tracing;
 
 import static java.util.concurrent.TimeUnit.MICROSECONDS;
 import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
@@ -41,16 +54,21 @@ public class AccordMetrics
     public final static AccordMetrics readMetrics = new AccordMetrics("ro");
     public final static AccordMetrics writeMetrics = new AccordMetrics("rw");
 
-    public static final String STABLE_LATENCY = "StableLatency";
+    public static final String ACCORD_REPLICA = "AccordReplica";
-    public static final String PREAPPLY_LATENCY = "PreApplyLatency";
+    public static final String REPLICA_STABLE_LATENCY = "StableLatency";
-    public static final String APPLY_LATENCY = "ApplyLatency";
+    public static final String REPLICA_PREAPPLY_LATENCY = "PreApplyLatency";
-    public static final String APPLY_DURATION = "ApplyDuration";
+    public static final String REPLICA_APPLY_LATENCY = "ApplyLatency";
-    public static final String PARTIAL_DEPENDENCIES = "PartialDependencies";
+    public static final String REPLICA_APPLY_DURATION = "ApplyDuration";
+    public static final String REPLICA_DEPENDENCIES = "Dependencies";
     public static final String PROGRESS_LOG_SIZE = "ProgressLogSize";
 
-    public static final String DEPENDENCIES = "Dependencies";
+    public static final String ACCORD_COORDINATOR = "AccordCoordinator";
-    public static final String EPHEMERAL = "Ephemeral";
+    public static final String COORDINATOR_DEPENDENCIES = "Dependencies";
+    public static final String COORDINATOR_PREACCEPT_LATENCY = "PreAcceptLatency";
+    public static final String COORDINATOR_EXECUTE_LATENCY = "ExecuteLatency";
+    public static final String COORDINATOR_APPLY_LATENCY = "ApplyLatency";
     public static final String FAST_PATHS = "FastPaths";
+    public static final String MEDIUM_PATHS = "MediumPaths";
     public static final String SLOW_PATHS = "SlowPaths";
     public static final String PREEMPTS = "Preempts";
     public static final String TIMEOUTS = "Timeouts";
@@ -58,51 +76,66 @@ public class AccordMetrics
     public static final String RECOVERY_DELAY = "RecoveryDelay";
     public static final String RECOVERY_TIME = "RecoveryTime";
     public static final String FAST_PATH_TO_TOTAL = "FastPathToTotal";
-    public static final String ACCORD_REPLICA = "AccordReplica";
-    public static final String ACCORD_COORDINATOR = "AccordCoordinator";
 
     /**
      * The time between start on the coordinator and commit on this replica.
      */
-    public final Timer stableLatency;
+    public final Timer replicaStableLatency;
 
     /**
      * The time between start on the coordinator and arrival of the result on this replica.
      */
-    public final Timer preapplyLatency;
+    public final Timer replicaPreapplyLatency;
 
     /**
      * The time between start on the coordinator and application on this replica.
      */
-    public final Timer applyLatency;
+    public final Timer replicaApplyLatency;
 
     /**
      * TODO (expected): probably more interesting is latency from preapplied to apply;
      *  we already track local write latencies, whch this effectively duplicates (but including queueing latencies)
      * Duration of applying changes.
      */
-    public final Timer applyDuration;
+    public final Timer replicaApplyDuration;
 
     /**
      * A histogram of the number of dependencies per transaction at this replica.
      */
-    public final Histogram localDependencies;
+    public final Histogram replicaDependencies;
 
-    /**
+    public final Gauge<Long> progressLogSize;
-     * TODO (expected): this should be a Gauge
-     */
-    public final Meter progressLogSize;
 
     /**
      * A histogram of the number of dependencies per transaction at this coordinator.
      */
     public final Histogram coordinatorDependencies;
 
+    /**
+     * A histogram of the time to preaccept on this coordinator
+     */
+    public final Histogram coordinatorPreacceptLatency;
+
+    /**
+     * A histogram of the time to begin execution on this coordinator
+     */
+    public final Histogram coordinatorExecuteLatency;
+
+    /**
+     * A histogram of the time to complete execution on this coordinator
+     */
+    public final Histogram coordinatorApplyLatency;
+
     /**
      * The number of fast path transactions executed on this coordinator.
      */
     public final Meter fastPaths;
 
+    /**
+     * The number of medium path transactions executed on this coordinator.
+     */
+    public final Meter mediumPaths;
+
     /**
      * The number of slow path transactions executed on this coordinator.
      */
@@ -141,23 +174,33 @@ public class AccordMetrics
     private AccordMetrics(String scope)
     {
         DefaultNameFactory replica = new DefaultNameFactory(ACCORD_REPLICA, scope);
-        stableLatency = Metrics.timer(replica.createMetricName(STABLE_LATENCY));
+        replicaStableLatency = Metrics.timer(replica.createMetricName(REPLICA_STABLE_LATENCY));
-        preapplyLatency = Metrics.timer(replica.createMetricName(PREAPPLY_LATENCY));
+        replicaPreapplyLatency = Metrics.timer(replica.createMetricName(REPLICA_PREAPPLY_LATENCY));
-        applyLatency = Metrics.timer(replica.createMetricName(APPLY_LATENCY));
+        replicaApplyLatency = Metrics.timer(replica.createMetricName(REPLICA_APPLY_LATENCY));
-        applyDuration = Metrics.timer(replica.createMetricName(APPLY_DURATION));
+        replicaApplyDuration = Metrics.timer(replica.createMetricName(REPLICA_APPLY_DURATION));
-        localDependencies = Metrics.histogram(replica.createMetricName(PARTIAL_DEPENDENCIES), true);
+        replicaDependencies = Metrics.histogram(replica.createMetricName(REPLICA_DEPENDENCIES), true);
-        progressLogSize = Metrics.meter(replica.createMetricName(PROGRESS_LOG_SIZE));
+        progressLogSize = Metrics.gauge(replica.createMetricName(PROGRESS_LOG_SIZE), () -> {
+            AtomicLong i = new AtomicLong();
+            AccordService.instance().node().commandStores().forEachCommandStore(store -> {
+                i.addAndGet(((DefaultProgressLog)store.unsafeProgressLog()).size());
+            });
+            return i.get();
+        });
 
         DefaultNameFactory coordinator = new DefaultNameFactory(ACCORD_COORDINATOR, scope);
-        coordinatorDependencies = Metrics.histogram(coordinator.createMetricName(DEPENDENCIES), true);
+        coordinatorDependencies = Metrics.histogram(coordinator.createMetricName(COORDINATOR_DEPENDENCIES), true);
+        coordinatorPreacceptLatency = Metrics.histogram(coordinator.createMetricName(COORDINATOR_PREACCEPT_LATENCY), true);
+        coordinatorExecuteLatency = Metrics.histogram(coordinator.createMetricName(COORDINATOR_EXECUTE_LATENCY), true);
+        coordinatorApplyLatency = Metrics.histogram(coordinator.createMetricName(COORDINATOR_APPLY_LATENCY), true);
         fastPaths = Metrics.meter(coordinator.createMetricName(FAST_PATHS));
+        mediumPaths = Metrics.meter(coordinator.createMetricName(MEDIUM_PATHS));
         slowPaths = Metrics.meter(coordinator.createMetricName(SLOW_PATHS));
         preempts = Metrics.meter(coordinator.createMetricName(PREEMPTS));
         timeouts = Metrics.meter(coordinator.createMetricName(TIMEOUTS));
         invalidations = Metrics.meter(coordinator.createMetricName(INVALIDATIONS));
         recoveryDelay = Metrics.timer(coordinator.createMetricName(RECOVERY_DELAY));
         recoveryDuration = Metrics.timer(coordinator.createMetricName(RECOVERY_TIME));
-        fastPathToTotal = new RatioGaugeSet(fastPaths, RatioGaugeSet.sum(fastPaths, slowPaths), coordinator, FAST_PATH_TO_TOTAL + ".%s");
+        fastPathToTotal = new RatioGaugeSet(fastPaths, RatioGaugeSet.sum(fastPaths, mediumPaths, slowPaths), coordinator, FAST_PATH_TO_TOTAL + ".%s");
     }
 
     @Override
@@ -186,7 +229,7 @@ public class AccordMetrics
         return builder.toString();
     }
 
-    public static class Listener implements EventListener
+    public static class Listener implements CoordinatorEventListener, LocalEventListener
     {
         public final static Listener instance = new Listener(AccordMetrics.readMetrics, AccordMetrics.writeMetrics);
 
@@ -212,112 +255,112 @@ public class AccordMetrics
         }
 
         @Override
-        public void onStable(Command cmd)
+        public void onStable(SafeCommandStore safeStore, Command cmd)
         {
+            Tracing.trace("Stable {} on {}", cmd.txnId(), safeStore.commandStore());
             long now = AccordTimeService.nowMicros();
             AccordMetrics metrics = forTransaction(cmd.txnId());
             if (metrics != null)
             {
                 long trxTimestamp = cmd.txnId().hlc();
-                metrics.stableLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS);
+                metrics.replicaStableLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS);
             }
         }
 
         @Override
-        public void onPreApplied(Command cmd)
+        public void onPreApplied(SafeCommandStore safeStore, Command cmd)
         {
+            Tracing.trace("Preapplied {} on {}", cmd.txnId(), safeStore.commandStore());
             long now = AccordTimeService.nowMicros();
             AccordMetrics metrics = forTransaction(cmd.txnId());
             if (metrics != null)
             {
                 Timestamp trxTimestamp = cmd.txnId();
-                metrics.preapplyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
+                metrics.replicaPreapplyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
                 PartialDeps deps = cmd.partialDeps();
-                metrics.localDependencies.update(deps != null ? deps.txnIdCount() : 0);
+                metrics.replicaDependencies.update(deps != null ? deps.txnIdCount() : 0);
             }
         }
 
         @Override
-        public void onApplied(Command cmd, long applyStartTimestamp)
+        public void onApplied(SafeCommandStore safeStore, Command cmd, long applyStartedAt)
         {
+            Tracing.trace("Applied {} on {}", cmd.txnId(), safeStore.commandStore());
             long now = AccordTimeService.nowMicros();
             AccordMetrics metrics = forTransaction(cmd.txnId());
             if (metrics != null)
             {
                 Timestamp trxTimestamp = cmd.txnId();
-                metrics.applyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
+                metrics.replicaApplyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS);
-                if (applyStartTimestamp > 0)
+                if (applyStartedAt > 0)
-                    metrics.applyDuration.update(now - applyStartTimestamp, TimeUnit.MICROSECONDS);
+                    metrics.replicaApplyDuration.update(now - applyStartedAt, TimeUnit.MICROSECONDS);
             }
         }
 
         @Override
-        public void onFastPathTaken(TxnId txnId, Deps deps)
+        public void onPreAccepted(TxnId txnId)
         {
             AccordMetrics metrics = forTransaction(txnId);
             if (metrics != null)
             {
-                metrics.fastPaths.mark();
+                long now = AccordTimeService.nowMicros();
-                metrics.coordinatorDependencies.update(deps.txnIdCount());
+                metrics.coordinatorPreacceptLatency.update(Math.max(0, now - txnId.hlc()));
             }
         }
 
         @Override
-        public void onSlowPathTaken(TxnId txnId, Deps deps)
+        public void onExecuting(TxnId txnId, @Nullable Ballot ballot, Deps deps, @Nullable ExecutePath path)
+        {
+            Tracing.trace("{} agreed {}", path, txnId);
+            AccordMetrics metrics = forTransaction(txnId);
+            if (metrics != null)
+            {
+                metrics.coordinatorDependencies.update(deps.txnIdCount());
+                long now = AccordTimeService.nowMicros();
+                metrics.coordinatorExecuteLatency.update(Math.max(0, now - txnId.hlc()));
+                if (path != null)
+                {
+                    switch (path)
+                    {
+                        case FAST: metrics.fastPaths.mark(); break;
+                        case MEDIUM: metrics.mediumPaths.mark(); break;
+                        case SLOW: metrics.slowPaths.mark(); break;
+                    }
+                }
+            }
+        }
+
+        @Override
+        public void onExecuted(TxnId txnId, Ballot ballot)
         {
             AccordMetrics metrics = forTransaction(txnId);
             if (metrics != null)
             {
-                metrics.slowPaths.mark();
+                long now = AccordTimeService.nowMicros();
-                metrics.coordinatorDependencies.update(deps.txnIdCount());
+                metrics.coordinatorApplyLatency.update(Math.max(0, now - txnId.hlc()));
             }
         }
 
         @Override
-        public void onRecover(TxnId txnId, Timestamp recoveryTimestamp)
+        public void onRecoveryStopped(Node node, TxnId txnId, Ballot ballot, Result result, Throwable failure)
         {
             AccordMetrics metrics = forTransaction(txnId);
             if (metrics != null)
             {
                 long now = AccordTimeService.nowMicros();
 
-                metrics.recoveryDuration.update(now - recoveryTimestamp.hlc(), MICROSECONDS);
+                metrics.recoveryDuration.update(Math.max(0, now - ballot.hlc()), MICROSECONDS);
-                metrics.recoveryDelay.update(recoveryTimestamp.hlc() - txnId.hlc(), MICROSECONDS);
+                metrics.recoveryDelay.update(Math.max(0, ballot.hlc() - txnId.hlc()), MICROSECONDS);
             }
         }
 
-        @Override
-        public void onPreempted(TxnId txnId)
-        {
-            AccordMetrics metrics = forTransaction(txnId);
-            if (metrics != null)
-                metrics.preempts.mark();
-        }
-
-        @Override
-        public void onTimeout(TxnId txnId)
-        {
-            // TODO (required): we appear to be marking this twice, once in AccordResult and once here.
-            //   why does AccordMetricsTest only see this one? remove duplication.
-            AccordMetrics metrics = forTransaction(txnId);
-            if (metrics != null)
-                metrics.timeouts.mark();
-        }
-
         @Override
         public void onInvalidated(TxnId txnId)
         {
+            Tracing.trace("Invalidated {}", txnId);
             AccordMetrics metrics = forTransaction(txnId);
             if (metrics != null)
                 metrics.invalidations.mark();
         }
-
-        @Override
-        public void onProgressLogSizeChange(TxnId txnId, int delta)
-        {
-            AccordMetrics metrics = forTransaction(txnId);
-            if (metrics != null)
-                metrics.progressLogSize.mark(delta);
-        }
     }
 }

src/java/org/apache/cassandra/service/accord/AccordCommandStore.java

@@ -70,7 +70,10 @@ import accord.utils.async.AsyncChains;
 import accord.utils.async.AsyncResults.CountingResult;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.TableMetadataRef;
 import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor;
 import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfo;
 import org.apache.cassandra.service.accord.api.TokenKey;
@@ -159,6 +162,7 @@ public class AccordCommandStore extends CommandStore
     private long lastSystemTimestampMicros = Long.MIN_VALUE;
     private final CommandsForRanges.Manager commandsForRanges;
     private final TableId tableId;
+    private TableMetadataRef metadata;
     volatile SafeRedundantBefore safeRedundantBefore;
 
     private AccordSafeCommandStore current;
@@ -502,7 +506,7 @@ public class AccordCommandStore extends CommandStore
     protected void ensureDurable(Ranges ranges, RedundantBefore onCommandStoreDurable)
     {
         long reportId = nextDurabilityLoggingId.incrementAndGet();
-        logger.info("{} awaiting local metadata durability for {} ({})", this, ranges, reportId);
+        logger.debug("{} awaiting local metadata durability for {} ({})", this, ranges, reportId);
         executor().afterSubmittedAndConsequences(() -> {
             logger.debug("{}: saving intersecting keys ({})", this, reportId);
             class Ready extends CountingResult implements Runnable
@@ -640,4 +644,29 @@ public class AccordCommandStore extends CommandStore
             return a.ticket >= b.ticket ? a : b;
         }
     }
+
+    private @Nullable TableMetadata tableMetadata()
+    {
+        TableMetadataRef metadataRef = this.metadata;
+        if (metadataRef != null)
+            return metadataRef.get();
+
+        TableMetadata metadata = Schema.instance.getTableMetadata(tableId);
+        if (metadata == null)
+            return null;
+        this.metadata = metadata.ref;
+        return metadata;
+    }
+
+    @Override
+    public String toString()
+    {
+        TableMetadata metadata = tableMetadata();
+        StringBuilder sb = new StringBuilder("[");
+        if (metadata != null)
+            sb.append(metadata).append('|');
+        sb.append(tableId.lsb());
+        sb.append(',').append(id).append(',').append(node.id().id).append(']');
+        return sb.toString();
+    }
 }

src/java/org/apache/cassandra/service/accord/AccordExecutor.java

@@ -69,7 +69,10 @@ import org.apache.cassandra.metrics.AccordCacheMetrics;
 import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
 import org.apache.cassandra.service.accord.AccordCacheEntry.SaveExecutor;
 import org.apache.cassandra.service.accord.AccordCacheEntry.UniqueSave;
+import org.apache.cassandra.concurrent.ExecutorLocals;
+import org.apache.cassandra.utils.Closeable;
 import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.WithResources;
 import org.apache.cassandra.utils.concurrent.AsyncPromise;
 import org.apache.cassandra.utils.concurrent.Condition;
 import org.apache.cassandra.utils.concurrent.Future;
@@ -900,6 +903,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
 
     static abstract class SubmittableTask extends Task
     {
+        final WithResources locals = ExecutorLocals.propagate();
         abstract void submitExclusive(AccordExecutor owner);
     }
 
@@ -1261,7 +1265,10 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
         @Override
         protected void runInternal()
         {
-            run.run();
+            try (Closeable close = locals.get())
+            {
+                run.run();
+            }
             if (result != null)
                 result.trySuccess(null);
         }
@@ -1358,7 +1365,10 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
         @Override
         public void runInternal()
         {
-            result = entry.owner.parent().adapter().load(entry.owner.commandStore, entry.key());
+            try (Closeable close = locals.get())
+            {
+                result = entry.owner.parent().adapter().load(entry.owner.commandStore, entry.key());
+            }
         }
 
         @Override
@@ -1400,7 +1410,10 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
         @Override
         protected void runInternal()
         {
-            wrapped.runInternal();
+            try (Closeable close = locals.get())
+            {
+                wrapped.runInternal();
+            }
         }
 
         @Override
@@ -1446,7 +1459,10 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
         @Override
         public void runInternal()
         {
-            run.run();
+            try (Closeable close = locals.get())
+            {
+                run.run();
+            }
             failure = null;
         }
 
@@ -1486,7 +1502,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
         protected void runInternal()
         {
             T success;
-            try
+            try (Closeable close = locals.get())
             {
                 success = call.call();
             }

src/java/org/apache/cassandra/service/accord/AccordTask.java

@@ -65,7 +65,9 @@ import org.apache.cassandra.service.accord.AccordExecutor.SubmittableTask;
 import org.apache.cassandra.service.accord.AccordExecutor.TaskQueue;
 import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor;
 import org.apache.cassandra.service.accord.api.TokenKey;
+import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.Clock;
+import org.apache.cassandra.utils.Closeable;
 import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.concurrent.Condition;
 
@@ -652,8 +654,11 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
     {
         logger.trace("Running {} with state {}", this, state);
         AccordSafeCommandStore safeStore = null;
-        try
+        try (Closeable close = locals.get())
         {
+            if (Tracing.isTracing())
+                Tracing.trace(preLoadContext.describe());
+
             if (state != RUNNING)
                 throw illegalState("Unexpected state " + toDescription());
 

src/java/org/apache/cassandra/service/accord/api/AccordAgent.java

@@ -29,9 +29,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import accord.api.Agent;
-import accord.api.EventListener;
+import accord.api.CoordinatorEventListener;
+import accord.api.LocalEventListener;
 import accord.api.ProgressLog.BlockedUntil;
-import accord.api.Result;
 import accord.api.RoutingKey;
 import accord.api.Tracing;
 import accord.api.TraceEventType;
@@ -127,11 +127,6 @@ public class AccordAgent implements Agent
         self = id;
     }
 
-    @Override
-    public void onRecover(Node node, Result success, Throwable fail)
-    {
-    }
-
     @Override
     public void onInconsistentTimestamp(Command command, Timestamp prev, Timestamp next)
     {
@@ -227,7 +222,13 @@ public class AccordAgent implements Agent
     }
 
     @Override
-    public EventListener eventListener()
+    public CoordinatorEventListener coordinatorEvents()
+    {
+        return AccordMetrics.Listener.instance;
+    }
+
+    @Override
+    public LocalEventListener localEvents()
     {
         return AccordMetrics.Listener.instance;
     }

src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java

@@ -143,7 +143,7 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
             if (!preserveTimestamps)
                 update = new PartitionUpdate.Builder(update, 0).updateAllTimestamp(timestamp).build();
             Mutation mutation = new Mutation(update, PotentialTxnConflicts.ALLOW);
-            return AsyncChains.ofRunnable(executor, mutation::applyUnsafe);
+            return AsyncChains.ofRunnable(executor, () -> mutation.apply(false, false));
         }
 
         @Override

test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java

@@ -266,7 +266,7 @@ public class AccordMetricsTest extends AccordTestBase
         assertThat(metric.apply(AccordMetrics.TIMEOUTS)).isEqualTo(timeouts);
         assertThat(metric.apply(AccordMetrics.RECOVERY_DELAY)).isEqualTo(recoveries);
         assertThat(metric.apply(AccordMetrics.RECOVERY_TIME)).isEqualTo(recoveries);
-        assertThat(metric.apply(AccordMetrics.DEPENDENCIES)).isEqualTo(fastPaths + slowPaths);
+        assertThat(metric.apply(AccordMetrics.COORDINATOR_DEPENDENCIES)).isEqualTo(fastPaths + slowPaths);
 
         // Verify that coordinator metrics are published to the appropriate virtual table:
         SimpleQueryResult res = SHARED_CLUSTER.get(node + 1)
@@ -290,11 +290,11 @@ public class AccordMetricsTest extends AccordTestBase
         DefaultNameFactory nameFactory = new DefaultNameFactory(AccordMetrics.ACCORD_REPLICA, scope);
         Map<String, Long> metrics = diff(countingMetrics0).get(node);
         Function<String, Long> metric = n -> metrics.get(nameFactory.createMetricName(n).getMetricName());
-        assertThat(metric.apply(AccordMetrics.STABLE_LATENCY)).isLessThanOrEqualTo(stable);
+        assertThat(metric.apply(AccordMetrics.REPLICA_STABLE_LATENCY)).isLessThanOrEqualTo(stable);
-        assertThat(metric.apply(AccordMetrics.PREAPPLY_LATENCY)).isEqualTo(executions);
+        assertThat(metric.apply(AccordMetrics.REPLICA_PREAPPLY_LATENCY)).isEqualTo(executions);
-        assertThat(metric.apply(AccordMetrics.APPLY_LATENCY)).isEqualTo(applications);
+        assertThat(metric.apply(AccordMetrics.REPLICA_APPLY_LATENCY)).isEqualTo(applications);
-        assertThat(metric.apply(AccordMetrics.APPLY_DURATION)).isEqualTo(scope.equals("rw") ? applications : 0);
+        assertThat(metric.apply(AccordMetrics.REPLICA_APPLY_DURATION)).isEqualTo(scope.equals("rw") ? applications : 0);
-        assertThat(metric.apply(AccordMetrics.PARTIAL_DEPENDENCIES)).isEqualTo(executions);
+        assertThat(metric.apply(AccordMetrics.REPLICA_DEPENDENCIES)).isEqualTo(executions);
 
         // Verify that replica metrics are published to the appropriate virtual table:
         SimpleQueryResult vtableResults = SHARED_CLUSTER.get(node + 1)

test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java

@@ -51,7 +51,6 @@ import accord.api.DataStore;
 import accord.api.Journal;
 import accord.api.Key;
 import accord.api.ProgressLog;
-import accord.api.Result;
 import accord.api.RoutingKey;
 import accord.api.Timeouts;
 import accord.impl.AbstractSafeCommandStore;
@@ -659,7 +658,6 @@ public class CommandsForKeySerializerTest
         @Override public void shutdown() { }
         @Override protected void registerTransitive(SafeCommandStore safeStore, RangeDeps deps){ }
         @Override public <T> AsyncChain<T> build(Callable<T> task) { throw new UnsupportedOperationException(); }
-        @Override public void onRecover(Node node, Result success, Throwable fail) { throw new UnsupportedOperationException(); }
         @Override public void onInconsistentTimestamp(Command command, Timestamp prev, Timestamp next) { throw new UnsupportedOperationException(); }
         @Override public void onFailedBootstrap(int attempts, String phase, Ranges ranges, Runnable retry, Throwable failure) { throw new UnsupportedOperationException(); }
         @Override public void onStale(Timestamp staleSince, Ranges ranges) { throw new UnsupportedOperationException(); }