Improve CMS initialization

* Better handling of DOWN unupgraded nodes
* Test that aborted CMS initialization cleans up state properly
* Clean up orphaned PreInitialize entries in the log on bounce
* Unconditionally reset initiator during abort
* Ensure that metadata log entries aren't exchanged before CMS
  initialization is complete
* Precalculate common serialization version, excluding non-upgraded and
  LEFT nodes
* Decide if metadata-impacting upgrade is in progress using min common
  version
* Add metadata identifier to nodetool cms output

Patch by Sam Tunnicliffe and Marcus Eriksson; reviewed by Sam
Tunnicliffe and Marcus Eriksson for CASSANDRA-21036

Co-authored-by: Marcus Eriksson <[email protected]>
Co-authored-by: Sam Tunnicliffe <[email protected]>

Author: beobal

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Improve CMS initialization (CASSANDRA-21036)
  * Introducing comments and security labels for schema elements (CASSANDRA-20943)
  * Extend nodetool tablestats for dictionary memory usage (CASSANDRA-20940)
  * Introduce separate GCInspector thresholds for concurrent GC events (CASSANDRA-20980)

src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java

@@ -56,7 +56,7 @@ public void doVerb(Message<T> message) throws IOException
 
     protected void processMessage(Message<T> message, InetAddressAndPort respondTo)
     {
-        if (message.epoch().isAfter(Epoch.EMPTY))
+        if (message.epoch().isAfter(Epoch.FIRST))
         {
             ClusterMetadata metadata = ClusterMetadata.current();
             metadata = checkTokenOwnership(metadata, message, respondTo);

src/java/org/apache/cassandra/gms/GossipVerbHandler.java

@@ -21,12 +21,14 @@
 import org.apache.cassandra.net.IVerbHandler;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
 
 public class GossipVerbHandler<T> implements IVerbHandler<T>
 {
     public void doVerb(Message<T> message)
     {
         Gossiper.instance.setLastProcessedMessageAt(message.creationTimeMillis());
-        ClusterMetadataService.instance().fetchLogFromPeerAsync(message.from(), message.epoch());
+        if (message.epoch().isAfter(Epoch.FIRST))
+            ClusterMetadataService.instance().fetchLogFromPeerAsync(message.from(), message.epoch());
     }
 }

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

@@ -28,6 +28,7 @@
 import org.apache.cassandra.exceptions.RequestFailureReason;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.FBUtilities;
 
@@ -92,6 +93,9 @@ public void doVerb(Message message)
     private void maybeFetchLogs(Message<?> message)
     {
         ClusterMetadata metadata = ClusterMetadata.current();
+        if (!message.epoch().isAfter(Epoch.FIRST))
+            return;
+
         if (!message.epoch().isAfter(metadata.epoch))
             return;
 

src/java/org/apache/cassandra/tcm/CMSOperations.java

@@ -63,6 +63,7 @@ public class CMSOperations implements CMSOperationsMBean
     public static final String LOCAL_PENDING = "LOCAL_PENDING";
     public static final String COMMITS_PAUSED = "COMMITS_PAUSED";
     public static final String REPLICATION_FACTOR = "REPLICATION_FACTOR";
+    public static final String CMS_ID = "CMS_ID";
 
     private static final Logger logger = LoggerFactory.getLogger(ClusterMetadataService.class);
     public static CMSOperations instance = new CMSOperations(ClusterMetadataService.instance());
@@ -161,6 +162,7 @@ public Map<String, String> describeCMS()
         info.put(LOCAL_PENDING, Integer.toString(cms.log().pendingBufferSize()));
         info.put(COMMITS_PAUSED, Boolean.toString(cms.commitsPaused()));
         info.put(REPLICATION_FACTOR, ReplicationParams.meta(metadata).toString());
+        info.put(CMS_ID, Integer.toString(metadata.metadataIdentifier));
         return info;
     }
 

src/java/org/apache/cassandra/tcm/ClusterMetadata.java

@@ -1004,7 +1004,7 @@ public NodeState myNodeState()
 
     public boolean metadataSerializationUpgradeInProgress()
     {
-        return !directory.clusterMaxVersion.serializationVersion().equals(directory.clusterMinVersion.serializationVersion());
+        return !directory.clusterMaxVersion.serializationVersion().equals(directory.commonSerializationVersion);
     }
 
     public static class Serializer implements MetadataSerializer<ClusterMetadata>

src/java/org/apache/cassandra/tcm/Startup.java

@@ -58,8 +58,8 @@
 import org.apache.cassandra.tcm.log.SystemKeyspaceStorage;
 import org.apache.cassandra.tcm.membership.NodeId;
 import org.apache.cassandra.tcm.membership.NodeState;
-import org.apache.cassandra.tcm.migration.Election;
 import org.apache.cassandra.tcm.migration.CMSInitializationRequest;
+import org.apache.cassandra.tcm.migration.Election;
 import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
 import org.apache.cassandra.tcm.sequences.InProgressSequences;
 import org.apache.cassandra.tcm.sequences.ReconfigureCMS;
@@ -77,7 +77,7 @@
 import static org.apache.cassandra.tcm.membership.NodeState.LEFT;
 import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
 
- /**
+/**
   * Initialize
   */
  public class Startup
@@ -141,7 +141,6 @@ public static void initializeAsFirstCMSNode()
         ClusterMetadataService.instance().log().bootstrap(addr, datacenter);
         ClusterMetadata metadata =  ClusterMetadata.current();
         assert ClusterMetadataService.state() == LOCAL : String.format("Can't initialize as node hasn't transitioned to CMS state. State: %s.\n%s", ClusterMetadataService.state(),  metadata);
-
         Initialize initialize = new Initialize(metadata.initializeClusterIdentifier(addr.hashCode()));
         ClusterMetadataService.instance().commit(initialize);
     }

src/java/org/apache/cassandra/tcm/log/SystemKeyspaceStorage.java

@@ -87,12 +87,25 @@ public void append(Entry entry)
 
     public synchronized static boolean hasAnyEpoch()
     {
-        String query = String.format("SELECT epoch FROM %s.%s LIMIT 1", SchemaConstants.SYSTEM_KEYSPACE_NAME, NAME);
+        String query = String.format("SELECT epoch, kind FROM %s.%s LIMIT 2", SchemaConstants.SYSTEM_KEYSPACE_NAME, NAME);
 
+        int count = 0;
+        long preInitializeEpoch = -1;
         for (UntypedResultSet.Row row : executeInternal(query))
-            return true;
+        {
+            count++;
+            if (Transformation.Kind.fromId(row.getInt("kind")) == Transformation.Kind.PRE_INITIALIZE_CMS)
+                preInitializeEpoch = row.getLong("epoch");
+        }
+        if (count == 1 && preInitializeEpoch != -1)
+        {
+            logger.warn("Cleaning up orphaned PreInitialize at epoch {} - restarting in gossip mode", preInitializeEpoch);
+            String cleanupQuery = String.format("DELETE from %s.%s where epoch = %d", SchemaConstants.SYSTEM_KEYSPACE_NAME, NAME, preInitializeEpoch);
+            executeInternal(cleanupQuery);
+            return false;
+        }
 
-        return false;
+        return count > 0;
     }
 
     @Override

src/java/org/apache/cassandra/tcm/membership/Directory.java

@@ -73,6 +73,7 @@ public class Directory implements MetadataValue<Directory>
     private final BTreeMap<String, Multimap<String, InetAddressAndPort>> racksByDC;
     public final NodeVersion clusterMinVersion;
     public final NodeVersion clusterMaxVersion;
+    public final Version commonSerializationVersion;
 
     public Directory()
     {
@@ -115,6 +116,7 @@ private Directory(int nextId,
         Pair<NodeVersion, NodeVersion> minMaxVer = minMaxVersions(states, versions);
         clusterMinVersion = minMaxVer.left;
         clusterMaxVersion = minMaxVer.right;
+        commonSerializationVersion = minCommonSerializationVersion(states, versions);
     }
 
     @Override
@@ -131,6 +133,9 @@ public String toString()
                ", hostIds=" + hostIds +
                ", endpointsByDC=" + endpointsByDC +
                ", racksByDC=" + racksByDC +
+               ", clusterMinVersion=" + clusterMinVersion +
+               ", clusterMaxVersion=" + clusterMaxVersion +
+               ", commonSerializationVersion=" + commonSerializationVersion +
                '}';
     }
 
@@ -778,6 +783,21 @@ private static Pair<NodeVersion, NodeVersion> minMaxVersions(BTreeMap<NodeId, No
         return Pair.create(minVersion, maxVersion);
     }
 
+    public static Version minCommonSerializationVersion(BTreeMap<NodeId, NodeState> states, BTreeMap<NodeId, NodeVersion> versions)
+    {
+        int commonVersion = Integer.MAX_VALUE;
+        for (Map.Entry<NodeId, NodeState> entry : states.entrySet())
+        {
+            if (entry.getValue() != NodeState.LEFT)
+            {
+                NodeVersion ver = versions.get(entry.getKey());
+                if (ver.serializationVersion > Version.OLD.asInt() && ver.serializationVersion < commonVersion)
+                    commonVersion = ver.serializationVersion;
+            }
+        }
+        return commonVersion == Integer.MAX_VALUE ? NodeVersion.CURRENT_METADATA_VERSION : Version.fromInt(commonVersion);
+    }
+
     @Override
     public int hashCode()
     {

src/java/org/apache/cassandra/tcm/membership/NodeVersion.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.util.Objects;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.tcm.serialization.MetadataSerializer;
@@ -39,6 +40,8 @@ public class NodeVersion implements Comparable<NodeVersion>
     private static final CassandraVersion SINCE_VERSION = CassandraVersion.CASSANDRA_5_1;
 
     public final CassandraVersion cassandraVersion;
+
+    // this must be kept as an int, otherwise we can't deserialize future NodeVersions
     public final int serializationVersion;
 
     public NodeVersion(CassandraVersion cassandraVersion, Version serializationVersion)
@@ -113,10 +116,9 @@ public static class Serializer implements MetadataSerializer<NodeVersion>
         @Override
         public void serialize(NodeVersion t, DataOutputPlus out, Version version) throws IOException
         {
-            out.writeUTF(t.cassandraVersion.toString());
             if (t.serializationVersion == Version.UNKNOWN.asInt())
                 throw new IllegalStateException("Should not serialize UNKNOWN version");
-            out.writeUnsignedVInt32(t.serializationVersion);
+            serializeHelper(out, t.cassandraVersion.toString(), t.serializationVersion);
         }
 
         @Override
@@ -133,5 +135,16 @@ public long serializedSize(NodeVersion t, Version version)
             return sizeof(t.cassandraVersion.toString()) +
                    sizeofUnsignedVInt(t.serializationVersion);
         }
+        /**
+         * Used to be able to generate a NodeVersion with a future serialization version, for test!
+         */
+        @VisibleForTesting
+        static void serializeHelper(DataOutputPlus out, String cassandraVersion, int serializationVersion) throws IOException
+        {
+            out.writeUTF(cassandraVersion);
+            out.writeUnsignedVInt32(serializationVersion);
+        }
     }
+
+
 }