Accord Fixes:

 - WatermarkCollector should not report same closed/retired epoch N times
 - AccordSyncPropagator can merge pending requests and back-off retries
 - AccordCommandLoader should notify listeners
 - AccordSegmentCompactor should estimate number of keys to ensure bloom filters work
 - txn_blocked_by table should report what it can, not throw IllegalStateException
 - Permit uncompressed system tables

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

Author: belliottsmith

modules/accord

@@ -173,7 +173,7 @@ public Keyspaces apply(ClusterMetadata metadata)
             throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
         }
 
-        if (!table.params.compression.isEnabled())
+        if (!table.params.compression.isEnabled() && !SchemaConstants.isSystemKeyspace(table.keyspace))
             Guardrails.uncompressedTablesEnabled.ensureEnabled(state);
 
         if (table.params.transactionalMode.accordIsEnabled && SchemaConstants.isSystemKeyspace(keyspaceName))

src/java/org/apache/cassandra/cql3/statements/schema/CreateTableStatement.java

@@ -998,7 +998,7 @@ public DataSet data(DecoratedKey partitionKey)
                 process(ds, commandStores, shard, processed, id, 0, id, Reason.Self, null);
                 // everything was processed right?
                 if (!shard.txns.isEmpty() && !shard.txns.keySet().containsAll(processed))
-                    throw new IllegalStateException("Skipped txns: " + Sets.difference(shard.txns.keySet(), processed));
+                    Invariants.expect(false, "Skipped txns: " + Sets.difference(shard.txns.keySet(), processed));
             }
 
             return ds;

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

@@ -199,6 +199,11 @@ public K lastId()
         return lastId;
     }
 
+    public int entryCount()
+    {
+        return entryCount;
+    }
+
     @Override
     public long[] lookUp(K id)
     {

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

@@ -239,6 +239,11 @@ OnDiskIndex<K> index()
         return index;
     }
 
+    public int entryCount()
+    {
+        return index.entryCount();
+    }
+
     @Override
     boolean isActive()
     {

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

@@ -84,7 +84,7 @@ boolean considerWritingKey()
         return false;
     }
 
-    abstract void initializeWriter();
+    abstract void initializeWriter(int estimatedKeyCount);
     abstract SSTableTxnWriter writer();
     abstract void finishAndAddWriter();
     abstract Throwable cleanupWriter(Throwable t);
@@ -99,9 +99,12 @@ public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment
         Invariants.require(segments.size() >= 2, () -> String.format("Can only compact 2 or more segments, but got %d", segments.size()));
         logger.info("Compacting {} static segments: {}", segments.size(), segments);
 
+        // TODO (expected): this will be a large over-estimate. should make segments an sstable format and include cardinality estimation
+        int estimatedKeyCount = 0;
         PriorityQueue<KeyOrderReader<JournalKey>> readers = new PriorityQueue<>();
         for (StaticSegment<JournalKey, V> segment : segments)
         {
+            estimatedKeyCount += segment.entryCount();
             KeyOrderReader<JournalKey> reader = segment.keyOrderReader();
             if (reader.advance())
                 readers.add(reader);
@@ -114,7 +117,7 @@ public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment
         if (readers.isEmpty())
             return Collections.emptyList();
 
-        initializeWriter();
+        initializeWriter(estimatedKeyCount);
 
         JournalKey key = null;
         FlyweightImage builder = null;

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

@@ -91,6 +91,7 @@
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.CompactionParams;
+import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.Indexes;
 import org.apache.cassandra.schema.KeyspaceMetadata;
@@ -144,8 +145,8 @@ public static TableMetadata journalMetadata(String tableName, boolean index)
                                                      + "user_version int,"
                                                      + "record blob,"
                                                      + "PRIMARY KEY((key), descriptor, offset)"
-                                                     + ") WITH CLUSTERING ORDER BY (descriptor DESC, offset DESC)" +
-                                                     " WITH compression = {'class':'NoopCompressor'};")
+                                                     + ") WITH CLUSTERING ORDER BY (descriptor DESC, offset DESC);")
+                                               .compression(CompressionParams.NOOP)
                                                .compaction(CompactionParams.lcs(emptyMap()))
                                                .bloomFilterFpChance(0.01)
                                                .partitioner(new LocalPartitioner(BytesType.instance));

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

@@ -37,12 +37,12 @@ public AccordSegmentCompactor(Version userVersion, ColumnFamilyStore cfs)
     }
 
     @Override
-    void initializeWriter()
+    void initializeWriter(int estimatedKeyCount)
     {
         Descriptor descriptor = cfs.newSSTableDescriptor(cfs.getDirectories().getDirectoryForNewSSTables());
         SerializationHeader header = new SerializationHeader(true, cfs.metadata(), cfs.metadata().regularAndStaticColumns(), EncodingStats.NO_STATS);
 
-        this.writer = SSTableTxnWriter.create(cfs, descriptor, 0, 0, null, false, header);
+        this.writer = SSTableTxnWriter.create(cfs, descriptor, estimatedKeyCount, 0, null, false, header);
     }
 
     @Override

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

@@ -24,6 +24,7 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.collect.ImmutableSet;
@@ -193,6 +194,7 @@ boolean ack(Node.Id id, Notification notifications)
     private final IFailureDetector failureDetector;
     private final ScheduledExecutorPlus scheduler;
     private final Listener listener;
