Accord Fixes:

 - Don't try to update metrics when no TxnId (e.g. GetMaxConflict)
 - maybeExecuteImmediately did not guarantee mutual exclusivity
 - Cancellation of a running 'plain' task could corrupt AccordExecutor state
 - GetLatestDeps message serializer
 - txn_blocked_by StackOverflowError
 - Not updating CommandsForKey in all cases on restart
Also Improve:
 - TableId.from/toString
 - Route toString methods
 - Tracing coverage of FetchRoute
 - Replay command store parallelism

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

Author: belliottsmith

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
 	url = https://github.com/apache/cassandra-accord.git
-	branch = trunk
+	branch = trunk

modules/accord

@@ -30,6 +30,11 @@
 public interface AsymmetricUnversionedSerializer<In, Out>
 {
     void serialize(In t, DataOutputPlus out) throws IOException;
+
+    /**
+     * Note: it is not guaranteed that this output is compatible with the DataInput/OutputPlus variations,
+     * as the ByteBuffer has an implied length.
+     */
     default ByteBuffer serialize(In t) throws IOException
     {
         int size = Math.toIntExact(serializedSize(t));
@@ -59,6 +64,11 @@ default ByteBuffer serializeUnchecked(In t)
         }
     }
     Out deserialize(DataInputPlus in) throws IOException;
+
+    /**
+     * Note: it is not guaranteed to be safe to provide an input created by the DataOutputPlus serializer varation
+     * as the ByteBuffer has an implied length.
+     */
     default Out deserialize(ByteBuffer buffer) throws IOException
     {
         try (DataInputBuffer in = new DataInputBuffer(buffer, true))
@@ -78,5 +88,6 @@ default Out deserializeUnchecked(ByteBuffer buffer)
             throw new UncheckedIOException(e);
         }
     }
+
     long serializedSize(In t);
 }

src/java/org/apache/cassandra/io/AsymmetricUnversionedSerializer.java

@@ -1097,6 +1097,12 @@ public ISSTableScanner getScanner(Iterator<AbstractBounds<PartitionPosition>> bo
         return new SSTableSimpleScanner(this, getPositionsForBoundsIterator(boundsIterator));
     }
 
+    public ISSTableScanner getScanner(AbstractBounds<PartitionPosition> bounds)
+    {
+        PartitionPositionBounds positionBounds = getPositionsForBounds(bounds);
+        return new SSTableSimpleScanner(this, positionBounds == null ? Collections.emptyList() : Collections.singletonList(positionBounds));
+    }
+
 
     /**
      * Create a {@link FileDataInput} for the data file of the sstable represented by this reader. This method returns

src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java

@@ -30,6 +30,9 @@
 import accord.utils.Invariants;
 import com.codahale.metrics.Timer;
 import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.utils.*;
 import org.apache.cassandra.utils.concurrent.OpOrder;
@@ -42,6 +45,8 @@
 @Simulate(with=MONITORS)
 public final class ActiveSegment<K, V> extends Segment<K, V>
 {
+    private static final Logger logger = LoggerFactory.getLogger(ActiveSegment.class);
+
     final FileChannel channel;
 
     // OpOrder used to order appends wrt flush
@@ -197,6 +202,7 @@ void persistComponents()
 
     private void discard()
     {
+        logger.debug("Discarding {}", this);
         selfRef.ensureReleased();
 
         descriptor.fileFor(Component.DATA).deleteIfExists();

src/java/org/apache/cassandra/journal/ActiveSegment.java

@@ -936,9 +936,9 @@ public interface Writer
     /**
      * Static segment iterator iterates all keys in _static_ segments in order.
      */
-    public StaticSegmentKeyIterator staticSegmentKeyIterator()
+    public StaticSegmentKeyIterator staticSegmentKeyIterator(K min, K max)
     {
-        return new StaticSegmentKeyIterator();
+        return new StaticSegmentKeyIterator(min, max);
     }
 
     /**
@@ -1011,17 +1011,22 @@ public class StaticSegmentKeyIterator implements CloseableIterator<KeyRefs<K>>
         private final ReferencedSegments<K, V> segments;
         private final MergeIterator<Head, KeyRefs<K>> iterator;
 
-        public StaticSegmentKeyIterator()
+        public StaticSegmentKeyIterator(K min, K max)
         {
-            this.segments = selectAndReference(Segment::isStatic);
+            this.segments = selectAndReference(s -> s.isStatic() && (min == null || keySupport.compare(s.index().lastId(), min) >= 0) && (max == null || keySupport.compare(s.index().firstId(), max) <= 0));
             List<Iterator<Head>> iterators = new ArrayList<>(segments.count());
 
             for (Segment<K, V> segment : segments.allSorted(true))
             {
-                StaticSegment<K, V> staticSegment = (StaticSegment<K, V>) segment;
+                final StaticSegment<K, V> staticSegment = (StaticSegment<K, V>) segment;
+                final OnDiskIndex<K>.IndexReader iter = staticSegment.index().reader();
+                if (min != null) iter.seek(min);
+                if (max != null) iter.seekEnd(max);
+                if (!iter.hasNext())
+                    continue;
+
                 iterators.add(new AbstractIterator<>()
                 {
-                    final Iterator<K> iter = staticSegment.index().reader();
                     final Head head = new Head(staticSegment.descriptor.timestamp);
 
                     @Override

src/java/org/apache/cassandra/journal/Journal.java

@@ -252,6 +252,7 @@ IndexReader reader()
 
     public class IndexReader extends AbstractIterator<K>
     {
+        int lastIdx = entryCount - 1;
         int idx;
         K key;
         int offset;
@@ -262,6 +263,20 @@ public class IndexReader extends AbstractIterator<K>
             idx = -1;
         }
 
+        public void seek(K key)
+        {
+            int i = binarySearch(key);
+            if (i < 0) i = -1 - i;
+            idx = i - 1;
+        }
+
+        public void seekEnd(K key)
+        {
+            int i = binarySearch(key);
+            if (i < 0) i = -2 - i;
+            lastIdx = i;
+        }
+
         protected K computeNext()
         {
             if (advance())
@@ -284,7 +299,7 @@ public int recordSize()
 
         public boolean advance()
         {
-            if (idx >= entryCount - 1)
+            if (idx >= lastIdx)
                 return false;
 
             idx++;

src/java/org/apache/cassandra/journal/OnDiskIndex.java

@@ -220,6 +220,12 @@ public void close()
         }
     }
 
+    @Override
+    public String toString()
+    {
+        return sorted.toString();
+    }
+
     private static final Long2ObjectHashMap<?> EMPTY_MAP = new Long2ObjectHashMap<>();
 
     @SuppressWarnings("unchecked")

src/java/org/apache/cassandra/journal/Segments.java

@@ -164,6 +164,8 @@ public void close(Journal<K, V> journal)
      */
     void discard(Journal<K, V> journal)
     {
+        logger.debug("Discarding {}", this);
+
         ((Tidier)selfRef.tidier()).discard = true;
         close(journal);
     }

src/java/org/apache/cassandra/journal/StaticSegment.java

@@ -201,12 +201,14 @@ public Listener(AccordMetrics readMetrics, AccordMetrics writeMetrics)
 
         private AccordMetrics forTransaction(TxnId txnId)
         {
-            if (txnId.isWrite())
-                return writeMetrics;
-            else if (txnId.isSomeRead())
-                return readMetrics;
-            else
-                return null;
+            if (txnId != null)
+            {
+                if (txnId.isWrite())
+                    return writeMetrics;
+                else if (txnId.isSomeRead())
+                    return readMetrics;
+            }
+            return null;
         }
 
         @Override