Merge branch 'main' into snity_block

dnhatn · web-flow · commit 204810f28616 · 2025-06-05T17:08:42.000-07:00
diff --git a/server/src/main/java/org/elasticsearch/TransportVersions.java b/server/src/main/java/org/elasticsearch/TransportVersions.java
@@ -189,7 +189,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion DATA_STREAM_OPTIONS_API_REMOVE_INCLUDE_DEFAULTS_8_19 = def(8_841_0_41);
     public static final TransportVersion JOIN_ON_ALIASES_8_19 = def(8_841_0_42);
     public static final TransportVersion ILM_ADD_SKIP_SETTING_8_19 = def(8_841_0_43);
-    public static final TransportVersion ML_INFERENCE_MISTRAL_CHAT_COMPLETION_ADDED_8_19 = def(8_841_0_44);
+    public static final TransportVersion ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY_8_19 = def(8_841_0_44);
     public static final TransportVersion V_9_0_0 = def(9_000_0_09);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_1 = def(9_000_0_10);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_2 = def(9_000_0_11);
diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java b/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java
@@ -26,14 +26,14 @@
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.core.CheckedRunnable;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.monitor.NodeHealthService;
 import org.elasticsearch.monitor.StatusInfo;
 import org.elasticsearch.threadpool.ThreadPool.Names;
 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;
@@ -137,7 +137,7 @@ public FollowersChecker(
         );
         transportService.addConnectionListener(new TransportConnectionListener() {
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 handleDisconnectedNode(node);
             }
         });
diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java b/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) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 handleDisconnectedNode(node);
             }
         });
