Force reconnect to remote clusters with a short timeout for CPS

pawankartik-elastic · pawankartik-elastic · commit 6da2ecafe534 · 2025-07-17T21:39:27.000+01:00
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/indices/cluster/RemoteFieldCapsForceConnectTimeoutIT.java b/server/src/internalClusterTest/java/org/elasticsearch/indices/cluster/RemoteFieldCapsForceConnectTimeoutIT.java
@@ -0,0 +1,112 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.indices.cluster;
+
+import org.elasticsearch.action.fieldcaps.FieldCapabilitiesRequest;
+import org.elasticsearch.action.fieldcaps.TransportFieldCapabilitiesAction;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.plugins.ClusterPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.AbstractMultiClustersTestCase;
+import org.elasticsearch.test.transport.MockTransportService;
+import org.elasticsearch.transport.TransportService;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+
+public class RemoteFieldCapsForceConnectTimeoutIT extends AbstractMultiClustersTestCase {
+    private static final String REMOTE_CLUSTER_1 = "cluster-a";
+
+    public static class ForceConnectTimeoutPlugin extends Plugin implements ClusterPlugin {
+        @Override
+        public List<Setting<?>> getSettings() {
+            return List.of(ForceConnectTimeoutSetting);
+        }
+    }
+
+    private static final Setting<String> ForceConnectTimeoutSetting = Setting.simpleString(
+        "search.ccs.force_connect_timeout",
+        Setting.Property.NodeScope
+    );
+
+    @Override
+    protected List<String> remoteClusterAlias() {
+        return List.of(REMOTE_CLUSTER_1);
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
+        return CollectionUtils.appendToCopy(super.nodePlugins(clusterAlias), ForceConnectTimeoutPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings() {
+        /*
+         * This is the setting that controls how long TransportFieldCapabilitiesAction will wait for establishing a connection
+         * with a remote. At present, we set it to low 1s to prevent stalling the test for too long -- this is consistent
+         * with what we've done in other tests.
+         */
+        return Settings.builder().put(super.nodeSettings()).put("search.ccs.force_connect_timeout", "1s").build();
+    }
+
+    @Override
+    protected Map<String, Boolean> skipUnavailableForRemoteClusters() {
+        return Map.of(REMOTE_CLUSTER_1, true);
+    }
+
+    public void testTimeoutSetting() {
+        var latch = new CountDownLatch(1);
+        for (String nodeName : cluster(LOCAL_CLUSTER).getNodeNames()) {
+            MockTransportService mts = (MockTransportService) cluster(LOCAL_CLUSTER).getInstance(TransportService.class, nodeName);
+
+            mts.addConnectBehavior(
+                cluster(REMOTE_CLUSTER_1).getInstance(TransportService.class, randomFrom(cluster(REMOTE_CLUSTER_1).getNodeNames())),
+                ((transport, discoveryNode, profile, listener) -> {
+                    try {
+                        latch.await();
+                    } catch (InterruptedException e) {
+                        throw new AssertionError(e);
+                    }
+
+                    transport.openConnection(discoveryNode, profile, listener);
+                })
+            );
+        }
+
+        // Add some dummy data to prove we are communicating fine with the remote.
+        assertAcked(client(REMOTE_CLUSTER_1).admin().indices().prepareCreate("test-index"));
+        client(REMOTE_CLUSTER_1).prepareIndex("test-index").setSource("sample-field", "sample-value").get();
+        client(REMOTE_CLUSTER_1).admin().indices().prepareRefresh("test-index").get();
+
+        /*
+         * Do a full restart so that our custom connect behaviour takes effect since it does not apply to
+         * pre-existing connections -- they're already established by the time this test runs.
+         */
+        try {
+            cluster(REMOTE_CLUSTER_1).fullRestart();
+        } catch (Exception e) {
+            throw new AssertionError(e);
+        } finally {
+            var fieldCapsRequest = new FieldCapabilitiesRequest();
+            fieldCapsRequest.indices("*", "*:*");
+            fieldCapsRequest.fields("foo", "bar", "baz");
+            var result = safeGet(client().execute(TransportFieldCapabilitiesAction.TYPE, fieldCapsRequest));
+
+            latch.countDown();
+            result.decRef();
+        }
+    }
+}
diff --git a/server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesAction.java b/server/src/main/java/org/elasticsearch/action/fieldcaps/TransportFieldCapabilitiesAction.java
@@ -13,6 +13,7 @@
 import org.apache.lucene.util.automaton.TooComplexToDeterminizeException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.OriginalIndices;
@@ -22,7 +23,7 @@
 import org.elasticsearch.action.support.ChannelActionListener;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.RefCountingRunnable;
-import org.elasticsearch.client.internal.RemoteClusterClient;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
@@ -37,6 +38,7 @@
 import org.elasticsearch.common.util.concurrent.ThrottledTaskRunner;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.IndicesService;
@@ -48,9 +50,10 @@
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.RemoteClusterAware;
-import org.elasticsearch.transport.RemoteClusterService;
+import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportChannel;
 import org.elasticsearch.transport.TransportRequestHandler;
+import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.ArrayList;
@@ -91,6 +94,8 @@ public class TransportFieldCapabilitiesAction extends HandledTransportAction<Fie
 
     private final IndicesService indicesService;
     private final boolean ccsCheckCompatibility;
+    private final ThreadPool threadPool;
+    private final TimeValue forceConnectTimeoutSecs;
 
     @Inject
     public TransportFieldCapabilitiesAction(
@@ -117,14 +122,22 @@ public TransportFieldCapabilitiesAction(
             new NodeTransportHandler()
         );
         this.ccsCheckCompatibility = SearchService.CCS_VERSION_CHECK_SETTING.get(clusterService.getSettings());
+        this.threadPool = threadPool;
+        this.forceConnectTimeoutSecs = clusterService.getSettings().getAsTime("search.ccs.force_connect_timeout", null);
     }
 
     @Override
     protected void doExecute(Task task, FieldCapabilitiesRequest request, final ActionListener<FieldCapabilitiesResponse> listener) {
         executeRequest(
             task,
             request,
-            (remoteClient, remoteRequest, remoteListener) -> remoteClient.execute(REMOTE_TYPE, remoteRequest, remoteListener),
+            (transportService, conn, fieldCapabilitiesRequest, responseHandler) -> transportService.sendRequest(
+                conn,
+                REMOTE_TYPE.name(),
+                fieldCapabilitiesRequest,
+                TransportRequestOptions.EMPTY,
+                responseHandler
+            ),
             listener
         );
     }
@@ -268,12 +281,6 @@ private void doExecuteForked(
             for (Map.Entry<String, OriginalIndices> remoteIndices : remoteClusterIndices.entrySet()) {
                 String clusterAlias = remoteIndices.getKey();
                 OriginalIndices originalIndices = remoteIndices.getValue();
-                var remoteClusterClient = transportService.getRemoteClusterService()
-                    .getRemoteClusterClient(
-                        clusterAlias,
-                        singleThreadedExecutor,
-                        RemoteClusterService.DisconnectedStrategy.RECONNECT_UNLESS_SKIP_UNAVAILABLE
-                    );
                 FieldCapabilitiesRequest remoteRequest = prepareRemoteRequest(clusterAlias, request, originalIndices, nowInMillis);
                 ActionListener<FieldCapabilitiesResponse> remoteListener = ActionListener.wrap(response -> {
                     for (FieldCapabilitiesIndexResponse resp : response.getIndexResponses()) {
@@ -299,18 +306,34 @@ private void doExecuteForked(
                         handleIndexFailure.accept(RemoteClusterAware.buildRemoteIndexName(clusterAlias, index), ex);
                     }
                 });
-                remoteRequestExecutor.executeRemoteRequest(
-                    remoteClusterClient,
-                    remoteRequest,
+
+                SubscribableListener<Transport.Connection> connectionListener = new SubscribableListener<>();
+                if (forceConnectTimeoutSecs != null) {
+                    connectionListener.addTimeout(forceConnectTimeoutSecs, threadPool, singleThreadedExecutor);
+                }
+
+                connectionListener.addListener(
                     // The underlying transport service may call onFailure with a thread pool other than search_coordinator.
                     // This fork is a workaround to ensure that the merging of field-caps always occurs on the search_coordinator.
                     // TODO: remove this workaround after we fixed https://github.com/elastic/elasticsearch/issues/107439
                     new ForkingOnFailureActionListener<>(
                         singleThreadedExecutor,
                         true,
                         ActionListener.releaseAfter(remoteListener, refs.acquire())
+                    ).delegateFailure(
+                        (responseListener, conn) -> remoteRequestExecutor.executeRemoteRequest(
+                            transportService,
+                            conn,
+                            remoteRequest,
+                            new ActionListenerResponseHandler<>(responseListener, FieldCapabilitiesResponse::new, singleThreadedExecutor)
+                        )
                     )
                 );
+
+                boolean ensureConnected = forceConnectTimeoutSecs != null
+                    || transportService.getRemoteClusterService().isSkipUnavailable(clusterAlias) == false;
+                transportService.getRemoteClusterService()
+                    .maybeEnsureConnectedAndGetConnection(clusterAlias, ensureConnected, connectionListener);
             }
         }
     }
@@ -340,9 +363,10 @@ public void onFailure(Exception e) {
 
     public interface RemoteRequestExecutor {
         void executeRemoteRequest(
-            RemoteClusterClient remoteClient,
+            TransportService transportService,
+            Transport.Connection conn,
             FieldCapabilitiesRequest remoteRequest,
-            ActionListener<FieldCapabilitiesResponse> remoteListener
+            ActionListenerResponseHandler<FieldCapabilitiesResponse> responseHandler
         );
     }
 
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlResolveFieldsAction.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlResolveFieldsAction.java
@@ -8,17 +8,19 @@
 
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.RemoteClusterActionType;
 import org.elasticsearch.action.fieldcaps.FieldCapabilitiesRequest;
 import org.elasticsearch.action.fieldcaps.FieldCapabilitiesResponse;
 import org.elasticsearch.action.fieldcaps.TransportFieldCapabilitiesAction;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
-import org.elasticsearch.client.internal.RemoteClusterClient;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.tasks.Task;
+import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportService;
 
 /**
@@ -53,15 +55,14 @@ protected void doExecute(Task task, FieldCapabilitiesRequest request, final Acti
     }
 
     void executeRemoteRequest(
-        RemoteClusterClient remoteClient,
+        TransportService transportService,
+        Transport.Connection conn,
         FieldCapabilitiesRequest remoteRequest,
-        ActionListener<FieldCapabilitiesResponse> remoteListener
+        ActionListenerResponseHandler<FieldCapabilitiesResponse> responseHandler
     ) {
-        remoteClient.getConnection(remoteRequest, remoteListener.delegateFailure((l, conn) -> {
-            var remoteAction = conn.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)
-                ? RESOLVE_REMOTE_TYPE
-                : TransportFieldCapabilitiesAction.REMOTE_TYPE;
-            remoteClient.execute(conn, remoteAction, remoteRequest, l);
-        }));
+        var remoteAction = conn.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)
+            ? RESOLVE_REMOTE_TYPE
+            : TransportFieldCapabilitiesAction.REMOTE_TYPE;
+        transportService.sendRequest(conn, remoteAction.name(), remoteRequest, TransportRequestOptions.EMPTY, responseHandler);
     }
 }
