Use a user configurable timeout than a boolean

pawankartik-elastic · pawankartik-elastic · commit da8bd52fa746 · 2025-07-10T11:06:38.000+01:00
diff --git a/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java b/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java
@@ -168,8 +168,7 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
     private final Client client;
     private final UsageService usageService;
     private final boolean collectTelemetry;
-    private final boolean alwaysEstablishConnection;
-    private static final long FORCE_CONNECT_CLUSTER_DEFAULT_TIMEOUT = 3L;
+    private final TimeValue forceConnectTimeoutSecs;
 
     @Inject
     public TransportSearchAction(
@@ -218,8 +217,8 @@ public TransportSearchAction(
         this.searchResponseMetrics = searchResponseMetrics;
         this.client = client;
         this.usageService = usageService;
-        alwaysEstablishConnection = settings.getAsBoolean("serverless.force_cluster_reconnect", false);
-        logger.info("Should force reconnect cluster: {}", alwaysEstablishConnection);
+        forceConnectTimeoutSecs = settings.getAsTime("search.ccs.force_connect_timeout", null);
+        logger.info("Should force reconnect cluster: {}", forceConnectTimeoutSecs);
     }
 
     private Map<String, OriginalIndices> buildPerIndexOriginalIndices(
@@ -452,7 +451,7 @@ void executeRequest(
                             searchPhaseProvider.apply(l)
                         ),
                         transportService,
-                        alwaysEstablishConnection
+                        forceConnectTimeoutSecs
                     );
                 } else {
                     final SearchContextId searchContext = resolvedIndices.getSearchContextId();
@@ -513,7 +512,7 @@ void executeRequest(
                                 searchPhaseProvider.apply(finalDelegate)
                             );
                         }),
-                        alwaysEstablishConnection
+                        forceConnectTimeoutSecs
                     );
                 }
             }
