Remove client connections from TcpTransport

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java

@@ -267,8 +267,12 @@ protected Netty4TcpServerChannel bind(String name, InetSocketAddress address) {
         return esChannel;
     }
 
-    ScheduledPing getPing() {
-        return scheduledPing;
+    long successfulPingCount() {
+        return successfulPings.count();
+    }
+
+    long failedPingCount() {
+        return failedPings.count();
     }
 
     @Override

modules/transport-netty4/src/test/java/org/elasticsearch/transport/netty4/Netty4ScheduledPingTests.java

@@ -26,16 +26,13 @@
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
-import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TcpTransport;
-import org.elasticsearch.transport.TransportChannel;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportRequest;
-import org.elasticsearch.transport.TransportRequestHandler;
 import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportResponseHandler;
@@ -83,22 +80,19 @@ public void testScheduledPing() throws Exception {
         serviceB.connectToNode(nodeA);
 
         assertBusy(() -> {
-            assertThat(nettyA.getPing().getSuccessfulPings(), greaterThan(100L));
-            assertThat(nettyB.getPing().getSuccessfulPings(), greaterThan(100L));
+            assertThat(nettyA.successfulPingCount(), greaterThan(100L));
+            assertThat(nettyB.successfulPingCount(), greaterThan(100L));
         });
-        assertThat(nettyA.getPing().getFailedPings(), equalTo(0L));
-        assertThat(nettyB.getPing().getFailedPings(), equalTo(0L));
+        assertThat(nettyA.failedPingCount(), equalTo(0L));
+        assertThat(nettyB.failedPingCount(), equalTo(0L));
 
         serviceA.registerRequestHandler("internal:sayHello", TransportRequest.Empty::new, ThreadPool.Names.GENERIC,
-            new TransportRequestHandler<TransportRequest.Empty>() {
-                @Override
-                public void messageReceived(TransportRequest.Empty request, TransportChannel channel, Task task) {
-                    try {
-                        channel.sendResponse(TransportResponse.Empty.INSTANCE, TransportResponseOptions.EMPTY);
-                    } catch (IOException e) {
-                        logger.error("Unexpected failure", e);
-                        fail(e.getMessage());
-                    }
+            (request, channel, task) -> {
+                try {
+                    channel.sendResponse(TransportResponse.Empty.INSTANCE, TransportResponseOptions.EMPTY);
+                } catch (IOException e) {
+                    logger.error("Unexpected failure", e);
+                    fail(e.getMessage());
                 }
             });
 
@@ -130,11 +124,11 @@ public void handleException(TransportException exp) {
         }
 
         assertBusy(() -> {
-            assertThat(nettyA.getPing().getSuccessfulPings(), greaterThan(200L));
-            assertThat(nettyB.getPing().getSuccessfulPings(), greaterThan(200L));
+            assertThat(nettyA.successfulPingCount(), greaterThan(200L));
+            assertThat(nettyB.successfulPingCount(), greaterThan(200L));
         });
-        assertThat(nettyA.getPing().getFailedPings(), equalTo(0L));
-        assertThat(nettyB.getPing().getFailedPings(), equalTo(0L));
+        assertThat(nettyA.failedPingCount(), equalTo(0L));
+        assertThat(nettyB.failedPingCount(), equalTo(0L));
 
         Releasables.close(serviceA, serviceB);
         terminate(threadPool);

server/src/main/java/org/elasticsearch/cluster/NodeConnectionsService.java

@@ -101,7 +101,7 @@ public void onFailure(Exception e) {
                     }
 
                     @Override
-                    protected void doRun() throws Exception {
+                    protected void doRun() {
                         try (Releasable ignored = nodeLocks.acquire(node)) {
                             validateAndConnectIfNeeded(node);
                         }

server/src/main/java/org/elasticsearch/transport/CloseableConnection.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.transport;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.concurrent.CompletableContext;
+
+
+/**
+ * Abstract Transport.Connection that provides common close logic.
+ */
+public abstract class CloseableConnection implements Transport.Connection {
+
+    private final CompletableContext<Void> closeContext = new CompletableContext<>();
+
+    @Override
+    public void addCloseListener(ActionListener<Void> listener) {
+        closeContext.addListener(ActionListener.toBiConsumer(listener));
+    }
+
+    @Override
+    public boolean isClosed() {
+        return closeContext.isDone();
+    }
+
+    @Override
+    public void close() {
+        closeContext.complete(null);
+    }
+}