From 52fd64b949d5bff2f7850cba11f75b73e2982fc0 Mon Sep 17 00:00:00 2001 From: Tim Vernum 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 | 64 +++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/core/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java b/core/src/main/java/org/elasticsearch/transport/RemoteClusterConnection.java index c08bf9b737e95..0006d36c0c362 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; @@ -270,6 +271,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 599f1dccda0aa..4d226adde3e66 100644 --- a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java +++ b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java @@ -54,6 +54,7 @@ import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.http.HttpInfo; import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.test.VersionUtils; import org.elasticsearch.test.transport.MockTransportService; import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; @@ -81,6 +82,12 @@ 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 { @@ -303,6 +310,63 @@ public void testConnectWithIncompatibleTransports() throws Exception { } } + public void testRemoteConnectionVersionMatchesTransportConnectionVersion() throws Exception { + List 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 ServerSocket()) {