CNDB-13203: Use MessagingSuccess's version for cnx version (#1621)

### What is the issue
Fixes riptano/cndb#13203

The underlying problem is that the `OutboundConnection` needs to store
the message version from the peer. Otherwise, when the local
`current_version` is higher than the remove `current_version`, we store
the wrong value (we were storing the local value, not the agreed upon
value).

The PR also fixes an issue in the ANN_OPTIONS serialization logic that
we can hit in rare cases where the peers haven't yet connected, so the
validation logic that would prevent serialization due to a peer having
too low of a message version is skipped.

### What does this PR fix and why was it fixed
This PR fixes several issues related to upgrades by improving handshake
logic.

---------

Co-authored-by: Sergio Bossa <[email protected]>

Author: michaeljmarshall

src/java/org/apache/cassandra/db/filter/ANNOptions.java

@@ -179,7 +179,11 @@ public void serialize(ANNOptions options, DataOutputPlus out, int version) throw
         {
             // ANN options are only supported in DS 11 and above, so don't serialize anything if the messaging version is lower
             if (version < MessagingService.VERSION_DS_11)
+            {
+                if (options != NONE)
+                    throw new IllegalStateException("Unable to serialize ANN options with messaging version: " + version);
                 return;
+            }
 
             int flags = flags(options);
             out.writeInt(flags);

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

@@ -37,6 +37,15 @@ public class EndpointMessagingVersions
     // protocol versions of the other nodes in the cluster
     private final ConcurrentMap<InetAddressAndPort, Integer> versions = new NonBlockingHashMap<>();
 
+    public EndpointMessagingVersions()
+    {
+    }
+
+    private EndpointMessagingVersions(EndpointMessagingVersions versions)
+    {
+        this.versions.putAll(versions.versions);
+    }
+
     /**
      * @return the last version associated with address, or @param version if this is the first such version
      */
@@ -91,4 +100,9 @@ public boolean knows(InetAddressAndPort endpoint)
     {
         return versions.containsKey(endpoint);
     }
+
+    public EndpointMessagingVersions copy()
+    {
+        return new EndpointMessagingVersions(this);
+    }
 }

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

@@ -75,7 +75,7 @@ public class InboundMessageHandler extends AbstractMessageHandler
     private final ConnectionType type;
     private final InetAddressAndPort self;
     private final InetAddressAndPort peer;
-    private final int version;
+    final int version;
 
     private final InboundMessageCallbacks callbacks;
     private final Consumer<Message<?>> consumer;

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

@@ -437,6 +437,13 @@ private long sumCounters(ToLongFunction<InboundCounters> mapping)
              + mapping.applyAsLong(legacyCounters);
     }
 
+    @VisibleForTesting
+    public void assertHandlersMessagingVersion(int expectedVersion)
+    {
+        for (InboundMessageHandler handler : handlers)
+            assert handler.version == expectedVersion : "Expected all handlers to be at version " + expectedVersion + " but found " + handler.version;
+    }
+
     interface HandlerProvider
     {
         InboundMessageHandler provide(FrameDecoder decoder,

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

@@ -1117,6 +1117,8 @@ void onCompletedHandshake(Result<MessagingSuccess> result)
                         assert !state.isClosed();
 
                         MessagingSuccess success = result.success();
+                        messagingVersion = success.messagingVersion;
+                        settings.endpointToVersion.set(settings.to, messagingVersion);
                         debug.onConnect(success.messagingVersion, settings);
                         state.disconnected().maintenance.cancel(false);
 

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

@@ -498,7 +498,10 @@ public OutboundConnectionSettings withDefaults(ConnectionCategory category)
                                               applicationSendQueueReserveGlobalCapacityInBytes(),
                                               tcpNoDelay(), flushLowWaterMark, flushHighWaterMark,
                                               tcpConnectTimeoutInMS(), tcpUserTimeoutInMS(category), acceptVersions(category),
-                                              from(), socketFactory(), callbacks(), debug(), endpointToVersion());
+                                              from(), socketFactory(), callbacks(), debug(),
+                                              // If a set of versions is passed, make sure we do a copy of it, as the version might be later updated
+                                              // depending on the handshake result (i.e. nodes might handshake a different version)
+                                              endpointToVersion().copy());
     }
 
     private static boolean isInLocalDC(IEndpointSnitch snitch, InetAddressAndPort localHost, InetAddressAndPort remoteHost)