@@ -644,21 +643,20 @@ public static boolean shouldMinimizeRoundtrips(SearchRequest searchRequest) {
     /**
      * Return a subscribable listener with optional timeout depending on force reconnect setting is registered or
      * not.
-     * @param alwaysEstablishConnection If we're running in a context where we always need to re-connect.
-     *                                  This value determines if we need to add a short timeout to avoid waiting
-     *                                  for long durations to reconnect.
+     * @param forceConnectTimeoutSecs Timeout in seconds that determines how long we'll wait to establish a connection
+     *                                to a remote.
      * @param threadPool The thread pool that'll be used for the timeout.
      * @param timeoutExecutor The executor that should be used for the timeout.
      * @return SubscribableListener A listener with optionally added timeout.
      */
     private static SubscribableListener<Transport.Connection> getListenerWithOptionalTimeout(
-        boolean alwaysEstablishConnection,
+        TimeValue forceConnectTimeoutSecs,
         ThreadPool threadPool,
         Executor timeoutExecutor
     ) {
         var subscribableListener = new SubscribableListener<Transport.Connection>();
-        if (alwaysEstablishConnection) {
-            subscribableListener.addTimeout(TimeValue.timeValueSeconds(FORCE_CONNECT_CLUSTER_DEFAULT_TIMEOUT), threadPool, timeoutExecutor);
+        if (forceConnectTimeoutSecs != null) {
+            subscribableListener.addTimeout(forceConnectTimeoutSecs, threadPool, timeoutExecutor);
         }
 
         return subscribableListener;
@@ -667,12 +665,13 @@ private static SubscribableListener<Transport.Connection> getListenerWithOptiona
     /**
      * The default disconnected strategy for Elasticsearch is RECONNECT_UNLESS_SKIP_UNAVAILABLE. So we either force
      * connect if required (like in CPS) or when skip unavailable is false for a cluster.
-     * @param alwaysEstablishConnection If we're running in a context where we always need to re-connect.
+     * @param forceConnectTimeoutSecs The timeout value from the force connect setting.
+     *                                If it is set, use it as it takes precedence.
      * @param skipUnavailable The usual skip unavailable setting.
      * @return boolean If we should always force reconnect.
      */
-    private static boolean shouldEstablishConnection(boolean alwaysEstablishConnection, boolean skipUnavailable) {
-        return alwaysEstablishConnection || skipUnavailable == false;
+    private static boolean shouldEstablishConnection(TimeValue forceConnectTimeoutSecs, boolean skipUnavailable) {
+        return forceConnectTimeoutSecs != null || skipUnavailable == false;
     }
 
     /**
@@ -691,7 +690,7 @@ static void ccsRemoteReduce(
         ActionListener<SearchResponse> listener,
         BiConsumer<SearchRequest, ActionListener<SearchResponse>> localSearchConsumer,
         TransportService transportService,
-        boolean shouldForceReconnectCluster
+        TimeValue forceConnectTimeoutSecs
     ) {
         final var remoteClientResponseExecutor = threadPool.executor(ThreadPool.Names.SEARCH_COORDINATION);
         if (resolvedIndices.getLocalIndices() == null && resolvedIndices.getRemoteClusterIndices().size() == 1) {
@@ -710,7 +709,7 @@ static void ccsRemoteReduce(
                 true
             );
 
-            var connectionListener = getListenerWithOptionalTimeout(shouldForceReconnectCluster, threadPool, remoteClientResponseExecutor);
+            var connectionListener = getListenerWithOptionalTimeout(forceConnectTimeoutSecs, threadPool, remoteClientResponseExecutor);
             var searchListener = new ActionListener<SearchResponse>() {
                 @Override
                 public void onResponse(SearchResponse searchResponse) {
@@ -770,7 +769,7 @@ public void onFailure(Exception e) {
 
             remoteClusterService.maybeEnsureConnectedAndGetConnection(
                 clusterAlias,
-                shouldEstablishConnection(shouldForceReconnectCluster, skipUnavailable),
+                shouldEstablishConnection(forceConnectTimeoutSecs, skipUnavailable),
                 connectionListener
             );
         } else {
@@ -809,7 +808,7 @@ public void onFailure(Exception e) {
                 );
 
                 SubscribableListener<Transport.Connection> connectionListener = getListenerWithOptionalTimeout(
-                    shouldForceReconnectCluster,
+                    forceConnectTimeoutSecs,
                     threadPool,
                     remoteClientResponseExecutor
                 );
@@ -828,7 +827,7 @@ public void onFailure(Exception e) {
 
                 remoteClusterService.maybeEnsureConnectedAndGetConnection(
                     clusterAlias,
-                    shouldEstablishConnection(shouldForceReconnectCluster, skipUnavailable),
+                    shouldEstablishConnection(forceConnectTimeoutSecs, skipUnavailable),
                     connectionListener
                 );
             }
@@ -897,7 +896,7 @@ static void collectSearchShards(
         SearchTimeProvider timeProvider,
         TransportService transportService,
         ActionListener<Map<String, SearchShardsResponse>> listener,
-        boolean shouldForceReconnectCluster
+        TimeValue forceConnectTimeoutSecs
     ) {
         RemoteClusterService remoteClusterService = transportService.getRemoteClusterService();
         final CountDown responsesCountDown = new CountDown(remoteIndicesByCluster.size());
@@ -929,7 +928,7 @@ Map<String, SearchShardsResponse> createFinalResponse() {
 
             var threadPool = transportService.getThreadPool();
             var connectionListener = getListenerWithOptionalTimeout(
-                shouldForceReconnectCluster,
+                forceConnectTimeoutSecs,
                 threadPool,
                 threadPool.executor(ThreadPool.Names.SEARCH_COORDINATION)
             );
@@ -977,7 +976,7 @@ Map<String, SearchShardsResponse> createFinalResponse() {
 
             remoteClusterService.maybeEnsureConnectedAndGetConnection(
                 clusterAlias,
-                shouldEstablishConnection(shouldForceReconnectCluster, skipUnavailable),
+                shouldEstablishConnection(forceConnectTimeoutSecs, skipUnavailable),
                 connectionListener
             );
         }
diff --git a/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java b/server/src/test/java/org/elasticsearch/action/search/TransportSearchActionTests.java
@@ -548,7 +548,7 @@ public void testCCSRemoteReduceMergeFails() throws Exception {
                 listener,
                 (r, l) -> setOnce.set(Tuple.tuple(r, l)),
                 service,
-                false
+                null
             );
             if (localIndices == null) {
                 assertNull(setOnce.get());
@@ -625,7 +625,7 @@ public void testCCSRemoteReduce() throws Exception {
                     listener,
                     (r, l) -> setOnce.set(Tuple.tuple(r, l)),
                     service,
-                    false
+                    null
                 );
                 if (localIndices == null) {
                     assertNull(setOnce.get());
@@ -683,7 +683,7 @@ public void testCCSRemoteReduce() throws Exception {
                     listener,
                     (r, l) -> setOnce.set(Tuple.tuple(r, l)),
                     service,
-                    false
+                    null
                 );
                 if (localIndices == null) {
                     assertNull(setOnce.get());
@@ -773,7 +773,7 @@ public void testCCSRemoteReduceWhereRemoteClustersFail() throws Exception {
                     listener,
                     (r, l) -> setOnce.set(Tuple.tuple(r, l)),
                     service,
-                    false
+                    null
                 );
                 if (localIndices == null) {
                     assertNull(setOnce.get());
@@ -874,7 +874,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     listener,
                     (r, l) -> setOnce.set(Tuple.tuple(r, l)),
                     service,
-                    false
+                    null
                 );
                 if (localIndices == null) {
                     assertNull(setOnce.get());
@@ -888,7 +888,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                 assertNotNull(failure.get());
                 assertThat(failure.get(), instanceOf(RemoteTransportException.class));
                 assertThat(failure.get().getMessage(), containsString("error while communicating with remote cluster ["));
-                // assertThat(failure.get().getCause(), instanceOf(NodeDisconnectedException.class));
+                assertThat(failure.get().getCause(), instanceOf(NodeDisconnectedException.class));
             }
 
             // setting skip_unavailable to true for all the disconnected clusters will make the request succeed again
@@ -927,7 +927,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     listener,
                     (r, l) -> setOnce.set(Tuple.tuple(r, l)),
                     service,
-                    false
+                    null
                 );
                 if (localIndices == null) {
                     assertNull(setOnce.get());
@@ -1002,7 +1002,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     listener,
                     (r, l) -> setOnce.set(Tuple.tuple(r, l)),
                     service,
-                    false
+                    null
                 );
                 if (localIndices == null) {
                     assertNull(setOnce.get());
@@ -1098,7 +1098,7 @@ public void testCollectSearchShards() throws Exception {
                     timeProvider,
                     service,
                     new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
-                    false
+                    null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
                 assertNotNull(response.get());
@@ -1128,7 +1128,7 @@ public void testCollectSearchShards() throws Exception {
                     timeProvider,
                     service,
                     new LatchedActionListener<>(ActionListener.wrap(r -> fail("no response expected"), failure::set), latch),
-                    false
+                    null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
                 assertEquals(numClusters, clusters.getClusterStateCount(SearchResponse.Cluster.Status.FAILED));
@@ -1177,7 +1177,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     timeProvider,
                     service,
                     new LatchedActionListener<>(ActionListener.wrap(r -> fail("no response expected"), failure::set), latch),
-                    false
+                    null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
                 assertEquals(numDisconnectedClusters, clusters.getClusterStateCount(SearchResponse.Cluster.Status.FAILED));
@@ -1208,7 +1208,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     timeProvider,
                     service,
                     new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
-                    false
+                    null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
                 assertNotNull(response.get());
@@ -1255,7 +1255,7 @@ public void onNodeDisconnected(DiscoveryNode node, @Nullable Exception closeExce
                     timeProvider,
                     service,
                     new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(response::set), latch),
-                    false
+                    null
                 );
                 awaitLatch(latch, 5, TimeUnit.SECONDS);
                 assertEquals(0, clusters.getClusterStateCount(SearchResponse.Cluster.Status.SKIPPED));
