Add accord journal standalone dump tool

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20738

Author: ifesdjeen

.gitmodules

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

src/java/org/apache/cassandra/config/AccordSpec.java

@@ -196,10 +196,11 @@ public static class JournalSpec implements Params
         private volatile long flushCombinedBlockPeriod = Long.MIN_VALUE;
         public Version version = Version.DOWNGRADE_SAFE_VERSION;
 
-        public void setFlushPeriod(DurationSpec newFlushPeriod)
+        public JournalSpec setFlushPeriod(DurationSpec newFlushPeriod)
         {
             flushPeriod = newFlushPeriod;
             flushCombinedBlockPeriod = Long.MIN_VALUE;
+            return this;
         }
 
         public void setPeriodicFlushLagBlock(DurationSpec newPeriodicFlushLagBlock)

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -2234,7 +2234,7 @@ public static void createAllDirectories()
         }
         catch (FSWriteError e)
         {
-            throw new IllegalStateException(e.getCause().getMessage() + "; unable to start server");
+            throw new IllegalStateException(e.getCause().getMessage() + "; unable to start server", e);
         }
     }
 

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

@@ -109,7 +109,7 @@ static Descriptor fromName(File directory, String name)
         return new Descriptor(directory, timestamp, generation, journalVersion, userVersion);
     }
 
-    static Descriptor fromFile(File file)
+    public static Descriptor fromFile(File file)
     {
         return fromName(file.parent(), file.name());
     }

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

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.journal;
+
+import java.util.function.Consumer;
+
+/**
+ * Helper file to avoid exposing components outside their package-local visibility scope
+ */
+public class DumpUtil
+{
+    public static void dumpMetadata(Descriptor descriptor, Consumer<String> out)
+    {
+        if (Component.METADATA.existsFor(descriptor))
+        {
+            out.accept(Metadata.load(descriptor).toString());
+        }
+        else
+            out.accept("Metadata absent for " + descriptor);
+    }
+
+    public static <K, V> StaticSegment<K, V> open(Descriptor descriptor, KeySupport<K> keySupport)
+    {
+        return StaticSegment.open(descriptor, keySupport);
+    }
+}

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

@@ -154,4 +154,13 @@ static <K> Metadata rebuildAndPersist(Descriptor descriptor, KeySupport<K> keySu
         metadata.persist(descriptor);
         return metadata;
     }
+
+    @Override
+    public String toString()
+    {
+        return "Metadata{" +
+               "fsyncLimit=" + fsyncLimit +
+               ", recordsCount=" + recordsCount +
+               '}';
+    }
 }

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

@@ -289,7 +289,7 @@ boolean read(int offset, int size, EntrySerializer.EntryHolder<K> into)
     /**
      * Iterate over and invoke the supplied callback on every record.
      */
-    void forEachRecord(RecordConsumer<K> consumer)
+    public void forEachRecord(RecordConsumer<K> consumer)
     {
         try (SequentialReader<K> reader = sequentialReader(descriptor, keySupport, fsyncLimit))
         {

src/java/org/apache/cassandra/net/InboundMessageHandler.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
+import org.apache.cassandra.utils.ByteBufferUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -167,7 +168,14 @@ private void processSmallMessage(ShareableBytes bytes, int size, Header header)
         {
             Message<?> m = serializer.deserialize(in, header, version);
             if (in.available() > 0) // bytes remaining after deser: deserializer is busted
-                throw new InvalidSerializedSizeException(header.verb, size, size - in.available());
+            {
+                Throwable t = new InvalidSerializedSizeException(header.verb, size, size - in.available());
+                ByteBuffer clone = bytes.get();
+                clone.limit(clone.position() + size); // cap to expected message size
+                logger.error("Could not deserialize the message: {}", ByteBufferUtil.bytesToHex(clone), t);
+                throw t;
+
+            }
             message = m;
         }
         catch (IncompatibleSchemaException e)

src/java/org/apache/cassandra/schema/SchemaConstants.java

@@ -92,6 +92,14 @@ public static boolean isValidName(String name)
         emptyVersion = UUID.nameUUIDFromBytes(Digest.forSchema().digest());
     }
 
+    /**
+     * @return whether the table is an Accord Journal tabe
+     */
+    public static boolean isAccordJournal(String keyspaceName, String tableName)
+    {
+        return keyspaceName.equals(SchemaConstants.ACCORD_KEYSPACE_NAME) && tableName.equals(AccordKeyspace.JOURNAL);
+    }
+
     /**
      * @return whether or not the keyspace is a really system one (w/ LocalStrategy, unmodifiable, hardcoded)
      */

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

@@ -415,7 +415,7 @@ public Builder load(int commandStoreId, TxnId txnId)
         return loadDiffs(commandStoreId, txnId, Load.ALL);
     }
 
-    private <BUILDER extends FlyweightImage> BUILDER readAll(JournalKey key)
+    public <BUILDER extends FlyweightImage> BUILDER readAll(JournalKey key)
     {
         BUILDER builder = (BUILDER) key.type.serializer.mergerFor();
         // TODO (expected): this can be further improved to avoid allocating lambdas
@@ -425,6 +425,11 @@ private <BUILDER extends FlyweightImage> BUILDER readAll(JournalKey key)
         return builder;
     }
 
+    public void forEachEntry(JournalKey key, AccordJournalTable.Reader reader)
+    {
+        journalTable.readAll(key, reader);
+    }
+
     private <T> RecordPointer appendInternal(JournalKey key, T write)
     {
         AccordJournalValueSerializers.FlyweightSerializer<T, ?> serializer = (AccordJournalValueSerializers.FlyweightSerializer<T, ?>) key.type.serializer;