[Transform] Improve reporting status of the transform that is about to finish (#95672)

docs/changelog/95672.yaml

@@ -0,0 +1,5 @@
+pr: 95672
+summary: Improve reporting status of the transform that is about to finish
+area: Transform
+type: bug
+issues: []

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformRestTestCase.java

@@ -473,10 +473,17 @@ public abstract class TransformRestTestCase extends ESRestTestCase {
         // start the transform
         startTransform(transformId, authHeader, secondaryAuthHeader, null, warnings);
         assertTrue(indexExists(transformIndex));
-        // wait until the transform has been created and all data is available
-        waitForTransformCheckpoint(transformId);
-
-        waitForTransformStopped(transformId);
+        assertBusy(() -> {
+            Map<?, ?> transformStatsAsMap = getTransformStateAndStats(transformId);
+            // wait until the transform has been created and all data is available
+            assertEquals(
+                "Stats were: " + transformStatsAsMap,
+                1,
+                XContentMapValues.extractValue("checkpointing.last.checkpoint", transformStatsAsMap)
+            );
+            // wait until the transform is stopped
+            assertEquals("Stats were: " + transformStatsAsMap, "stopped", XContentMapValues.extractValue("state", transformStatsAsMap));
+        }, 30, TimeUnit.SECONDS);
         refreshIndex(transformIndex);
     }
 
@@ -539,16 +546,16 @@ public abstract class TransformRestTestCase extends ESRestTestCase {
         return createRequestWithSecondaryAuth(method, endpoint, authHeader, null);
     }
 
-    void waitForTransformStopped(String transformId) throws Exception {
-        assertBusy(() -> { assertEquals("stopped", getTransformState(transformId)); }, 15, TimeUnit.SECONDS);
-    }
-
-    void waitForTransformCheckpoint(String transformId) throws Exception {
-        waitForTransformCheckpoint(transformId, 1L);
-    }
-
     void waitForTransformCheckpoint(String transformId, long checkpoint) throws Exception {
-        assertBusy(() -> assertEquals(checkpoint, getTransformCheckpoint(transformId)), 30, TimeUnit.SECONDS);
+        assertBusy(() -> {
+            Map<?, ?> transformStatsAsMap = getTransformStateAndStats(transformId);
+            assertNotEquals("Stats were: " + transformStatsAsMap, "failed", XContentMapValues.extractValue("state", transformStatsAsMap));
+            assertEquals(
+                "Stats were: " + transformStatsAsMap,
+                (int) checkpoint,
+                XContentMapValues.extractValue("checkpointing.last.checkpoint", transformStatsAsMap)
+            );
+        }, 30, TimeUnit.SECONDS);
     }
 
     void refreshIndex(String index) throws IOException {
@@ -632,16 +639,6 @@ public abstract class TransformRestTestCase extends ESRestTestCase {
         wipeAllIndices();
     }
 
-    static int getTransformCheckpoint(String transformId) throws IOException {
-        Response statsResponse = client().performRequest(new Request("GET", getTransformEndpoint() + transformId + "/_stats"));
-
-        Map<?, ?> transformStatsAsMap = (Map<?, ?>) ((List<?>) entityAsMap(statsResponse).get("transforms")).get(0);
-
-        // assert that the transform did not fail
-        assertNotEquals("Stats were: " + transformStatsAsMap, "failed", XContentMapValues.extractValue("state", transformStatsAsMap));
-        return (int) XContentMapValues.extractValue("checkpointing.last.checkpoint", transformStatsAsMap);
-    }
-
     protected void setupDataAccessRole(String role, String... indices) throws IOException {
         String indicesStr = Arrays.stream(indices).collect(Collectors.joining("\",\"", "\"", "\""));
         Request request = new Request("PUT", "/_security/role/" + role);

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformRobustnessIT.java

@@ -113,6 +113,27 @@ public class TransformRobustnessIT extends TransformRestTestCase {
         deleteTransform(transformId);
     }
 
+    public void testCreateAndDeleteTransformInALoop() throws IOException {
+        createReviewsIndex();
+
+        String transformId = "test_create_and_delete_in_a_loop";
+        String destIndex = transformId + "-dest";
+        for (int i = 0; i < 100; ++i) {
+            try {
+                // Create the batch transform
+                createPivotReviewsTransform(transformId, destIndex, null);
+                // Wait until the transform finishes
+                startAndWaitForTransform(transformId, destIndex);
+                // After the transform finishes, there should be no transform task left
+                assertEquals(0, getNumberOfTransformTasks());
+                // Delete the transform
+                deleteTransform(transformId);
+            } catch (AssertionError | Exception e) {
+                fail("Failure at iteration " + i + ": " + e.getMessage());
+            }
+        }
+    }
+
     @SuppressWarnings("unchecked")
     private int getNumberOfTransformTasks() throws IOException {
         final Request tasksRequest = new Request("GET", "/_tasks");

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java

@@ -335,7 +335,6 @@ public class TransportGetTransformStatsAction extends TransportTasksAction<Trans
         ClusterState clusterState,
         ActionListener<Void> listener
     ) {
-
         if (statsForTransformsWithoutTasks.isEmpty()) {
             // No work to do, but we must respond to the listener
             listener.onResponse(null);
@@ -366,10 +365,14 @@ public class TransportGetTransformStatsAction extends TransportTasksAction<Trans
                             )
                         );
                     } else {
+                        final boolean transformPersistentTaskIsStillRunning = TransformTask.getTransformTask(
+                            stat.getId(),
+                            clusterState
+                        ) != null;
                         allStateAndStats.add(
                             new TransformStats(
                                 stat.getId(),
-                                TransformStats.State.STOPPED,
+                                transformPersistentTaskIsStillRunning ? TransformStats.State.STOPPING : TransformStats.State.STOPPED,
                                 null,
                                 null,
                                 stat.getTransformStats(),

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformTask.java

@@ -19,6 +19,7 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.persistent.AllocatedPersistentTask;
+import org.elasticsearch.persistent.PersistentTaskParams;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata.PersistentTask;
 import org.elasticsearch.persistent.PersistentTasksService;
@@ -47,6 +48,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.function.Predicate;
 
@@ -464,6 +466,17 @@ public class TransformTask extends AllocatedPersistentTask implements TransformS
         logger.debug("[{}] shutdown of transform requested", transform.getId());
         transformScheduler.deregisterTransform(getTransformId());
         markAsCompleted();
+        waitForPersistentTask(Objects::isNull, null, new PersistentTasksService.WaitForPersistentTaskListener<>() {
+            @Override
+            public void onResponse(PersistentTask<PersistentTaskParams> persistentTask) {
+                logger.trace("[{}] successfully finished waiting for persistent task to disappear.", transform.getId());
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                logger.error(() -> "[" + transform.getId() + "] failure while waiting for persistent task to disappear.", e);
+            }
+        });
     }
 
     void persistStateToClusterState(TransformState state, ActionListener<PersistentTask<?>> listener) {