diff --git a/server/src/main/java/org/elasticsearch/transport/ClusterConnectionManager.java b/server/src/main/java/org/elasticsearch/transport/ClusterConnectionManager.java
@@ -229,11 +229,26 @@ private void connectToNodeOrRetry(
                             try {
                                 connectionListener.onNodeConnected(node, conn);
                             } finally {
-                                conn.addCloseListener(ActionListener.running(() -> {
-                                    connectedNodes.remove(node, conn);
-                                    connectionListener.onNodeDisconnected(node, conn);
-                                    managerRefs.decRef();
-                                }));
+                                conn.addCloseListener(new ActionListener<Void>() {
+                                    @Override
+                                    public void onResponse(Void ignored) {
+                                        handleClose(null);
+                                    }
+
+                                    @Override
+                                    public void onFailure(Exception e) {
+                                        handleClose(e);
+                                    }
+
+                                    void handleClose(@Nullable Exception e) {
+                                        connectedNodes.remove(node, conn);
+                                        try {
+                                            connectionListener.onNodeDisconnected(node, e);
+                                        } finally {
+                                            managerRefs.decRef();
+                                        }
+                                    }
+                                });
 
                                 conn.addCloseListener(ActionListener.running(() -> {
                                     if (connectingRefCounter.hasReferences() == false) {
diff --git a/server/src/main/java/org/elasticsearch/transport/ConnectionManager.java b/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) {
+        public void onNodeDisconnected(DiscoveryNode key, @Nullable Exception closeException) {
             for (TransportConnectionListener listener : listeners) {
-                listener.onNodeDisconnected(key, connection);
+                listener.onNodeDisconnected(key, closeException);
             }
         }
 
diff --git a/server/src/main/java/org/elasticsearch/transport/RemoteConnectionManager.java b/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) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 removeConnectedNode(node);
             }
         });
diff --git a/server/src/main/java/org/elasticsearch/transport/RemoteConnectionStrategy.java b/server/src/main/java/org/elasticsearch/transport/RemoteConnectionStrategy.java
@@ -21,6 +21,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.threadpool.ThreadPool;
 
@@ -339,7 +340,7 @@ boolean shouldRebuildConnection(Settings newSettings) {
     protected abstract ConnectionStrategy strategyType();
 
     @Override
-    public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+    public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
         if (shouldOpenMoreConnections()) {
             // try to reconnect and fill up the slot of the disconnected node
             connect(
diff --git a/server/src/main/java/org/elasticsearch/transport/TransportConnectionListener.java b/server/src/main/java/org/elasticsearch/transport/TransportConnectionListener.java
@@ -10,6 +10,7 @@
 package org.elasticsearch.transport;
 
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.core.Nullable;
 
 /**
  * A listener interface that allows to react on transport events. All methods may be
@@ -38,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) {}
+    default void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {}
 }
diff --git a/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java b/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java
@@ -52,6 +52,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
@@ -831,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) {
+                public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                     if (disconnectedNodes.remove(node)) {
                         disconnectedLatch.countDown();
                     }
@@ -1134,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) {
+                public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                     if (disconnectedNodes.remove(node)) {
                         disconnectedLatch.countDown();
                     }
diff --git a/server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/NodeConnectionsServiceTests.java
@@ -30,6 +30,7 @@
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
 import org.elasticsearch.core.AbstractRefCounted;
 import org.elasticsearch.core.CheckedRunnable;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.ESTestCase;
@@ -251,7 +252,7 @@ public void testOnlyBlocksOnConnectionsToNewNodes() throws Exception {
         final AtomicReference<ActionListener<DiscoveryNode>> disconnectListenerRef = new AtomicReference<>();
         transportService.addConnectionListener(new TransportConnectionListener() {
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 final ActionListener<DiscoveryNode> disconnectListener = disconnectListenerRef.getAndSet(null);
                 if (disconnectListener != null) {
                     disconnectListener.onResponse(node);
diff --git a/server/src/test/java/org/elasticsearch/transport/ClusterConnectionManagerTests.java b/server/src/test/java/org/elasticsearch/transport/ClusterConnectionManagerTests.java
@@ -23,6 +23,7 @@
 import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.core.AbstractRefCounted;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.TimeValue;
@@ -101,7 +102,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
             }
 
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 nodeDisconnectedCount.incrementAndGet();
             }
         });
@@ -658,7 +659,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
             }
 
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 nodeDisconnectedCount.incrementAndGet();
             }
         });
@@ -698,7 +699,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
             }
 
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 nodeDisconnectedCount.incrementAndGet();
             }
         });
diff --git a/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java b/test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java
@@ -202,7 +202,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
             }
 
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 fail("disconnect should not be called " + node);
             }
         };
@@ -924,7 +924,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
             }
 
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 latch.countDown();
             }
         };
@@ -2118,7 +2118,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)
             }
 
             @Override
-            public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
+            public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {
                 fail("disconnect should not be called " + node);
             }
         };
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/regex/RegexMatch.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/regex/RegexMatch.java
@@ -72,22 +72,25 @@ public String nodeString() {
     }
 
     void serializeCaseInsensitivity(StreamOutput out) throws IOException {
-        if (out.getTransportVersion().before(TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY)) {
-            if (caseInsensitive()) {
-                // The plan has been optimized to run a case-insensitive match, which the remote peer cannot be notified of. Simply avoiding
-                // the serialization of the boolean would result in wrong results.
-                throw new EsqlIllegalArgumentException(
-                    name() + " with case insensitivity is not supported in peer node's version [{}]. Upgrade to version [{}] or newer.",
-                    out.getTransportVersion(),
-                    TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY
-                );
-            } // else: write nothing, the remote peer can execute the case-sensitive query
-        } else {
+        var transportVersion = out.getTransportVersion();
+        if (transportVersion.onOrAfter(TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY)
+            || transportVersion.isPatchFrom(TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY_8_19)) {
             out.writeBoolean(caseInsensitive());
-        }
+        } else if (caseInsensitive()) {
+            // The plan has been optimized to run a case-insensitive match, which the remote peer cannot be notified of. Simply avoiding
+            // the serialization of the boolean would result in wrong results.
+            throw new EsqlIllegalArgumentException(
+                name() + " with case insensitivity is not supported in peer node's version [{}]. Upgrade to version [{}, {}] or newer.",
+                transportVersion,
+                TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY_8_19,
+                TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY
+            );
+        } // else: write nothing, the remote peer can execute the case-sensitive query
     }
 
     static boolean deserializeCaseInsensitivity(StreamInput in) throws IOException {
-        return in.getTransportVersion().onOrAfter(TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY) && in.readBoolean();
+        var transportVersion = in.getTransportVersion();
+        return (transportVersion.onOrAfter(TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY)
+            || transportVersion.isPatchFrom(TransportVersions.ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY_8_19)) && in.readBoolean();
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -60,9 +60,9 @@ final class DelegatingNodeConnectionListener implements TransportConnectionListe`
`60`	`60`	`private final CopyOnWriteArrayList<TransportConnectionListener> listeners = new CopyOnWriteArrayList<>();`
`61`	`61`
`62`	`62`	`@Override`
`63`		`- public void onNodeDisconnected(DiscoveryNode key, Transport.Connection connection) {`
	`63`	`+ public void onNodeDisconnected(DiscoveryNode key, @Nullable Exception closeException) {`
`64`	`64`	`for (TransportConnectionListener listener : listeners) {`
`65`		`- listener.onNodeDisconnected(key, connection);`
	`65`	`+ listener.onNodeDisconnected(key, closeException);`
`66`	`66`	`}`
`67`	`67`	`}`
`68`	`68`
Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,7 @@ public void onNodeConnected(DiscoveryNode node, Transport.Connection connection)`
`61`	`61`	`}`
`62`	`62`
`63`	`63`	`@Override`
`64`		`- public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {`
	`64`	`+ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeException) {`
`65`	`65`	`removeConnectedNode(node);`
`66`	`66`	`}`
`67`	`67`	`});`