+    private final ConcurrentHashMap<RetryKey, Notification> retryingNotifications = new ConcurrentHashMap<>();
 
     public AccordSyncPropagator(Node.Id localId, AccordEndpointMapper endpointMapper,
                                 MessageDelivery messagingService, IFailureDetector failureDetector, ScheduledExecutorPlus scheduler,
@@ -304,6 +306,27 @@ private boolean hasSyncCompletedFor(long epoch)
         });
     }
 
+    private void scheduleRetry(Node.Id to, Notification notification)
+    {
+        Notification retry = new Notification(notification.epoch, notification.syncComplete, notification.closed, notification.retired, notification.attempts + 1);
+        RetryKey key = new RetryKey(to, notification.epoch);
+        retryingNotifications.compute(key, (k, cur) -> {
+            if (cur == null)
+            {
+                scheduler.schedule(() -> retry(k), Math.max(1, Math.min(15, retry.attempts)), TimeUnit.MINUTES);
+                return retry;
+            }
+            return cur.merge(retry);
+        });
+    }
+
+    private void retry(RetryKey key)
+    {
+        Notification retry = retryingNotifications.remove(key);
+        if (retry != null)
+            notify(key.to, retry);
+    }
+
     private boolean notify(Node.Id to, Notification notification)
     {
         InetAddressAndPort toEp = endpointMapper.mappedEndpoint(to);
@@ -335,7 +358,7 @@ public void onResponse(Message<SimpleReply> msg)
             @Override
             public void onFailure(InetAddressAndPort from, RequestFailure failure)
             {
-                scheduler.schedule(() -> AccordSyncPropagator.this.notify(to, notification), 1, TimeUnit.SECONDS);
+                scheduleRetry(to, notification);
             }
 
             @Override
@@ -355,7 +378,7 @@ public boolean invokeOnFailure()
                 return true;
             }
             noSpamLogger.warn("Node{} is not alive, unable to notify of {}", to, notification);
-            scheduler.schedule(() -> notify(to, notification), 1, TimeUnit.MINUTES);
+            scheduleRetry(to, notification);
             return false;
         }
         messagingService.sendWithCallback(msg, toEp, cb);
@@ -397,13 +420,30 @@ public long serializedSize(Notification notification)
         final long epoch;
         final Collection<Node.Id> syncComplete;
         final Ranges closed, retired;
+        final int attempts;
 
         public Notification(long epoch, Collection<Node.Id> syncComplete, Ranges closed, Ranges retired)
+        {
+            this(epoch, syncComplete, closed, retired, 0);
+        }
+
+        public Notification(long epoch, Collection<Node.Id> syncComplete, Ranges closed, Ranges retired, int attempts)
         {
             this.epoch = epoch;
             this.syncComplete = syncComplete;
             this.closed = closed;
             this.retired = retired;
+            this.attempts = attempts;
+        }
+
+        Notification merge(Notification add)
+        {
+            Invariants.require(add.epoch == this.epoch);
+            Collection<Node.Id> syncComplete = ImmutableSet.<Node.Id>builder()
+                                                           .addAll(this.syncComplete)
+                                                           .addAll(add.syncComplete)
+                                                           .build();
+            return new Notification(epoch, syncComplete, closed.with(add.closed), retired.with(add.retired), Math.max(add.attempts, this.attempts));
         }
 
         @Override
@@ -417,4 +457,32 @@ public String toString()
                    '}';
         }
     }
+
+    static class RetryKey
+    {
+        final Node.Id to;
+        final long epoch;
+
+        RetryKey(Node.Id id, long epoch)
+        {
+            to = id;
+            this.epoch = epoch;
+        }
+
+        @Override
+        public int hashCode()
+        {
+            return to.id * 31 + (int)epoch;
+        }
+
+        @Override
+        public boolean equals(Object obj)
+        {
+            if (!(obj instanceof RetryKey))
+                return false;
+
+            RetryKey that = (RetryKey) obj;
+            return that.epoch == this.epoch && that.to.equals(this.to);
+        }
+    }
 }

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

@@ -135,22 +135,19 @@ static void fetchAndReportWatermarksAsync(AccordConfigurationService configServi
                                                                  MessageDelivery.RetryErrorMessage.EMPTY)
                .addCallback((m, fail) -> {
                    if (fail != null)
-                   {
                        return;
-                   }
+
                    Snapshot snapshot = m.payload;
                    long minEpoch = configService.minEpoch();
                    for (Map.Entry<Range, Long> e : snapshot.closed.entrySet())
                    {
                        Ranges r = Ranges.of(e.getKey());
-                       for (long epoch = minEpoch; epoch <= e.getValue(); epoch++)
-                           configService.receiveClosed(r, e.getValue());
+                       configService.receiveClosed(r, e.getValue());
                    }
                    for (Map.Entry<Range, Long> e : snapshot.retired.entrySet())
                    {
                        Ranges r = Ranges.of(e.getKey());
-                       for (long epoch = minEpoch; epoch <= e.getValue(); epoch++)
-                           configService.receiveRetired(r, e.getValue());
+                       configService.receiveRetired(r, e.getValue());
                    }
                    for (Map.Entry<Integer, Long> e : snapshot.synced.entrySet())
                    {