Addressed next round of feedback:

- updated ConnectionChangeListener constructor and moved registration to service.doStart()
- renamed DisconnectionHistory record and ConnectionChangeListener
- fixed up DiscoveryNode vs. Transport.Connection.getNode() confusion (these are the same)
- fixed log formatting
- edited TransportConnectionListener interface to take a nullable closeException instead of the Transport.Connection
- completed test of DisconnectionHistory, at init, post-connection, post-disconnection, and post-reconnection
- moved docs for ConnectionChangeListener, and added docs for DisconnectionHistory

Author: schase-es

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

@@ -83,14 +83,12 @@ public class NodeConnectionsService extends AbstractLifecycleComponent {
 
     private final TimeValue reconnectInterval;
     private volatile ConnectionChecker connectionChecker;
-    private final ConnectionHistoryListener connectionHistoryListener;
 
     @Inject
     public NodeConnectionsService(Settings settings, ThreadPool threadPool, TransportService transportService) {
         this.threadPool = threadPool;
         this.transportService = transportService;
         this.reconnectInterval = NodeConnectionsService.CLUSTER_NODE_RECONNECT_INTERVAL_SETTING.get(settings);
-        this.connectionHistoryListener = new ConnectionHistoryListener();
     }
 
     /**
@@ -194,6 +192,7 @@ public String toString() {
 
     @Override
     protected void doStart() {
+        transportService.addConnectionListener(new ConnectionChangeListener());
         final ConnectionChecker connectionChecker = new ConnectionChecker();
         this.connectionChecker = connectionChecker;
         connectionChecker.scheduleNextCheck();
@@ -215,16 +214,35 @@ public void reconnectToNodes(DiscoveryNodes discoveryNodes, Runnable onCompletio
         });
     }
 
-    record ConnectionHistory(String ephemeralId, long disconnectTime, Exception disconnectCause) {}
+    // exposed for testing
+    protected ConnectionTarget connectionTargetForNode(DiscoveryNode node) {
+        synchronized (mutex) {
+            return targetsByNode.get(node);
+        }
+    }
+
+    /**
+     * Time of disconnect in absolute time ({@link ThreadPool#absoluteTimeInMillis()}),
+     * and disconnect-causing exception, if any
+     */
+    record DisconnectionHistory(long disconnectTimeMillis, @Nullable Exception disconnectCause) {
+        public long getDisconnectTimeMillis() {
+            return disconnectTimeMillis;
+        }
 
-    private class ConnectionTarget {
+        public Exception getDisconnectCause() {
+            return disconnectCause;
+        }
+    }
+
+    protected class ConnectionTarget {
         private final DiscoveryNode discoveryNode;
 
         private final AtomicInteger consecutiveFailureCount = new AtomicInteger();
         private final AtomicReference<Releasable> connectionRef = new AtomicReference<>();
 
         // access is synchronized by the service mutex
-        protected ConnectionHistory connectionHistory = null;
+        protected DisconnectionHistory disconnectionHistory = null;
 
         // all access to these fields is synchronized
         private List<Releasable> pendingRefs;
@@ -359,43 +377,41 @@ public String toString() {
         }
     }
 
-    private class ConnectionHistoryListener implements TransportConnectionListener {
-        /**
-         * Receives connection/disconnection events from the transport, and records in per-node ConnectionHistory
-         * structures for logging network issues. ConnectionHistory records are stored in ConnectionTargets.
-         *
-         * Network issues (that this listener monitors for) occur whenever a reconnection to a node succeeds,
-         * and it has the same ephemeral Id as it did during the last connection.
-         */
-        ConnectionHistoryListener() {
-            transportService.addConnectionListener(ConnectionHistoryListener.this);
-        }
-
+    /**
+     * Receives connection/disconnection events from the transport, and records them in per-node DisconnectionHistory
+     * structures for logging network issues. DisconnectionHistory records are stored their node's ConnectionTarget.
+     *
+     * Network issues (that this listener monitors for) occur whenever a reconnection to a node succeeds,
+     * and it has the same ephemeral ID as it did during the last connection; this happens when a connection event
+     * occurs, and its ConnectionTarget entry has a previous DisconnectionHistory stored.
+     */
+    private class ConnectionChangeListener implements TransportConnectionListener {
         @Override
         public void onNodeConnected(DiscoveryNode node, Transport.Connection connection) {
-            ConnectionHistory connectionHistory = null;
+            DisconnectionHistory disconnectionHistory = null;
             synchronized (mutex) {
                 ConnectionTarget connectionTarget = targetsByNode.get(node);
                 if (connectionTarget != null) {
-                    connectionHistory = connectionTarget.connectionHistory;
-                    connectionTarget.connectionHistory = null;
+                    disconnectionHistory = connectionTarget.disconnectionHistory;
+                    connectionTarget.disconnectionHistory = null;
                 }
             }
 
-            if (connectionHistory != null && connectionHistory.ephemeralId.equals(connection.getNode().getEphemeralId())) {
-                long millisSinceDisconnect = threadPool.absoluteTimeInMillis() - connectionHistory.disconnectTime;
-                if (connectionHistory.disconnectCause != null) {
+            if (disconnectionHistory != null) {
+                long millisSinceDisconnect = threadPool.absoluteTimeInMillis() - disconnectionHistory.disconnectTimeMillis;
+                if (disconnectionHistory.disconnectCause != null) {
                     logger.warn(
                         () -> format(
-                            "reopened transport connection to node [%s] "
-                                + "which disconnected exceptionally [%dms] ago but did not "
-                                + "restart, so the disconnection is unexpected; "
-                                + "see [%s] for troubleshooting guidance",
-                            connection.getNode().descriptionWithoutAttributes(),
+                            """
+                                reopened transport connection to node [%s] \
+                                which disconnected exceptionally [%dms] ago but did not \
+                                restart, so the disconnection is unexpected; \
+                                see [%s] for troubleshooting guidance""",
+                            node.descriptionWithoutAttributes(),
                             millisSinceDisconnect,
                             ReferenceDocs.NETWORK_DISCONNECT_TROUBLESHOOTING
                         ),
-                        connectionHistory.disconnectCause
+                        disconnectionHistory.disconnectCause
                     );
                 } else {
                     logger.warn(
@@ -404,7 +420,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
                             which disconnected gracefully [{}ms] ago but did not \
                             restart, so the disconnection is unexpected; \
                             see [{}] for troubleshooting guidance""",
-                        connection.getNode().descriptionWithoutAttributes(),
+                        node.descriptionWithoutAttributes(),
                         millisSinceDisconnect,
                         ReferenceDocs.NETWORK_DISCONNECT_TROUBLESHOOTING
                     );
@@ -413,16 +429,12 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
         }
 
         @Override
-        public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
-            ConnectionHistory connectionHistory = new ConnectionHistory(
-                connection.getNode().getEphemeralId(),
-                threadPool.absoluteTimeInMillis(),
-                closeException
-            );
+        public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
+            DisconnectionHistory disconnectionHistory = new DisconnectionHistory(threadPool.absoluteTimeInMillis(), closeException);
             synchronized (mutex) {
                 ConnectionTarget connectionTarget = targetsByNode.get(node);
                 if (connectionTarget != null) {
-                    connectionTarget.connectionHistory = connectionHistory;
+                    connectionTarget.disconnectionHistory = disconnectionHistory;
                 }
             }
         }

server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java

@@ -34,7 +34,6 @@
 import org.elasticsearch.transport.AbstractTransportRequest;
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.ReceiveTimeoutTransportException;
-import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportConnectionListener;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportRequestOptions;
@@ -138,7 +137,7 @@ public FollowersChecker(
         );
         transportService.addConnectionListener(new TransportConnectionListener() {
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 handleDisconnectedNode(node);
             }
         });

server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java

@@ -32,7 +32,6 @@
 import org.elasticsearch.transport.ConnectTransportException;
 import org.elasticsearch.transport.NodeDisconnectedException;
 import org.elasticsearch.transport.ReceiveTimeoutTransportException;
-import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportConnectionListener;
 import org.elasticsearch.transport.TransportException;
 import org.elasticsearch.transport.TransportRequestOptions;
@@ -124,7 +123,7 @@ public class LeaderChecker {
 
         transportService.addConnectionListener(new TransportConnectionListener() {
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 handleDisconnectedNode(node);
             }
         });

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

@@ -242,7 +242,7 @@ public void onFailure(Exception e) {
 
                                     void handleClose(@Nullable Exception e) {
                                         connectedNodes.remove(node, conn);
-                                        connectionListener.onNodeDisconnected(node, conn, e);
+                                        connectionListener.onNodeDisconnected(node, e);
                                         managerRefs.decRef();
                                     }
                                 });

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

@@ -60,9 +60,9 @@ final class DelegatingNodeConnectionListener implements TransportConnectionListe
         private final CopyOnWriteArrayList<TransportConnectionListener> listeners = new CopyOnWriteArrayList<>();
 
         @Override
-        public void onNodeDisconnected(DiscoveryNode key, Transport.Connection connection, @Nullable Exception closeException) {
+        public void onNodeDisconnected(DiscoveryNode key, @Nullable Exception closeException) {
             for (TransportConnectionListener listener : listeners) {
-                listener.onNodeDisconnected(key, connection, closeException);
+                listener.onNodeDisconnected(key, closeException);
             }
         }
 

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

@@ -61,7 +61,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
             }
 
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 removeConnectedNode(node);
             }
         });

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

@@ -340,7 +340,7 @@ boolean shouldRebuildConnection(Settings newSettings) {
     protected abstract ConnectionStrategy strategyType();
 
     @Override
-    public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
+    public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
         if (shouldOpenMoreConnections()) {
             // try to reconnect and fill up the slot of the disconnected node
             connect(

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

@@ -39,5 +39,5 @@ default void onNodeConnected(DiscoveryNode node, Transport.Connection connection
     /**
      * Called once a node connection is closed and unregistered.
      */
-    default void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {}
+    default void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {}
 }

server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java

@@ -832,7 +832,7 @@ public void testCCSRemoteReduceWithDisconnectedRemoteClusters() throws Exception
             CountDownLatch disconnectedLatch = new CountDownLatch(numDisconnectedClusters);
             RemoteClusterServiceTests.addConnectionListener(remoteClusterService, new TransportConnectionListener() {
                 @Override
-                public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
+                public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                     if (disconnectedNodes.remove(node)) {
                         disconnectedLatch.countDown();
                     }
@@ -1135,7 +1135,7 @@ public void testCollectSearchShards() throws Exception {
             CountDownLatch disconnectedLatch = new CountDownLatch(numDisconnectedClusters);
             RemoteClusterServiceTests.addConnectionListener(remoteClusterService, new TransportConnectionListener() {
                 @Override
-                public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
+                public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                     if (disconnectedNodes.remove(node)) {
                         disconnectedLatch.countDown();
                     }

server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.SubscribableListener;
+import org.elasticsearch.cluster.NodeConnectionsService.ConnectionTarget;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
@@ -49,6 +50,7 @@
 import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.TransportStats;
+import org.hamcrest.Matchers;
 import org.junit.After;
 import org.junit.Before;
 
@@ -247,36 +249,38 @@ public String toString() {
         assertConnectedExactlyToNodes(transportService, targetNodes);
     }
 
-    public void testConnectionHistoryLogging() {
+    public void testDisconnectionHistory() {
         final Settings.Builder settings = Settings.builder();
-        settings.put(CLUSTER_NODE_RECONNECT_INTERVAL_SETTING.getKey(), "10ms");
+        settings.put(CLUSTER_NODE_RECONNECT_INTERVAL_SETTING.getKey(), "100ms");
 
         final DeterministicTaskQueue deterministicTaskQueue = new DeterministicTaskQueue();
+        final ThreadPool threadPool = deterministicTaskQueue.getThreadPool();
 
         MockTransport transport = new MockTransport(deterministicTaskQueue.getThreadPool());
         TestTransportService transportService = new TestTransportService(transport, deterministicTaskQueue.getThreadPool());
         transportService.start();
         transportService.acceptIncomingRequests();
 
-        final NodeConnectionsService service = new NodeConnectionsService(
-            settings.build(),
-            deterministicTaskQueue.getThreadPool(),
-            transportService
-        );
+        final NodeConnectionsService service = new NodeConnectionsService(settings.build(), threadPool, transportService);
         service.start();
 
+        final DiscoveryNode noClose = DiscoveryNodeUtils.create("noClose");
         final DiscoveryNode gracefulClose = DiscoveryNodeUtils.create("gracefulClose");
         final DiscoveryNode exceptionalClose = DiscoveryNodeUtils.create("exceptionalClose");
 
         nodeCloseExceptions.put(exceptionalClose, new RuntimeException());
 
         final AtomicBoolean connectionCompleted = new AtomicBoolean();
-        DiscoveryNodes nodes = DiscoveryNodes.builder().add(gracefulClose).add(exceptionalClose).build();
+        DiscoveryNodes nodes = DiscoveryNodes.builder().add(noClose).add(gracefulClose).add(exceptionalClose).build();
 
         service.connectToNodes(nodes, () -> connectionCompleted.set(true));
         deterministicTaskQueue.runAllRunnableTasks();
         assertTrue(connectionCompleted.get());
 
+        assertNullDisconnectionHistory(service, noClose);
+        assertNullDisconnectionHistory(service, gracefulClose);
+        assertNullDisconnectionHistory(service, exceptionalClose);
+
         try (var mockLog = MockLog.capture(NodeConnectionsService.class)) {
             mockLog.addExpectation(
                 new MockLog.SeenEventExpectation(
@@ -305,19 +309,51 @@ public void testConnectionHistoryLogging() {
             transportService.disconnectFromNode(gracefulClose);
             transportService.disconnectFromNode(exceptionalClose);
 
-            runTasksUntil(deterministicTaskQueue, 1000);
+            // check disconnection history set after close
+            assertNullDisconnectionHistory(service, noClose);
+            assertDisconnectionHistoryDetails(service, threadPool, gracefulClose, null);
+            assertDisconnectionHistoryDetails(service, threadPool, exceptionalClose, RuntimeException.class);
+
+            runTasksUntil(deterministicTaskQueue, 200);
+
+            // check on reconnect -- disconnection history is reset
+            assertNullDisconnectionHistory(service, noClose);
+            assertNullDisconnectionHistory(service, gracefulClose);
+            assertNullDisconnectionHistory(service, exceptionalClose);
 
             mockLog.assertAllExpectationsMatched();
         }
     }
 
+    private void assertNullDisconnectionHistory(NodeConnectionsService service, DiscoveryNode node) {
+        ConnectionTarget nodeTarget = service.connectionTargetForNode(node);
+        assertNull(nodeTarget.disconnectionHistory);
+    }
+
+    private void assertDisconnectionHistoryDetails(
+        NodeConnectionsService service,
+        ThreadPool threadPool,
+        DiscoveryNode node,
+        @Nullable Class<?> disconnectCauseClass
+    ) {
+        ConnectionTarget nodeTarget = service.connectionTargetForNode(node);
+        assertNotNull(nodeTarget.disconnectionHistory);
+        assertTrue(threadPool.absoluteTimeInMillis() - nodeTarget.disconnectionHistory.getDisconnectTimeMillis() >= 0);
+        assertTrue(threadPool.absoluteTimeInMillis() - nodeTarget.disconnectionHistory.getDisconnectTimeMillis() <= 200);
+        if (disconnectCauseClass != null) {
+            assertThat(nodeTarget.disconnectionHistory.getDisconnectCause(), Matchers.isA(disconnectCauseClass));
+        } else {
+            assertNull(nodeTarget.disconnectionHistory.getDisconnectCause());
+        }
+    }
+
     public void testOnlyBlocksOnConnectionsToNewNodes() throws Exception {
         final NodeConnectionsService service = new NodeConnectionsService(Settings.EMPTY, threadPool, transportService);
 
         final AtomicReference<ActionListener<DiscoveryNode>> disconnectListenerRef = new AtomicReference<>();
         transportService.addConnectionListener(new TransportConnectionListener() {
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection, @Nullable Exception closeException) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 final ActionListener<DiscoveryNode> disconnectListener = disconnectListenerRef.getAndSet(null);
                 if (disconnectListener != null) {
                     disconnectListener.onResponse(node);