Remove LocalTransport in favor of MockTcpTransport (#20695)

This change proposes the removal of all non-tcp transport implementations. The
mock transport can be used by default to run tests instead of local transport that has
roughly the same performance compared to TCP or at least not noticeably slower.

This is a master only change, deprecation notice in 5.x will be committed as a
separate change.

Author: s1monw

TESTING.asciidoc

@@ -16,22 +16,6 @@ following:
 gradle assemble
 -----------------------------
 
-== Other test options
-
-To disable and enable network transport, set the `tests.es.node.mode` system property.
-
-Use network transport:
-
-------------------------------------
--Dtests.es.node.mode=network
-------------------------------------
-
-Use local transport (default since 1.3):
-
--------------------------------------
--Dtests.es.node.mode=local
--------------------------------------
-
 === Running Elasticsearch from a checkout
 
 In order to run Elasticsearch from source without building a package, you can

benchmarks/src/main/java/org/elasticsearch/benchmark/routing/allocation/Allocators.java

@@ -31,15 +31,18 @@
 import org.elasticsearch.cluster.routing.allocation.decider.AllocationDeciders;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.LocalTransportAddress;
+import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.gateway.GatewayAllocator;
 
 import java.lang.reflect.InvocationTargetException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public final class Allocators {
     private static class NoopGatewayAllocator extends GatewayAllocator {
@@ -91,8 +94,11 @@ public static AllocationDeciders defaultAllocationDeciders(Settings settings, Cl
 
     }
 
+    private static final AtomicInteger portGenerator = new AtomicInteger();
+
     public static DiscoveryNode newNode(String nodeId, Map<String, String> attributes) {
-        return new DiscoveryNode("", nodeId, LocalTransportAddress.buildUnique(), attributes, Sets.newHashSet(DiscoveryNode.Role.MASTER,
+        return new DiscoveryNode("", nodeId, new TransportAddress(TransportAddress.META_ADDRESS,
+            portGenerator.incrementAndGet()), attributes, Sets.newHashSet(DiscoveryNode.Role.MASTER,
             DiscoveryNode.Role.DATA), Version.CURRENT);
     }
 }

client/benchmark/src/main/java/org/elasticsearch/client/benchmark/transport/TransportClientBenchmark.java

@@ -27,7 +27,7 @@
 import org.elasticsearch.client.benchmark.ops.search.SearchRequestExecutor;
 import org.elasticsearch.client.transport.TransportClient;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.plugin.noop.NoopPlugin;
 import org.elasticsearch.plugin.noop.action.bulk.NoopBulkAction;
@@ -51,7 +51,7 @@ public static void main(String[] args) throws Exception {
     @Override
     protected TransportClient client(String benchmarkTargetHost) throws Exception {
         TransportClient client = new PreBuiltTransportClient(Settings.EMPTY, NoopPlugin.class);
-        client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(benchmarkTargetHost), 9300));
+        client.addTransportAddress(new TransportAddress(InetAddress.getByName(benchmarkTargetHost), 9300));
         return client;
     }
 

core/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodes.java

@@ -27,7 +27,6 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.InetSocketTransportAddress;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
@@ -75,10 +74,7 @@ public class ClusterStatsNodes implements ToXContent {
 
             // now do the stats that should be deduped by hardware (implemented by ip deduping)
             TransportAddress publishAddress = nodeResponse.nodeInfo().getTransport().address().publishAddress();
-            InetAddress inetAddress = null;
-            if (publishAddress.uniqueAddressTypeId() == 1) {
-                inetAddress = ((InetSocketTransportAddress) publishAddress).address().getAddress();
-            }
+            final InetAddress inetAddress = publishAddress.address().getAddress();
             if (!seenAddresses.add(inetAddress)) {
                 continue;
             }

core/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java

@@ -26,7 +26,6 @@
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
-import org.elasticsearch.common.transport.TransportAddressSerializers;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.node.Node;
@@ -40,7 +39,6 @@
 import java.util.Set;
 import java.util.function.Predicate;
 
-import static org.elasticsearch.common.transport.TransportAddressSerializers.addressToStream;
 
 /**
  * A discovery node represents a node that is part of the cluster.
@@ -217,7 +215,7 @@ public DiscoveryNode(StreamInput in) throws IOException {
         this.ephemeralId = in.readString().intern();
         this.hostName = in.readString().intern();
         this.hostAddress = in.readString().intern();
-        this.address = TransportAddressSerializers.addressFromStream(in);
+        this.address = new TransportAddress(in);
         int size = in.readVInt();
         this.attributes = new HashMap<>(size);
         for (int i = 0; i < size; i++) {
@@ -242,7 +240,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(ephemeralId);
         out.writeString(hostName);
         out.writeString(hostAddress);
-        addressToStream(out, address);
+        address.writeTo(out);
         out.writeVInt(attributes.size());
         for (Map.Entry<String, String> entry : attributes.entrySet()) {
             out.writeString(entry.getKey());

core/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeFilters.java

@@ -24,7 +24,7 @@
 import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.transport.TransportAddress;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -82,8 +82,8 @@ public boolean match(DiscoveryNode node) {
             if ("_ip".equals(attr)) {
                 // We check both the host_ip or the publish_ip
                 String publishAddress = null;
-                if (node.getAddress() instanceof InetSocketTransportAddress) {
-                    publishAddress = NetworkAddress.format(((InetSocketTransportAddress) node.getAddress()).address().getAddress());
+                if (node.getAddress() instanceof TransportAddress) {
+                    publishAddress = NetworkAddress.format(node.getAddress().address().getAddress());
                 }
 
                 boolean match = matchByIP(values, node.getHostAddress(), publishAddress);
@@ -116,8 +116,8 @@ public boolean match(DiscoveryNode node) {
             } else if ("_publish_ip".equals(attr)) {
                 // We check explicitly only the publish_ip
                 String address = null;
-                if (node.getAddress() instanceof InetSocketTransportAddress) {
-                    address = NetworkAddress.format(((InetSocketTransportAddress) node.getAddress()).address().getAddress());
+                if (node.getAddress() instanceof TransportAddress) {
+                    address = NetworkAddress.format(node.getAddress().address().getAddress());
                 }
 
                 boolean match = matchByIP(values, address, null);

core/src/main/java/org/elasticsearch/common/network/NetworkModule.java

@@ -45,10 +45,8 @@
 import org.elasticsearch.transport.TransportInterceptor;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportRequestHandler;
-import org.elasticsearch.transport.local.LocalTransport;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -112,7 +110,6 @@ public NetworkModule(Settings settings, boolean transportClient, List<NetworkPlu
                          NetworkService networkService) {
         this.settings = settings;
         this.transportClient = transportClient;
-        registerTransport(LOCAL_TRANSPORT, () -> new LocalTransport(settings, threadPool, namedWriteableRegistry, circuitBreakerService));
         for (NetworkPlugin plugin : plugins) {
             if (transportClient == false && HTTP_ENABLED.get(settings)) {
                 Map<String, Supplier<HttpServerTransport>> httpTransportFactory = plugin.getHttpTransports(settings, threadPool, bigArrays,
@@ -126,7 +123,7 @@ public NetworkModule(Settings settings, boolean transportClient, List<NetworkPlu
             for (Map.Entry<String, Supplier<Transport>> entry : httpTransportFactory.entrySet()) {
                 registerTransport(entry.getKey(), entry.getValue());
             }
-            List<TransportInterceptor> transportInterceptors = plugin.getTransportInterceptors();
+            List<TransportInterceptor> transportInterceptors = plugin.getTransportInterceptors(namedWriteableRegistry);
             for (TransportInterceptor interceptor : transportInterceptors) {
                 registerTransportInterceptor(interceptor);
             }

core/src/main/java/org/elasticsearch/common/transport/BoundTransportAddress.java

@@ -68,18 +68,18 @@ public void readFrom(StreamInput in) throws IOException {
         int boundAddressLength = in.readInt();
         boundAddresses = new TransportAddress[boundAddressLength];
         for (int i = 0; i < boundAddressLength; i++) {
-            boundAddresses[i] = TransportAddressSerializers.addressFromStream(in);
+            boundAddresses[i] = new TransportAddress(in);
         }
-        publishAddress = TransportAddressSerializers.addressFromStream(in);
+        publishAddress = new TransportAddress(in);
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeInt(boundAddresses.length);
         for (TransportAddress address : boundAddresses) {
-            TransportAddressSerializers.addressToStream(out, address);
+            address.writeTo(out);
         }
-        TransportAddressSerializers.addressToStream(out, publishAddress);
+        publishAddress.writeTo(out);
     }
 
     @Override