Topology compaction omits image records if minEpoch is ahead of last image

Also Fix:
 - RegisteredCallback should remove itself from callback map when cancelled
 - Do not throw CancellationException when processing requests that have been aborted, as may be caused by a successful meaningful reply that can be overridden
 - system_accord_debug.{executors,coordinations} fail with ClassCastException
 - CommandStore.updateMinHlc eats up CPU as called much too often
 - AccordCache not notifying flushed on shutdown
Also Improve:
 - Support skipping Deps
 - Violation information reported
 - Sort CommandStore shards by id

patch by Benedict; reviewed by Aleksey Yeschenko for CASSANDRA-20896

Author: belliottsmith

modules/accord

@@ -95,6 +95,7 @@
 import org.apache.cassandra.service.accord.JournalKey;
 import org.apache.cassandra.service.accord.api.TokenKey;
 import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate;
+import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.TopologyImage;
 import org.apache.cassandra.service.accord.serializers.Version;
 import org.apache.cassandra.service.paxos.PaxosRepairHistory;
 import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
@@ -118,6 +119,8 @@
 import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
 import static org.apache.cassandra.service.accord.AccordKeyspace.CFKAccessor;
 import static org.apache.cassandra.service.accord.AccordKeyspace.JournalColumns.getJournalKey;
