Improve:

 - Journal debugging vtable support
 - Background task tracing support
Fix:
 - HLC_BOUND only valid for strictly lower HLC
 - HAS_UNIQUE_HLC can only be safely computed if READY_TO_EXECUTE
 - Break recursion in CommandStore.ensureReadyToCoordinate
 - Fix find intersecting shard scheduler
 - Separate adhoc ShardScheduler from normal to avoid overwriting
 - Should still use execution listeners to detect invalid reads for certain transactions even with dataStoreDetectsFutureReads

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20746

Author: belliottsmith

modules/accord

@@ -443,6 +443,12 @@ public Command loadCommand(TxnId txnId)
         return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), durableBefore());
     }
 
+    @VisibleForTesting
+    public List<AccordJournal.DebugEntry> debugCommand(TxnId txnId)
+    {
+        return (List<AccordJournal.DebugEntry>) journal.debugCommand(id, txnId);
+    }
+
     public static Command prepareToCache(Command command)
     {
         // TODO (required): validate we don't have duplicate objects

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

@@ -19,13 +19,16 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
 import java.util.NavigableMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -228,6 +231,39 @@ public Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redu
         return builder.construct(redundantBefore);
     }
 
+    public static class DebugEntry implements Supplier<CommandChange.Builder>
+    {
+        public final long segment;
+        public final int position;
+        public final Builder builder;
+
+        public DebugEntry(long segment, int position, Builder builder)
+        {
+            this.segment = segment;
+            this.position = position;
+            this.builder = builder;
+        }
+
+        @Override
+        public CommandChange.Builder get()
+        {
+            return builder;
+        }
+    }
+
+    @Override
+    public List<DebugEntry> debugCommand(int commandStoreId, TxnId txnId)
+    {
+        JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId);
+        List<DebugEntry> result = new ArrayList<>();
+        journalTable.readAll(key, (long segment, int position, JournalKey k, ByteBuffer buffer, int userVersion) -> {
+            Builder builder = new Builder(txnId);
+            new AccordJournalTable.RecordConsumerAdapter<>(builder::deserializeNext).accept(segment, position, k, buffer, userVersion);
+            result.add(new DebugEntry(segment, position, builder));
+        });
+        return result;
+    }
+
     @Override
     public Command.Minimal loadMinimal(int commandStoreId, TxnId txnId, Load load, RedundantBefore redundantBefore, DurableBefore durableBefore)
     {

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

@@ -192,7 +192,7 @@ public interface Reader
         void read(DataInputPlus input, Version userVersion) throws IOException;
     }
 
-    private static class RecordConsumerAdapter<K> implements RecordConsumer<K>
+    static class RecordConsumerAdapter<K> implements RecordConsumer<K>
     {
         protected final Reader reader;