From 6344f0b78a7b0cca1b1914cd943ec543e032bad3 Mon Sep 17 00:00:00 2001 From: Tim Vernum Date: Fri, 5 Jan 2018 15:14:19 +1100 Subject: [PATCH 1/3] Use the underlying connection version for CCS connections 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. --- .../elasticsearch/transport/RemoteClusterConnection.java | 6 ++++++ 1 file changed, 6 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(); + } }; } From bdb6dd50f794288da532532883e92f2316fcefc9 Mon Sep 17 00:00:00 2001 From: Tim Vernum Date: Fri, 5 Jan 2018 15:56:13 +1100 Subject: [PATCH 2/3] Add test for remote connection version --- .../RemoteClusterConnectionTests.java | 103 ++++++++++++++---- 1 file changed, 82 insertions(+), 21 deletions(-) diff --git a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java index e6d278af085b2..40ed90d1a7bd9 100644 --- a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java +++ b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java @@ -18,6 +18,27 @@ */ package org.elasticsearch.transport; +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Base64; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; + import org.apache.lucene.store.AlreadyClosedException; import org.apache.lucene.util.IOUtils; import org.elasticsearch.Build; @@ -58,30 +79,13 @@ import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; -import java.io.IOException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.ServerSocket; -import java.net.Socket; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Base64; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.BrokenBarrierException; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.CyclicBarrier; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; -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 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 MockServerSocket()) { From fdc697c2d4e087b2970b0c380ed09bbeaf1af250 Mon Sep 17 00:00:00 2001 From: Tim Vernum Date: Fri, 5 Jan 2018 16:31:12 +1100 Subject: [PATCH 3/3] Remove import churn --- .../RemoteClusterConnectionTests.java | 42 +++++++++---------- 1 file changed, 21 insertions(+), 21 deletions(-) diff --git a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java index 40ed90d1a7bd9..e7dccf702fe26 100644 --- a/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java +++ b/core/src/test/java/org/elasticsearch/transport/RemoteClusterConnectionTests.java @@ -18,27 +18,6 @@ */ package org.elasticsearch.transport; -import java.io.IOException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.ServerSocket; -import java.net.Socket; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Base64; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.BrokenBarrierException; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.CyclicBarrier; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Function; - import org.apache.lucene.store.AlreadyClosedException; import org.apache.lucene.util.IOUtils; import org.elasticsearch.Build; @@ -79,6 +58,27 @@ import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Base64; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; + import static java.util.Collections.emptyMap; import static java.util.Collections.emptySet; import static org.hamcrest.Matchers.equalTo;