+import static org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.Kind.Image;
+import static org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.Kind.Repeat;
 
 /**
  * Merge multiple iterators over the content of sstable into a "compacted" iterator.
@@ -868,7 +871,6 @@ class AccordJournalPurger extends AbstractPurger
         public AccordJournalPurger(AccordCompactionInfos compactionInfos, Version version, ColumnFamilyStore cfs)
         {
             this.userVersion = version;
-
             this.infos = compactionInfos;
             this.recordColumn = cfs.metadata().getColumn(ColumnIdentifier.getInterned("record", false));
             this.versionColumn = cfs.metadata().getColumn(ColumnIdentifier.getInterned("user_version", false));
@@ -943,8 +945,10 @@ static abstract class AccordRowCompactor<T extends FlyweightImage>
 
     static class TopologyCompactor extends AccordMergingCompactor<AccordTopologyUpdate.Accumulator>
     {
-        AccordTopologyUpdate.TopologyImage lastChangedTopology;
+        TopologyImage lastImage;
+        boolean hasWritten;
         final long minEpoch;
+
         TopologyCompactor(FlyweightSerializer<Object, AccordTopologyUpdate.Accumulator> serializer, Version userVersion, long minEpoch)
         {
             super(serializer, userVersion);
@@ -960,21 +964,33 @@ void reset(JournalKey key, UnfilteredRowIterator partition)
         @Override
         UnfilteredRowIterator result(JournalKey journalKey, DecoratedKey partitionKey) throws IOException
         {
-            AccordTopologyUpdate.TopologyImage current = builder.get();
+            Invariants.require(lastImage != null || !hasWritten);
+            TopologyImage read = builder.read();
 
-            if (lastChangedTopology != null && current.getUpdate() != null && lastChangedTopology.getUpdate().isEquivalent(current.getUpdate()))
-                builder.update(current.asNoOp());
+            if (read.epoch() < minEpoch)
+            {
+                if (read.kind() == Image)
+                    lastImage = read;
+                return null;
+            }
 
-            if (builder.get().kind() != AccordTopologyUpdate.Kind.NoOp)
+            TopologyImage write = read;
+            if (read.kind() == Repeat && !hasWritten)
             {
-                lastChangedTopology = builder.get();
-                Invariants.nonNull(lastChangedTopology.getUpdate());
+                Invariants.require(lastImage != null);
+                write = new TopologyImage(read.epoch(), Image, lastImage.getUpdate());
+            }
+            else if (hasWritten && read.kind() == Repeat && lastImage.getUpdate().isEquivalent(read.getUpdate()))
+            {
+                write = read.asRepeat();
             }
 
-            if (builder.get().epoch() >= minEpoch)
-                return super.result(journalKey, partitionKey);
-            else
-                return null;
+            if (write.kind() == Image)
+                lastImage = write;
+
+            hasWritten = true;
+            builder.write(write);
+            return super.result(journalKey, partitionKey);
         }
     }
 

src/java/org/apache/cassandra/db/compaction/CompactionIterator.java

@@ -219,8 +219,8 @@ private ExecutorsTable()
                         "  txn_id 'TxnIdUtf8Type',\n" +
                         "  txn_id_additional 'TxnIdUtf8Type',\n" +
                         "  keys text,\n" +
-                        "  keysLoad text,\n" +
-                        "  keysLoadFor text,\n" +
+                        "  keys_loading text,\n" +
+                        "  keys_loading_for text,\n" +
                         "  PRIMARY KEY (executor_id, status, position, unique_position)" +
                         ')', UTF8Type.instance));
         }
@@ -242,14 +242,14 @@ public DataSet data()
                     else uniquePos = 0;
                     prev = info;
                     PreLoadContext preLoadContext = info.preLoadContext();
-                    ds.row(executorId, info.status(), info.position(), uniquePos)
+                    ds.row(executorId, Objects.toString(info.status()), info.position(), uniquePos)
                       .column("description", info.describe())
                       .column("command_store_id", info.commandStoreId())
-                      .column("txn_id", preLoadContext == null ? null : preLoadContext.primaryTxnId())
-                      .column("txn_id_additional", preLoadContext == null ? null : preLoadContext.additionalTxnId())
-                      .column("keys", preLoadContext == null ? null : preLoadContext.keys())
-                      .column("keysLoad", preLoadContext == null ? null : preLoadContext.loadKeys())
-                      .column("keysLoadFor", preLoadContext == null ? null : preLoadContext.loadKeysFor())
+                      .column("txn_id", preLoadContext == null ? null : toStringOrNull(preLoadContext.primaryTxnId()))
+                      .column("txn_id_additional", preLoadContext == null ? null : toStringOrNull(preLoadContext.additionalTxnId()))
+                      .column("keys", preLoadContext == null ? null : toStringOrNull(preLoadContext.keys()))
+                      .column("keys_loading", preLoadContext == null ? null : toStringOrNull(preLoadContext.loadKeys()))
+                      .column("keys_loading_for", preLoadContext == null ? null : toStringOrNull(preLoadContext.loadKeysFor()))
                     ;
                 }
             }
@@ -265,9 +265,9 @@ private CoordinationsTable()
             super(parse(VIRTUAL_ACCORD_DEBUG, COORDINATIONS,
                         "Accord Coordination State",
                         "CREATE TABLE %s (\n" +
-                        "  txn_id int,\n" +
+                        "  txn_id 'TxnIdUtf8Type',\n" +
                         "  kind text,\n" +
-                        "  coordination_id int,\n" +
+                        "  coordination_id bigint,\n" +
                         "  description text,\n" +
                         "  nodes text,\n" +
                         "  nodes_inflight text,\n" +
@@ -286,7 +286,7 @@ public DataSet data()
             SimpleDataSet ds = new SimpleDataSet(metadata());
             for (Coordination c : coordinations)
             {
-                ds.row(c.txnId(), c.kind().toString(), c.coordinationId())
+                ds.row(toStringOrNull(c.txnId()), c.kind().toString(), c.coordinationId())
                   .column("nodes", toStringOrNull(c.nodes()))
                   .column("nodes_inflight", toStringOrNull(c.inflight()))
                   .column("nodes_contacted", toStringOrNull(c.contacted()))

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -335,6 +335,21 @@ void onSuccess(Runnable onSuccess)
                 this.onSuccess = new ArrayList<>();
             this.onSuccess.add(onSuccess);
         }
+
+        static void notify(List<Runnable> onSuccess)
+        {
+            if (onSuccess != null)
+            {
+                onSuccess.forEach(run -> {
+                    try { run.run(); }
+                    catch (Throwable t)
+                    {
+                        Thread thread = Thread.currentThread();
+                        thread.getUncaughtExceptionHandler().uncaughtException(thread, t);
+                    }
+                });
+            }
+        }
     }
 
     public interface SaveExecutor
@@ -506,8 +521,7 @@ boolean save()
             setStatus(LOADED);
             if (waitingToSave != null)
                 this.state = state;
-            if (identity.onSuccess != null)
-                identity.onSuccess.forEach(Runnable::run);
+            UniqueSave.notify(identity.onSuccess);
             return false;
         }
         else
@@ -521,6 +535,9 @@ boolean save()
 
     boolean saved(Object identity, Throwable fail)
     {
+        if (identity instanceof UniqueSave)
+            UniqueSave.notify(((UniqueSave) identity).onSuccess);
+
         if (!is(SAVING))
             return false;
 

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

@@ -74,7 +74,6 @@
 import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfo;
 import org.apache.cassandra.service.accord.api.TokenKey;
 import org.apache.cassandra.service.accord.txn.TxnRead;
-import org.apache.cassandra.service.paxos.PaxosState;
 import org.apache.cassandra.utils.Clock;
 
 import static accord.api.Journal.CommandUpdate;
@@ -584,13 +583,6 @@ void maybeLoadRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch)
             loadRangesForEpoch(rangesForEpoch);
     }
 
-    @Override
-    public void updateRangesForEpoch(SafeCommandStore safeStore)
-    {
-        super.updateRangesForEpoch(safeStore);
-        updateMinHlc(PaxosState.ballotTracker().getLowBound().unixMicros() + 1);
-    }
-
     // TODO (expected): handle journal failures, and consider how we handle partial failures.
     //  Very likely we will not be able to safely or cleanly handle partial failures of this logic, but decide and document.
     // TODO (desired): consider merging with PersistentField? This version is cheaper to manage which may be preferable at the CommandStore level.

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

@@ -389,12 +389,17 @@ public boolean hasNext()
             public TopologyUpdate next()
             {
                 Journal.KeyRefs<JournalKey> ref = iter.next();
-                System.out.println(ref.key());
-                Accumulator read = readAll(ref.key());
-                if (read.accumulated.kind() == Kind.NoOp)
-                    prev = read.accumulated.asImage(Invariants.nonNull(prev.getUpdate()));
-                else
-                    prev = read.accumulated;
+                Accumulator reader = readAll(ref.key());
+                if (reader.read().kind() == Kind.Repeat)
+                {
+                    if (prev == null)
+                    {
+                        logger.error("Encountered TopologyImage Repeat record for epoch {}, but no prior image record was found", ref.key().id.epoch());
+                        return null;
+                    }
+                    prev = reader.read().asImage(Invariants.nonNull(prev.getUpdate()));
+                }
+                else prev = reader.read();
 
                 return new TopologyUpdate(prev.getUpdate().commandStores,
                                           prev.getUpdate().global);
@@ -411,6 +416,9 @@ public void close()
             while (iter.hasNext())
             {
                 TopologyUpdate next = iter.next();
+                if (next == null)
+                    continue;
+
                 Invariants.require(prev == null || next.global.epoch() > prev.global.epoch());
                 // Due to partial compaction, we can clean up only some of the old epochs, creating gaps. We skip these epochs here.
                 if (prev != null && next.global.epoch() > prev.global.epoch() + 1)

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

@@ -42,6 +42,7 @@
 import accord.primitives.Unseekables;
 import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
 import org.apache.cassandra.service.accord.AccordCommandStore.SafeRedundantBefore;
+import org.apache.cassandra.service.paxos.PaxosState;
 
 import static accord.utils.Invariants.illegalState;
 
@@ -169,6 +170,13 @@ protected AccordSafeCommandsForKey getInternal(RoutingKey key)
         return commandsForKey.get(key);
     }
 
+    @Override
+    public void setRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch)
+    {
+        super.setRangesForEpoch(rangesForEpoch);
+        commandStore.updateMinHlc(PaxosState.ballotTracker().getLowBound().unixMicros() + 1);
+    }
+
     @Override
     public AccordCommandStore commandStore()
     {

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

@@ -39,6 +39,7 @@
 import com.google.common.primitives.Ints;
 
 import org.apache.cassandra.metrics.AccordReplicaMetrics;
+import org.apache.cassandra.service.accord.api.AccordViolationHandler;
 import org.apache.cassandra.utils.Clock;
 import org.apache.cassandra.utils.concurrent.ImmediateFuture;
 import org.slf4j.Logger;
@@ -132,6 +133,7 @@
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.concurrent.AsyncPromise;
 import org.apache.cassandra.utils.concurrent.Future;
+import org.apache.cassandra.utils.concurrent.Promise;
 import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
 
 import static accord.api.Journal.TopologyUpdate;
@@ -315,7 +317,10 @@ public synchronized static AccordService startup(NodeId tcmId)
         as.node.durability().start();
 
         instance = as;
+        
         AccordReplicaMetrics.touch();
+        AccordViolationHandler.setup();
+
         WatermarkCollector.fetchAndReportWatermarksAsync(as.configService);
         return as;
     }
@@ -596,11 +601,27 @@ public static <V> V getBlocking(AsyncChain<V> async, @Nullable TxnId txnId, Seek
         async.begin(result);
         return result.awaitAndGet();
     }
+
     public static <V> V getBlocking(AsyncChain<V> async, Seekables<?, ?> keysOrRanges, RequestBookkeeping bookkeeping, long startedAt, long deadline)
     {
         return getBlocking(async, keysOrRanges, bookkeeping, startedAt, deadline, false);
     }
 
+    public static <V> V getBlocking(AsyncChain<V> async)
+    {
+        return asPromise(async).syncUninterruptibly().getNow();
+    }
+
+    public static <V> Promise<V> asPromise(AsyncChain<V> async)
+    {
+        AsyncPromise<V> promise = new AsyncPromise<>();
+        async.begin((result, failure) -> {
+            if (failure == null) promise.trySuccess(result);
+            else promise.tryFailure(failure);
+        });
+        return promise;
+    }
+
     public static Keys intersecting(Keys keys)
     {
         if (keys.isEmpty())
@@ -974,7 +995,7 @@ public Node node()
     @Override
     public void ensureMinHlc(long minHlc)
     {
-        node.updateMinHlc(minHlc >= 0 ? minHlc : 0);
+        asPromise(node.updateMinHlc(minHlc >= 0 ? minHlc : 0)).syncUninterruptibly();
     }
 
     public AccordJournal journal()
@@ -985,13 +1006,7 @@ public AccordJournal journal()
     @Override
     public Future<Void> epochReady(Epoch epoch)
     {
-        AsyncPromise<Void> promise = new AsyncPromise<>();
-        AsyncChain<Void> ready = configService.epochReady(epoch.getEpoch());
-        ready.begin((result, failure) -> {
-            if (failure == null) promise.trySuccess(result);
-            else promise.tryFailure(failure);
-        });
-        return promise;
+        return asPromise(configService.epochReady(epoch.getEpoch()));
     }
 
     @Override

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

@@ -131,15 +131,6 @@ public void setNodeId(Node.Id id)
         self = id;
     }
 
-    @Override
-    public void onInconsistentTimestamp(Command command, Timestamp prev, Timestamp next)
-    {
-        // TODO (expected): better reporting
-        AssertionError error = new AssertionError("Inconsistent execution timestamp detected for txnId " + command.txnId() + ": " + prev + " != " + next);
-        onUncaughtException(error);
-        throw error;
-    }
-
     @Override
     public void onFailedBootstrap(int attempts, String phase, Ranges ranges, Runnable retry, Throwable failure)
     {

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

@@ -18,28 +18,37 @@
 
 package org.apache.cassandra.service.accord.api;
 
-import javax.annotation.Nullable;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import accord.api.ViolationHandler;
+import accord.local.Command;
+import accord.local.SafeCommandStore;
 import accord.primitives.Participants;
+import accord.primitives.Route;
 import accord.primitives.Timestamp;
 import accord.primitives.TxnId;
 
+import static accord.utils.Invariants.illegalState;
+
 public class AccordViolationHandler implements ViolationHandler
 {
     private static final Logger logger = LoggerFactory.getLogger(AccordViolationHandler.class);
 
-    static
+    public static void setup()
     {
         ViolationHandlerHolder.set(AccordViolationHandler::new);
     }
 
     @Override
-    public void onViolation(String message, Participants<?> participants, @Nullable TxnId notWitnessed, @Nullable Timestamp notWitnessedExecuteAt, @Nullable TxnId by, @Nullable Timestamp byEexecuteAt)
+    public void onTimestampViolation(SafeCommandStore safeStore, Command command, Participants<?> otherParticipants, Route<?> otherRoute, Timestamp otherExecuteAt)
+    {
+        throw illegalState(ViolationHandler.timestampViolationMessage(safeStore, command, otherParticipants, otherRoute, otherExecuteAt));
+    }
+
+    @Override
+    public void onDependencyViolation(Participants<?> participants, TxnId notWitnessed, Timestamp notWitnessedExecuteAt, TxnId by, Timestamp byExecuteAt)
     {
-        logger.error(message);
+        logger.error(ViolationHandler.dependencyViolationMessage(participants, notWitnessed, notWitnessedExecuteAt, by, byExecuteAt));
     }
 }