CNDB-12922 ANNOptions validation now checks the endpoint connection messaging version (#1624)

This makes sure the proper version is used, as
following CNDB-13203, each connection maintains its own version,
separated from the one recorded in the MessagingService.

### What is the issue
`ANNOptions#validate()` was previously checking endpoint versions via
`MessagingService#versions`: but following #13203, in order to resolve a
race condition, each endpoint connection tracks its own
`EndpointMessagingVersions` object, with the version negotiated during
handshake, while `MessagingService#versions` is updated with the max
version of each endpoint. It follows checking
`MessagingService#versions` might not rely on the right version used for
the connection.

### What does this PR fix and why was it fixed
This PR introduces a new `MessagingService` method to check the
connection version, and uses it inside `ANNOptions#validate()`.

Author: sbtourist

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

@@ -77,7 +77,7 @@ public void validate(QueryState state, String keyspace, int limit)
 
         // Ensure that all nodes in the cluster are in a version that supports ANN options, including this one
         assert keyspace != null;
-        Set<InetAddressAndPort> badNodes = MessagingService.instance().endpointsWithVersionBelow(keyspace, MessagingService.VERSION_DS_11);
+        Set<InetAddressAndPort> badNodes = MessagingService.instance().endpointsWithConnectionsOnVersionBelow(keyspace, MessagingService.VERSION_DS_11);
         if (MessagingService.current_version < MessagingService.VERSION_DS_11)
             badNodes.add(FBUtilities.getBroadcastAddressAndPort());
         if (!badNodes.isEmpty())

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

@@ -396,7 +396,7 @@ private void doSend(Message message, InetAddressAndPort to, ConnectionType speci
         // expire the callback if the message failed to enqueue (failed to establish a connection or exceeded queue capacity)
         while (true)
         {
-            OutboundConnections connections = getOutbound(to);
+            OutboundConnections connections = getOutbound(to, true);
             try
             {
                 connections.enqueue(message, specifyConnection);
@@ -602,10 +602,10 @@ private void shutdownExecutors(long deadlineNanos) throws TimeoutException, Inte
         socketFactory.awaitTerminationUntil(deadlineNanos);
     }
 
-    private OutboundConnections getOutbound(InetAddressAndPort to)
+    private OutboundConnections getOutbound(InetAddressAndPort to, boolean tryRegister)
     {
         OutboundConnections connections = channelManagers.get(to);
-        if (connections == null)
+        if (connections == null && tryRegister)
             connections = OutboundConnections.tryRegister(channelManagers, to, new OutboundConnectionSettings(to).withDefaults(ConnectionCategory.MESSAGING));
         return connections;
     }
@@ -662,4 +662,28 @@ public Set<InetAddressAndPort> endpointsWithVersionBelow(String keyspace, int ve
         }
         return nodes;
     }
+
+    /**
+     * Returns the endpoints for the given keyspace that are known to be alive and have a connection whose
+     * messaging version is older than the given version. To be used for example when we want to be sure a message
+     * can be serialized to all endpoints, according to their negotiated version at connection time.
+     *
+     * @param keyspace a keyspace
+     * @param version a messaging version
+     * @return a set of alive endpoints in the given keyspace with messaging version below the given version
+     */
+    public Set<InetAddressAndPort> endpointsWithConnectionsOnVersionBelow(String keyspace, int version)
+    {
+        Set<InetAddressAndPort> nodes = new HashSet<>();
+        for (InetAddressAndPort node : StorageService.instance.getTokenMetadataForKeyspace(keyspace).getAllEndpoints())
+        {
+            ConnectionType.MESSAGING_TYPES.forEach(type -> {
+                OutboundConnections connections = getOutbound(node, false);
+                OutboundConnection connection = connections != null ? connections.connectionFor(type) : null;
+                if (connection != null && connection.messagingVersion() < version)
+                    nodes.add(node);
+            });
+        }
+        return nodes;
+    }
 }

test/distributed/org/apache/cassandra/distributed/test/sai/ANNOptionsDistributedTest.java

@@ -16,6 +16,8 @@
 
 package org.apache.cassandra.distributed.test.sai;
 
+import java.util.concurrent.ThreadLocalRandom;
+
 import org.junit.Test;
 
 import net.bytebuddy.ByteBuddy;
@@ -118,9 +120,9 @@ public static class BB
     {
         public static void install(ClassLoader classLoader, int node)
         {
-            if (node == 1)
+            // inject randomly first or second node to make sure it works if the node is a coordinator or replica
+            if (node == ThreadLocalRandom.current().nextInt(1, 3))
             {
-                // set the current verson to DS 11, which suppors ANN options
                 new ByteBuddy().rebase(MessagingService.class)
                                .method(named("currentVersion"))
                                .intercept(MethodDelegation.to(BB.class))