Use the underlying connection version for CCS connections (#28093)

Previously this would default to the version of the remote Node.
However, if the remote cluster was mixed-version (e.g. it was part way through a
rolling upgrade), then the TransportService may have negotiated a connection
version that is not identical to connected Node's version.

This mismatch would cause the Stream and the (Remote)Connection to report
different version numbers, which could cause data to be sent over the wire
using an incorrect serialization version.

core/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java

@@ -23,6 +23,7 @@ import org.apache.logging.log4j.util.Supplier;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.util.IOUtils;
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
 import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction;
@@ -280,6 +281,11 @@ final class RemoteClusterConnection extends AbstractComponent implements Transpo
             public void close() throws IOException {
                 assert false: "proxy connections must not be closed";
             }
+
+            @Override
+            public Version getVersion() {
+                return connection.getVersion();
+            }
         };
     }
 

core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java

@@ -81,7 +81,11 @@ import java.util.function.Function;
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.emptySet;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.iterableWithSize;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.startsWith;
 
 public class RemoteClusterConnectionTests extends ESTestCase {
@@ -305,6 +309,63 @@ public class RemoteClusterConnectionTests extends ESTestCase {
         }
     }
 
+    public void testRemoteConnectionVersionMatchesTransportConnectionVersion() throws Exception {
+        List<DiscoveryNode> knownNodes = new CopyOnWriteArrayList<>();
+        final Version previousVersion = VersionUtils.getPreviousVersion();
+        try (MockTransportService seedTransport = startTransport("seed_node", knownNodes, previousVersion);
+             MockTransportService discoverableTransport = startTransport("discoverable_node", knownNodes, Version.CURRENT)) {
+
+            DiscoveryNode seedNode = seedTransport.getLocalDiscoNode();
+            assertThat(seedNode, notNullValue());
+            knownNodes.add(seedNode);
+
+            DiscoveryNode oldVersionNode = discoverableTransport.getLocalDiscoNode();
+            assertThat(oldVersionNode, notNullValue());
+            knownNodes.add(oldVersionNode);
+
+            assertThat(seedNode.getVersion(), not(equalTo(oldVersionNode.getVersion())));
+            try (MockTransportService service = MockTransportService.createNewService(Settings.EMPTY, Version.CURRENT, threadPool, null)) {
+                final Transport.Connection seedConnection = new Transport.Connection() {
+                    @Override
+                    public DiscoveryNode getNode() {
+                        return seedNode;
+                    }
+
+                    @Override
+                    public void sendRequest(long requestId, String action, TransportRequest request, TransportRequestOptions options)
+                        throws IOException, TransportException {
+                        // no-op
+                    }
+
+                    @Override
+                    public void close() throws IOException {
+                        // no-op
+                    }
+                };
+                service.addDelegate(seedNode.getAddress(), new MockTransportService.DelegateTransport(service.getOriginalTransport()) {
+                    @Override
+                    public Connection getConnection(DiscoveryNode node) {
+                        if (node == seedNode) {
+                            return seedConnection;
+                        }
+                        return super.getConnection(node);
+                    }
+                });
+                service.start();
+                service.acceptIncomingRequests();
+                try (RemoteClusterConnection connection = new RemoteClusterConnection(Settings.EMPTY, "test-cluster",
+                    Arrays.asList(seedNode), service, Integer.MAX_VALUE, n -> true)) {
+                    connection.addConnectedNode(seedNode);
+                    for (DiscoveryNode node : knownNodes) {
+                        final Transport.Connection transportConnection = connection.getConnection(node);
+                        assertThat(transportConnection.getVersion(), equalTo(previousVersion));
+                    }
+                    assertThat(knownNodes, iterableWithSize(2));
+                }
+            }
+        }
+    }
+
     @SuppressForbidden(reason = "calls getLocalHost here but it's fine in this case")
     public void testSlowNodeCanBeCanceled() throws IOException, InterruptedException {
         try (ServerSocket socket = new MockServerSocket()) {