Transport Client: When adding an address was already added, ignore it, closes #1906.

src/main/java/org/elasticsearch/client/transport/TransportClient.java

@@ -184,6 +184,13 @@ public class TransportClient extends AbstractClient {
         return nodesService.connectedNodes();
     }
 
+    /**
+     * Returns the listed nodes in the transport client (ones added to it).
+     */
+    public ImmutableList<DiscoveryNode> listedNodes() {
+        return nodesService.listedNodes();
+    }
+
     /**
      * Adds a transport address that will be used to connect to.
      * <p/>
@@ -192,7 +199,7 @@ public class TransportClient extends AbstractClient {
      * <p/>
      * <p>In order to get the list of all the current connected nodes, please see {@link #connectedNodes()}.
      */
-    public TransportClient addTransportAddress(TransportAddress transportAddress) {
+    public TransportClient addTransportAddress(TransportAddress... transportAddress) {
         nodesService.addTransportAddress(transportAddress);
         return this;
     }

src/main/java/org/elasticsearch/client/transport/TransportClientNodesService.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.client.transport;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import jsr166y.LinkedTransferQueue;
 import org.elasticsearch.ElasticSearchException;
@@ -42,6 +43,7 @@ import org.elasticsearch.transport.*;
 
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledFuture;
@@ -119,12 +121,37 @@ public class TransportClientNodesService extends AbstractComponent {
         return this.nodes;
     }
 
-    public TransportClientNodesService addTransportAddress(TransportAddress transportAddress) {
+    public ImmutableList<DiscoveryNode> listedNodes() {
+        return this.listedNodes;
+    }
+
+    public TransportClientNodesService addTransportAddress(TransportAddress... transportAddresses) {
         synchronized (transportMutex) {
+            List<TransportAddress> filtered = Lists.newArrayListWithExpectedSize(transportAddresses.length);
+            for (TransportAddress transportAddress : transportAddresses) {
+                boolean found = false;
+                for (DiscoveryNode otherNode : listedNodes) {
+                    if (otherNode.address().equals(transportAddress)) {
+                        found = true;
+                        logger.debug("address [{}] already exists with [{}], ignoring...", transportAddress, otherNode);
+                        break;
+                    }
+                }
+                if (!found) {
+                    filtered.add(transportAddress);
+                }
+            }
+            if (filtered.isEmpty()) {
+                return this;
+            }
             ImmutableList.Builder<DiscoveryNode> builder = ImmutableList.builder();
+            builder.addAll(listedNodes());
+            for (TransportAddress transportAddress : filtered) {
                 DiscoveryNode node = new DiscoveryNode("#transport#-" + tempNodeIdGenerator.incrementAndGet(), transportAddress);
-            logger.debug("adding address {}", node);
+                logger.debug("adding address [{}]", node);
-            listedNodes = builder.addAll(listedNodes).add(node).build();
+                builder.add(node);
+            }
+            listedNodes = builder.build();
         }
         nodesSampler.sample();
         return this;
@@ -137,7 +164,7 @@ public class TransportClientNodesService extends AbstractComponent {
                 if (!otherNode.address().equals(transportAddress)) {
                     builder.add(otherNode);
                 } else {
-                    logger.debug("removing address {}", otherNode);
+                    logger.debug("removing address [{}]", otherNode);
                 }
             }
             listedNodes = builder.build();

src/main/java/org/elasticsearch/transport/netty/MessageChannelHandler.java

@@ -142,7 +142,6 @@ public class MessageChannelHandler extends SimpleChannelUpstreamHandler {
         }
 
         if (cumulationBuffer) {
-            assert buffer == this.cumulation;
             if (!buffer.readable()) {
                 this.cumulation = null;
             } else if (buffer.readerIndex() > 0) {

src/main/java/org/elasticsearch/transport/netty/NettyTransport.java

@@ -139,6 +139,8 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
 
     private volatile BoundTransportAddress boundAddress;
 
+    private final Object[] connectMutex;
+
     public NettyTransport(ThreadPool threadPool) {
         this(EMPTY_SETTINGS, threadPool, new NetworkService(EMPTY_SETTINGS));
     }
@@ -153,6 +155,11 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
         this.threadPool = threadPool;
         this.networkService = networkService;
 
+        this.connectMutex = new Object[500];
+        for (int i = 0; i < connectMutex.length; i++) {
+            connectMutex[i] = new Object();
+        }
+
         this.workerCount = componentSettings.getAsInt("worker_count", Runtime.getRuntime().availableProcessors() * 2);
         this.blockingServer = settings.getAsBoolean("transport.tcp.blocking_server", settings.getAsBoolean(TCP_BLOCKING_SERVER, settings.getAsBoolean(TCP_BLOCKING, false)));
         this.blockingClient = settings.getAsBoolean("transport.tcp.blocking_client", settings.getAsBoolean(TCP_BLOCKING_CLIENT, settings.getAsBoolean(TCP_BLOCKING, false)));
@@ -489,6 +496,7 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
         if (node == null) {
             throw new ConnectTransportException(null, "Can't connect to a null node");
         }
+        synchronized (connectLock(node.id())) {
             try {
                 NodeChannels nodeChannels = connectedNodes.get(node);
                 if (nodeChannels != null) {
@@ -513,7 +521,7 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
                     nodeChannels.close();
                 } else {
                     if (logger.isDebugEnabled()) {
-                    logger.debug("Connected to node [{}]", node);
+                        logger.debug("connected to node [{}]", node);
                     }
                     transportServiceAdapter.raiseNodeConnected(node);
                 }
@@ -524,6 +532,7 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
                 throw new ConnectTransportException(node, "General node connection failure", e);
             }
         }
+    }
 
     private NodeChannels connectToChannelsLight(DiscoveryNode node) {
         InetSocketAddress address = ((InetSocketTransportAddress) node.address()).address();
@@ -620,16 +629,18 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
 
     @Override
     public void disconnectFromNode(DiscoveryNode node) {
+        synchronized (connectLock(node.id())) {
             NodeChannels nodeChannels = connectedNodes.remove(node);
             if (nodeChannels != null) {
                 try {
                     nodeChannels.close();
                 } finally {
-                logger.debug("Disconnected from [{}]", node);
+                    logger.debug("disconnected from [{}]", node);
                     transportServiceAdapter.raiseNodeDisconnected(node);
                 }
             }
         }
+    }
 
     private Channel nodeChannel(DiscoveryNode node, TransportRequestOptions options) throws ConnectTransportException {
         NodeChannels nodeChannels = connectedNodes.get(node);
@@ -639,6 +650,15 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
         return nodeChannels.channel(options.type());
     }
 
+    private Object connectLock(String nodeId) {
+        int hash = nodeId.hashCode();
+        // abs returns Integer.MIN_VALUE, so we need to protect against it...
+        if (hash == Integer.MIN_VALUE) {
+            hash = 0;
+        }
+        return connectMutex[Math.abs(hash) % connectMutex.length];
+    }
+
     private class ChannelCloseListener implements ChannelFutureListener {
 
         private final DiscoveryNode node;