test/unit/org/apache/cassandra/db/filter/ANNOptionsTest.java

@@ -241,13 +241,26 @@ private void testTransport(String query, ANNOptions expectedOptions)
 
             // ...with a version that doesn't support ANN options
             out = new DataOutputBuffer();
-            ReadCommand.serializer.serialize(command, out, MessagingService.VERSION_DS_10);
-            Assertions.assertThat(ReadCommand.serializer.serializedSize(command, MessagingService.VERSION_DS_10))
-                      .isEqualTo(out.buffer().remaining());
-            in = new DataInputBuffer(out.buffer(), true);
-            command = ReadCommand.serializer.deserialize(in, MessagingService.VERSION_DS_10);
-            actualOptions = command.rowFilter().annOptions();
-            Assertions.assertThat(actualOptions).isEqualTo(ANNOptions.NONE);
+            if (expectedOptions != ANNOptions.NONE) {
+                try
+                {
+                    ReadCommand.serializer.serialize(command, out, MessagingService.VERSION_DS_10);
+                }
+                catch (IllegalStateException e)
+                {
+                    // expected
+                    Assertions.assertThat(e)
+                              .hasMessageContaining("Unable to serialize ANN options with messaging version: " + MessagingService.VERSION_DS_10);
+                }
+            } else {
+                ReadCommand.serializer.serialize(command, out, MessagingService.VERSION_DS_10);
+                Assertions.assertThat(ReadCommand.serializer.serializedSize(command, MessagingService.VERSION_DS_10))
+                          .isEqualTo(out.buffer().remaining());
+                in = new DataInputBuffer(out.buffer(), true);
+                command = ReadCommand.serializer.deserialize(in, MessagingService.VERSION_DS_10);
+                actualOptions = command.rowFilter().annOptions();
+                Assertions.assertThat(actualOptions).isEqualTo(ANNOptions.NONE);
+            }
         }
         catch (IOException e)
         {

test/unit/org/apache/cassandra/net/ConnectionTest.java

@@ -74,6 +74,9 @@
 import static org.apache.cassandra.net.MessagingService.VERSION_30;
 import static org.apache.cassandra.net.MessagingService.VERSION_3014;
 import static org.apache.cassandra.net.MessagingService.VERSION_40;
+import static org.apache.cassandra.net.MessagingService.VERSION_DS_10;
+import static org.apache.cassandra.net.MessagingService.VERSION_DS_11;
+import static org.apache.cassandra.net.MessagingService.minimum_version;
 import static org.apache.cassandra.net.NoPayload.noPayload;
 import static org.apache.cassandra.net.MessagingService.current_version;
 import static org.apache.cassandra.net.ConnectionUtils.*;
@@ -121,17 +124,24 @@ public void resetVerbs() throws Throwable
         timeouts.clear();
     }
 
+    private static volatile long originalRpcTimeout = 0;
+
     @BeforeClass
     public static void startup()
     {
         DatabaseDescriptor.daemonInitialization();
         CommitLog.instance.start();
+        // At the time of this commit, the default is 20 seconds and leads to significant delays
+        // in this test class, especially in testMessagePurging and testCloseIfEndpointDown.
+        originalRpcTimeout = DatabaseDescriptor.getRpcTimeout(TimeUnit.MILLISECONDS);
+        DatabaseDescriptor.setRpcTimeout(5000L);
     }
 
     @AfterClass
     public static void cleanup() throws InterruptedException
     {
         factory.shutdownNow();
+        DatabaseDescriptor.setRpcTimeout(originalRpcTimeout);
     }
 
     interface SendTest
@@ -192,30 +202,54 @@ Settings override(Settings settings)
 
     // 30 is used for CNDB compatibility
     static final AcceptVersions legacy = new AcceptVersions(VERSION_3014, VERSION_3014);
