From 0c0e1797291808130e9c866b097fdcacb5b68a14 Mon Sep 17 00:00:00 2001
From: Hendrik Muhs <hendrik.muhs@elastic.co>
Date: Wed, 27 Sep 2023 10:38:22 +0200
Subject: [PATCH] [Transform] let _stats internally timeout if checkpoint
 information can not be retrieved (#99914)

getting the stats of a transform can timeout, this change lets the retrieval of checkpointing info timeout if it takes too
long. As stats are usually requested for all transforms, this prevents a failure if just 1 transform does not respond.
---
 docs/changelog/99914.yaml                     |  5 +++
 .../TransportGetTransformStatsAction.java     | 10 +++++-
 .../transform/transforms/TransformTask.java   | 33 +++++++++++++------
 3 files changed, 37 insertions(+), 11 deletions(-)
 create mode 100644 docs/changelog/99914.yaml

diff --git a/docs/changelog/99914.yaml b/docs/changelog/99914.yaml
new file mode 100644
index 0000000000000..8b0026a8ff9ca
--- /dev/null
+++ b/docs/changelog/99914.yaml
@@ -0,0 +1,5 @@
+pr: 99914
+summary: Let `_stats` internally timeout if checkpoint information can not be retrieved
+area: Transform
+type: bug
+issues: []
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java
index 07b753d8aea77..b62f0c95785b2 100644
--- a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java
+++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata.Assignment;
@@ -63,6 +64,9 @@ public class TransportGetTransformStatsAction extends TransportTasksAction<Trans
 
     private static final Logger logger = LogManager.getLogger(TransportGetTransformStatsAction.class);
 
+    // default timeout share to receive checkpoint info, with the default of 30s: 30s * 0.8 = 24s
+    private static final double CHECKPOINT_INFO_TIMEOUT_SHARE = 0.8;
+
     private final TransformConfigManager transformConfigManager;
     private final TransformCheckpointService transformCheckpointService;
     private final Client client;
@@ -114,6 +118,7 @@ protected void taskOperation(CancellableTask actionTask, Request request, Transf
         // Little extra insurance, make sure we only return transforms that aren't cancelled
         ClusterState state = clusterService.state();
         String nodeId = state.nodes().getLocalNode().getId();
+
         if (task.isCancelled() == false) {
             task.getCheckpointingInfo(
                 transformCheckpointService,
@@ -132,7 +137,10 @@ protected void taskOperation(CancellableTask actionTask, Request request, Transf
                             )
                         );
                     }
-                )
+                ),
+                // at this point the transport already spend some time budget in `doExecute`, it is hard to tell what is left:
+                // recording the time spend would be complex and crosses machine boundaries, that's why we use a heuristic here
+                TimeValue.timeValueMillis((long) (request.getTimeout().millis() * CHECKPOINT_INFO_TIMEOUT_SHARE))
             );
         } else {
             listener.onResponse(new Response(Collections.emptyList(), 0L));
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformTask.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformTask.java
index 8478525cb4a53..76e43818720fa 100644
--- a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformTask.java
+++ b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformTask.java
@@ -12,12 +12,15 @@
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ListenerTimeouts;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.ParentTaskAssigningClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.persistent.AllocatedPersistentTask;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata.PersistentTask;
@@ -179,18 +182,28 @@ public TransformIndexerStats getStats() {
 
     public void getCheckpointingInfo(
         TransformCheckpointService transformsCheckpointService,
-        ActionListener<TransformCheckpointingInfo> listener
+        ActionListener<TransformCheckpointingInfo> listener,
+        TimeValue timeout
     ) {
-        ActionListener<TransformCheckpointingInfoBuilder> checkPointInfoListener = ActionListener.wrap(infoBuilder -> {
-            if (context.getChangesLastDetectedAt() != null) {
-                infoBuilder.setChangesLastDetectedAt(context.getChangesLastDetectedAt());
-            }
-            if (context.getLastSearchTime() != null) {
-                infoBuilder.setLastSearchTime(context.getLastSearchTime());
-            }
-            listener.onResponse(infoBuilder.build());
-        }, listener::onFailure);
+        ActionListener<TransformCheckpointingInfoBuilder> checkPointInfoListener = ListenerTimeouts.wrapWithTimeout(
+            threadPool,
+            timeout,
+            threadPool.generic(),
+            ActionListener.wrap(infoBuilder -> {
+                if (context.getChangesLastDetectedAt() != null) {
+                    infoBuilder.setChangesLastDetectedAt(context.getChangesLastDetectedAt());
+                }
+                if (context.getLastSearchTime() != null) {
+                    infoBuilder.setLastSearchTime(context.getLastSearchTime());
+                }
+                listener.onResponse(infoBuilder.build());
+            }, listener::onFailure),
+            (ignore) -> listener.onFailure(
+                new ElasticsearchTimeoutException(format("Timed out retrieving checkpointing info after [%s]", timeout))
+            )
+        );
 
+        // TODO: pass `timeout` to the lower layers
         ClientTransformIndexer transformIndexer = getIndexer();
         if (transformIndexer == null) {
             transformsCheckpointService.getCheckpointingInfo(