From db186311c65227b721ad49f275d5b26bdf23986f Mon Sep 17 00:00:00 2001
From: Tim Vernum <tim@adjective.org>
Date: Mon, 8 Jan 2018 11:11:41 +1000
Subject: [PATCH] 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.
---
 .../transport/RemoteClusterConnection.java    |  6 ++
 .../RemoteClusterConnectionTests.java         | 61 +++++++++++++++++++
 2 files changed, 67 insertions(+)

diff --git a/core/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java b/core/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java
index 55edd0c86ec29..e73debc601430 100644
--- a/core/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java
+++ b/core/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java
@@ -23,6 +23,7 @@
 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 @@ public void sendRequest(long requestId, String action, TransportRequest request,
             public void close() throws IOException {
                 assert false: "proxy connections must not be closed";
             }
+
+            @Override
+            public Version getVersion() {
+                return connection.getVersion();
+            }
         };
     }
 
diff --git a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java
index e6d278af085b2..e7dccf702fe26 100644
--- a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java
+++ b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java
@@ -81,7 +81,11 @@
 
 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 void testConnectWithIncompatibleTransports() throws Exception {
         }
     }
 
+    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()) {