+    static final AcceptVersions ds10 = new AcceptVersions(minimum_version, VERSION_DS_10);
+    static final AcceptVersions ds11 = new AcceptVersions(minimum_version, VERSION_DS_11);
+    static final AcceptVersions current = new AcceptVersions(current_version, current_version);
 
-    static final List<Function<Settings, Settings>> MODIFIERS = ImmutableList.of(
+    static final List<Function<Settings, Settings>> MESSAGGING_VERSIONS = ImmutableList.of(
         settings -> settings.outbound(outbound -> outbound.withAcceptVersions(legacy))
                             .inbound(inbound -> inbound.withAcceptMessaging(legacy)),
+        // Mismatched versions (in both directions) to ensure both peers will still agree on the same version.
+        settings -> settings.outbound(outbound -> outbound.withAcceptVersions(ds11))
+                            .inbound(inbound -> inbound.withAcceptMessaging(ds10)),
+        settings -> settings.outbound(outbound -> outbound.withAcceptVersions(ds10))
+                            .inbound(inbound -> inbound.withAcceptMessaging(ds11)),
+        // This setting ensures that we cover the current case for the power set where no versions are overridden.
+        settings -> settings.outbound(outbound -> outbound.withAcceptVersions(current))
+                            .inbound(inbound -> inbound.withAcceptMessaging(current))
+    );
+
+
+    static final List<Function<Settings, Settings>> MODIFIERS = ImmutableList.of(
         settings -> settings.outbound(outbound -> outbound.withEncryption(encryptionOptions))
                             .inbound(inbound -> inbound.withEncryption(encryptionOptions)),
         settings -> settings.outbound(outbound -> outbound.withFraming(LZ4))
     );
 
+    // Messaging versions are a kind of modifier, but they can only be applied once per setting, so they are broken
+    // out into a separate list.
     static final List<Settings> SETTINGS = applyPowerSet(
-        ImmutableList.of(Settings.SMALL, Settings.LARGE),
+        ImmutableList.of(ConnectionTest.Settings.SMALL, ConnectionTest.Settings.LARGE),
+        MESSAGGING_VERSIONS,
         MODIFIERS
     );
 
-    private static <T> List<T> applyPowerSet(List<T> settings, List<Function<T, T>> modifiers)
+    private static List<Settings> applyPowerSet(List<ConnectionTest.Settings> settings,
+                                                List<Function<ConnectionTest.Settings, ConnectionTest.Settings>> messagingVersions,
+                                                List<Function<ConnectionTest.Settings, ConnectionTest.Settings>> modifiers)
     {
-        List<T> result = new ArrayList<>();
-        for (Set<Function<T, T>> set : Sets.powerSet(new HashSet<>(modifiers)))
+        List<Settings> result = new ArrayList<>();
+        for (Function<ConnectionTest.Settings, ConnectionTest.Settings> messagingVersion : messagingVersions)
         {
-            for (T s : settings)
+            for (Set<Function<Settings, ConnectionTest.Settings>> set : Sets.powerSet(new HashSet<>(modifiers)))
             {
-                for (Function<T, T> f : set)
-                    s = f.apply(s);
-                result.add(s);
+                for (ConnectionTest.Settings s : settings)
+                {
+                    for (Function<Settings, ConnectionTest.Settings> f : set)
+                        s = f.apply(s);
+                    s = messagingVersion.apply(s);
+                    result.add(s);
+                }
             }
         }
         return result;
@@ -306,6 +340,10 @@ public void testSendSmall() throws Throwable
                            .expired  ( 0,  0)
                            .error    ( 0,  0)
                            .check();
+
+            // Ensure version is the same
+            inbound.assertHandlersMessagingVersion(outbound.messagingVersion());
+            Assert.assertEquals(outbound.settings().endpointToVersion.get(endpoint), outbound.messagingVersion());
         });
     }
 
@@ -360,6 +398,10 @@ public long serializedSize(Object noPayload, int version)
                            .expired  ( 0,  0)
                            .error    ( 0,  0)
                            .check();
+
+            // Ensure version is the same
+            inbound.assertHandlersMessagingVersion(outbound.messagingVersion());
+            Assert.assertEquals(outbound.settings().endpointToVersion.get(endpoint), outbound.messagingVersion());
         });
     }