HCD-136 – DSE 6.8 to CC4 internode transport compatibility

CC4 can only communicate with DSE 6.8 on messaging version VERSION_3014 (messaging/protocol version 11) and VERSION_3014 (messaging/protocol version 11).
Manully setting the CC4 node's `MessagingService.current_version` achieves this, done using the jvm flag `-Dds.current_messaging_version=11`.  Using VERSION_30 (messaging/protocol version 10) is also accepted, but hits the column filter bug on system tables `select *` queries during startup.

This commits automates the DSE 6.8 detection and negotiation when the maxMessagingVersion in the handshake is above 256, replying with a max accept of VERSION_3014. The ds.current_messaging_version flag is kept, as when it is used everything behaves as it previously did.  This is important for CNDB, as it hasn't dropped support for bdp/cndb-6.8 (a customised version of DSE 6.8).  The flag and fields related to it are marked deprecated and can be removed in CC5.

Logging in OutboundConnectionInitiator is improved with an introduced ClosedOutboundConnectionException that extends ClosedChannelException. ClosedOutboundConnectionException adds a message including the requestMessagingVersion, settings.framing and settings.encryption variables.

In StartupClusterConnectivityChecker,  don't send PING_REQ messages to any DSE 6.x/C* 3.x peers. They are not required (only speed up startup times), and isn't understood anyway by those peers.

Fix commit log versions, as they were being incorrectly written with filenames using messaging versions not corresponding to commit log versions.  Introduce CommitLogDescriptor.currentVersion() to translate messaging versions to commit log versions. This is important to honour custom ds.current_messaging_version values.

Author: michaelsembwever

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

@@ -371,7 +371,9 @@ public enum CassandraRelevantProperties
     /**
      * The current messaging version. This is used when we add new messaging versions without adopting them immediately,
      * or to force the node to use a specific version for testing purposes.
+     * @deprecated remove when cndb no longer supports bdp/6.8-cndb
      */
+    @Deprecated(since = "5.0")
     DS_CURRENT_MESSAGING_VERSION("ds.current_messaging_version", Integer.toString(MessagingService.VERSION_DS_20)),
     DTEST_API_LOG_TOPOLOGY("cassandra.dtest.api.log.topology"),
     /** This property indicates if the code is running under the in-jvm dtest framework */

src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java

@@ -76,9 +76,9 @@ public class CommitLogDescriptor
 
     /**
      * Increment this number if there is a changes in the commit log disc layout or MessagingVersion changes.
-     * Note: make sure to handle {@link #getMessagingVersion()}
+     * Note: make sure to handle  {@link #currentVersion()} and {@link #getMessagingVersion()}
      */
-    public static final int CURRENT_VERSION = MessagingService.current_version;
+    public static final int CURRENT_VERSION = currentVersion();
 
     final int version;
     public final long id;
@@ -226,7 +226,13 @@ private static Matcher extactFromFileName(String name)
 
     public int getMessagingVersion()
     {
-        switch (version)
+        return getMessagingVersion(version);
+    }
+
+    @VisibleForTesting
+    static int getMessagingVersion(int commitLogVersion)
+    {
+        switch (commitLogVersion)
         {
             case VERSION_30:
                 return MessagingService.Version.VERSION_30.value;
@@ -243,7 +249,37 @@ public int getMessagingVersion()
             case VERSION_DSE_68:
                 return MessagingService.Version.VERSION_DSE_68.value;
             default:
-                throw new IllegalStateException("Unknown commitlog version " + version);
+                throw new IllegalStateException("Unknown commitlog version " + commitLogVersion);
+        }
+    }
+
+    private static int currentVersion()
+    {
+        return currentVersion(MessagingService.current_version);
+    }
+
+    @VisibleForTesting
+    static int currentVersion(int messagingVersion)
+    {
+        switch(messagingVersion)
+        {
+            case MessagingService.VERSION_30:
+            case MessagingService.VERSION_3014:
+                return VERSION_30;
+            case MessagingService.VERSION_40:
+                return VERSION_40;
+            case MessagingService.VERSION_50:
+                return VERSION_50;
+            case MessagingService.VERSION_DSE_68:
+                return VERSION_DSE_68;
+            case MessagingService.VERSION_DS_10:
+                return VERSION_DS_10;
+            case MessagingService.VERSION_DS_11:
+                return VERSION_DS_11;
+            case MessagingService.VERSION_DS_20:
+                return VERSION_DS_20;
+            default:
+                throw new IllegalStateException("Unknown messaging version " + messagingVersion);
         }
     }
 

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

@@ -297,6 +297,7 @@ public static boolean supportsExtendedDeletionTime(int value)
     }
     static Map<Integer, Integer> versionOrdinalMap = Arrays.stream(Version.values()).collect(Collectors.toMap(v -> v.value, Enum::ordinal));
 
+    @Deprecated(since = "5.0") // remove when cndb no longer supports bdp/6.8-cndb
     private static int currentVersion()
     {
         int version = CassandraRelevantProperties.DS_CURRENT_MESSAGING_VERSION.getInt();

test/unit/org/apache/cassandra/db/commitlog/CommitLogDescriptorTest.java

@@ -349,4 +349,32 @@ public void testDSE68MessagingVersion()
         CommitLogDescriptor descriptor = new CommitLogDescriptor(680, 1, null, null);
         Assert.assertEquals(MessagingService.VERSION_DSE_68, descriptor.getMessagingVersion());
     }
+
+    @Test
+    public void testMessagingToCommitLogVersionSymmetry()
+    {
+        for (MessagingService.Version messagingVersion : MessagingService.Version.values())
+        {
+            int commitLogVersion = CommitLogDescriptor.currentVersion(messagingVersion.value);
+
+            // special case for MessagingService.VERSION_3014
+            int expectedMessagingVersion = messagingVersion.value;
+            if (MessagingService.Version.VERSION_3014 == messagingVersion)
+                expectedMessagingVersion = MessagingService.VERSION_30;
+
+            Assert.assertEquals(expectedMessagingVersion, CommitLogDescriptor.getMessagingVersion(commitLogVersion));
+        }
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testMessagingToCommitLogVersionFailure()
+    {
+        CommitLogDescriptor.currentVersion(1);
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void testCommitLogToMessagingVersionFailure()
+    {
+        CommitLogDescriptor.getMessagingVersion(1);
+    }
 }