From db186311c65227b721ad49f275d5b26bdf23986f 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 | 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 55edd0c86ec..e73debc6014 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.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(); + } }; } diff --git a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java index e6d278af085..e7dccf702fe 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 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 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()) {