[7.x][Transform] Transform fix force stop race condition (#49249) (#49420)

fix force stopping transform if indexer state hasn't been written and/or is set to STOPPED. In certain situations the transform could not be stopped, which means the task could not be removed. Introduces improved abstraction in order to better test state handling in future.

Author: Hendrik Muhs

x-pack/plugin/transform/qa/multi-node-tests/src/test/java/org/elasticsearch/xpack/transform/integration/TransformIntegTestCase.java

@@ -89,10 +89,8 @@ protected StopTransformResponse stopTransform(String id) throws IOException {
         return stopTransform(id, true, null, false);
     }
 
-    protected StopTransformResponse stopTransform(String id,
-                                                  boolean waitForCompletion,
-                                                  TimeValue timeout,
-                                                  boolean waitForCheckpoint) throws IOException {
+    protected StopTransformResponse stopTransform(String id, boolean waitForCompletion, TimeValue timeout, boolean waitForCheckpoint)
+        throws IOException {
         RestHighLevelClient restClient = new TestRestHighLevelClient();
         return restClient.transform()
             .stopTransform(new StopTransformRequest(id, waitForCompletion, timeout, waitForCheckpoint), RequestOptions.DEFAULT);
@@ -105,8 +103,7 @@ protected StartTransformResponse startTransform(String id, RequestOptions option
 
     protected AcknowledgedResponse deleteTransform(String id) throws IOException {
         RestHighLevelClient restClient = new TestRestHighLevelClient();
-        AcknowledgedResponse response =
-            restClient.transform().deleteTransform(new DeleteTransformRequest(id), RequestOptions.DEFAULT);
+        AcknowledgedResponse response = restClient.transform().deleteTransform(new DeleteTransformRequest(id), RequestOptions.DEFAULT);
         if (response.isAcknowledged()) {
             transformConfigs.remove(id);
         }
@@ -118,8 +115,7 @@ protected AcknowledgedResponse putTransform(TransformConfig config, RequestOptio
             throw new IllegalArgumentException("transform [" + config.getId() + "] is already registered");
         }
         RestHighLevelClient restClient = new TestRestHighLevelClient();
-        AcknowledgedResponse response =
-            restClient.transform().putTransform(new PutTransformRequest(config), options);
+        AcknowledgedResponse response = restClient.transform().putTransform(new PutTransformRequest(config), options);
         if (response.isAcknowledged()) {
             transformConfigs.put(config.getId(), config);
         }
@@ -141,30 +137,33 @@ protected void waitUntilCheckpoint(String id, long checkpoint) throws Exception
     }
 
     protected void waitUntilCheckpoint(String id, long checkpoint, TimeValue waitTime) throws Exception {
-        assertBusy(() ->
-            assertEquals(checkpoint, getTransformStats(id)
-                .getTransformsStats()
-                .get(0)
-                .getCheckpointingInfo()
-                .getLast()
-                .getCheckpoint()),
+        assertBusy(
+            () -> assertEquals(
+                checkpoint,
+                getTransformStats(id).getTransformsStats().get(0).getCheckpointingInfo().getLast().getCheckpoint()
+            ),
             waitTime.getMillis(),
-            TimeUnit.MILLISECONDS);
+            TimeUnit.MILLISECONDS
+        );
     }
 
-    protected DateHistogramGroupSource createDateHistogramGroupSourceWithFixedInterval(String field,
-                                                                                       DateHistogramInterval interval,
-                                                                                       ZoneId zone) {
+    protected DateHistogramGroupSource createDateHistogramGroupSourceWithFixedInterval(
+        String field,
+        DateHistogramInterval interval,
+        ZoneId zone
+    ) {
         DateHistogramGroupSource.Builder builder = DateHistogramGroupSource.builder()
             .setField(field)
             .setInterval(new DateHistogramGroupSource.FixedInterval(interval))
             .setTimeZone(zone);
         return builder.build();
     }
 
-    protected DateHistogramGroupSource createDateHistogramGroupSourceWithCalendarInterval(String field,
-                                                                                          DateHistogramInterval interval,
-                                                                                          ZoneId zone) {
+    protected DateHistogramGroupSource createDateHistogramGroupSourceWithCalendarInterval(
+        String field,
+        DateHistogramInterval interval,
+        ZoneId zone
+    ) {
         DateHistogramGroupSource.Builder builder = DateHistogramGroupSource.builder()
             .setField(field)
             .setInterval(new DateHistogramGroupSource.CalendarInterval(interval))
@@ -188,35 +187,38 @@ protected AggregationConfig createAggConfig(AggregatorFactories.Builder aggregat
         return new AggregationConfig(aggregations);
     }
 
-    protected PivotConfig createPivotConfig(Map<String, SingleGroupSource> groups,
-                                            AggregatorFactories.Builder aggregations) throws Exception {
+    protected PivotConfig createPivotConfig(Map<String, SingleGroupSource> groups, AggregatorFactories.Builder aggregations)
+        throws Exception {
         return createPivotConfig(groups, aggregations, null);
     }
 
-    protected PivotConfig createPivotConfig(Map<String, SingleGroupSource> groups,
-                                            AggregatorFactories.Builder aggregations,
-                                            Integer size) throws Exception {
+    protected PivotConfig createPivotConfig(Map<String, SingleGroupSource> groups, AggregatorFactories.Builder aggregations, Integer size)
+        throws Exception {
         PivotConfig.Builder builder = PivotConfig.builder()
             .setGroups(createGroupConfig(groups))
             .setAggregationConfig(createAggConfig(aggregations))
             .setMaxPageSearchSize(size);
         return builder.build();
     }
 
-    protected TransformConfig createTransformConfig(String id,
-                                                    Map<String, SingleGroupSource> groups,
-                                                    AggregatorFactories.Builder aggregations,
-                                                    String destinationIndex,
-                                                    String... sourceIndices) throws Exception {
+    protected TransformConfig createTransformConfig(
+        String id,
+        Map<String, SingleGroupSource> groups,
+        AggregatorFactories.Builder aggregations,
+        String destinationIndex,
+        String... sourceIndices
+    ) throws Exception {
         return createTransformConfig(id, groups, aggregations, destinationIndex, QueryBuilders.matchAllQuery(), sourceIndices);
     }
 
-    protected TransformConfig.Builder createTransformConfigBuilder(String id,
-                                                                   Map<String, SingleGroupSource> groups,
-                                                                   AggregatorFactories.Builder aggregations,
-                                                                   String destinationIndex,
-                                                                   QueryBuilder queryBuilder,
-                                                                   String... sourceIndices) throws Exception {
+    protected TransformConfig.Builder createTransformConfigBuilder(
+        String id,
+        Map<String, SingleGroupSource> groups,
+        AggregatorFactories.Builder aggregations,
+        String destinationIndex,
+        QueryBuilder queryBuilder,
+        String... sourceIndices
+    ) throws Exception {
         return TransformConfig.builder()
             .setId(id)
             .setSource(SourceConfig.builder().setIndex(sourceIndices).setQueryConfig(createQueryConfig(queryBuilder)).build())
@@ -226,12 +228,14 @@ protected TransformConfig.Builder createTransformConfigBuilder(String id,
             .setDescription("Test transform config id: " + id);
     }
 
-    protected TransformConfig createTransformConfig(String id,
-                                                    Map<String, SingleGroupSource> groups,
-                                                    AggregatorFactories.Builder aggregations,
-                                                    String destinationIndex,
-                                                    QueryBuilder queryBuilder,
-                                                    String... sourceIndices) throws Exception {
+    protected TransformConfig createTransformConfig(
+        String id,
+        Map<String, SingleGroupSource> groups,
+        AggregatorFactories.Builder aggregations,
+        String destinationIndex,
+        QueryBuilder queryBuilder,
+        String... sourceIndices
+    ) throws Exception {
         return createTransformConfigBuilder(id, groups, aggregations, destinationIndex, queryBuilder, sourceIndices).build();
     }
 
@@ -272,8 +276,8 @@ protected void createReviewsIndex(String indexName, int numDocs) throws Exceptio
                     .endObject();
             }
             builder.endObject();
-            CreateIndexResponse response =
-                restClient.indices().create(new CreateIndexRequest(indexName).mapping(builder), RequestOptions.DEFAULT);
+            CreateIndexResponse response = restClient.indices()
+                .create(new CreateIndexRequest(indexName).mapping(builder), RequestOptions.DEFAULT);
             assertThat(response.isAcknowledged(), is(true));
         }
 
@@ -320,10 +324,14 @@ protected void createReviewsIndex(String indexName, int numDocs) throws Exceptio
 
     protected Map<String, Object> toLazy(ToXContent parsedObject) throws Exception {
         BytesReference bytes = XContentHelper.toXContent(parsedObject, XContentType.JSON, false);
-        try(XContentParser parser = XContentHelper.createParser(xContentRegistry(),
-            DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
-            bytes,
-            XContentType.JSON)) {
+        try (
+            XContentParser parser = XContentHelper.createParser(
+                xContentRegistry(),
+                DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
+                bytes,
+                XContentType.JSON
+            )
+        ) {
             return parser.mapOrdered();
         }
     }
@@ -349,16 +357,18 @@ protected NamedXContentRegistry xContentRegistry() {
 
     @Override
     protected Settings restClientSettings() {
-        final String token = "Basic " +
-            Base64.getEncoder().encodeToString(("x_pack_rest_user:x-pack-test-password").getBytes(StandardCharsets.UTF_8));
-        return Settings.builder()
-            .put(ThreadContext.PREFIX + ".Authorization", token)
-            .build();
+        final String token = "Basic "
+            + Base64.getEncoder().encodeToString(("x_pack_rest_user:x-pack-test-password").getBytes(StandardCharsets.UTF_8));
+        return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", token).build();
     }
 
     protected static class TestRestHighLevelClient extends RestHighLevelClient {
-        private static final List<NamedXContentRegistry.Entry> X_CONTENT_ENTRIES =
-            new SearchModule(Settings.EMPTY, false, Collections.emptyList()).getNamedXContents();
+        private static final List<NamedXContentRegistry.Entry> X_CONTENT_ENTRIES = new SearchModule(
+            Settings.EMPTY,
+            false,
+            Collections.emptyList()
+        ).getNamedXContents();
+
         TestRestHighLevelClient() {
             super(client(), restClient -> {}, X_CONTENT_ENTRIES);
         }

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

@@ -38,14 +38,11 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 
-
 public class TransformInternalIndexIT extends ESRestTestCase {
 
-
     private static final String CURRENT_INDEX = TransformInternalIndexConstants.LATEST_INDEX_NAME;
     private static final String OLD_INDEX = TransformInternalIndexConstants.INDEX_PATTERN + "001";
 
-
     public void testUpdateDeletesOldTransformConfig() throws Exception {
         TestRestHighLevelClient client = new TestRestHighLevelClient();
         // The mapping does not need to actually be the "OLD" mapping, we are testing that the old doc gets deleted, and the new one
@@ -63,8 +60,12 @@ public void testUpdateDeletesOldTransformConfig() throws Exception {
         createSourceIndex(transformIndex);
         String transformId = "transform-update-deletes-old-transform-config";
         String config = "{\"dest\": {\"index\":\"bar\"},"
-            + " \"source\": {\"index\":\"" + transformIndex + "\", \"query\": {\"match_all\":{}}},"
-            + " \"id\": \""+transformId+"\","
+            + " \"source\": {\"index\":\""
+            + transformIndex
+            + "\", \"query\": {\"match_all\":{}}},"
+            + " \"id\": \""
+            + transformId
+            + "\","
             + " \"doc_type\": \"data_frame_transform_config\","
             + " \"pivot\": {"
             + "   \"group_by\": {"
@@ -79,22 +80,23 @@ public void testUpdateDeletesOldTransformConfig() throws Exception {
             + " } } } },"
             + "\"frequency\":\"1s\""
             + "}";
-        client.index(new IndexRequest(OLD_INDEX)
-                .id(TransformConfig.documentId(transformId))
+        client.index(
+            new IndexRequest(OLD_INDEX).id(TransformConfig.documentId(transformId))
                 .source(config, XContentType.JSON)
                 .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE),
-            RequestOptions.DEFAULT);
-        GetResponse getResponse = client.get(new GetRequest(OLD_INDEX, TransformConfig.documentId(transformId)),
-            RequestOptions.DEFAULT);
+            RequestOptions.DEFAULT
+        );
+        GetResponse getResponse = client.get(new GetRequest(OLD_INDEX, TransformConfig.documentId(transformId)), RequestOptions.DEFAULT);
         assertThat(getResponse.isExists(), is(true));
 
-        GetTransformResponse response = client.transform()
-            .getTransform(new GetTransformRequest(transformId), RequestOptions.DEFAULT);
+        GetTransformResponse response = client.transform().getTransform(new GetTransformRequest(transformId), RequestOptions.DEFAULT);
         assertThat(response.getTransformConfigurations().get(0).getId(), equalTo(transformId));
 
-        UpdateTransformResponse updated = client.transform().updateTransform(
-            new UpdateTransformRequest(TransformConfigUpdate.builder().setDescription("updated").build(), transformId),
-            RequestOptions.DEFAULT);
+        UpdateTransformResponse updated = client.transform()
+            .updateTransform(
+                new UpdateTransformRequest(TransformConfigUpdate.builder().setDescription("updated").build(), transformId),
+                RequestOptions.DEFAULT
+            );
 
         assertThat(updated.getTransformConfiguration().getId(), equalTo(transformId));
         assertThat(updated.getTransformConfiguration().getDescription(), equalTo("updated"));
@@ -104,19 +106,15 @@ public void testUpdateDeletesOldTransformConfig() throws Exception {
         assertThat(getResponse.isExists(), is(false));
 
         // New should be here
-        getResponse = client.get(new GetRequest(CURRENT_INDEX, TransformConfig.documentId(transformId)),
-            RequestOptions.DEFAULT);
+        getResponse = client.get(new GetRequest(CURRENT_INDEX, TransformConfig.documentId(transformId)), RequestOptions.DEFAULT);
         assertThat(getResponse.isExists(), is(true));
     }
 
-
     @Override
     protected Settings restClientSettings() {
-        final String token = "Basic " +
-            Base64.getEncoder().encodeToString(("x_pack_rest_user:x-pack-test-password").getBytes(StandardCharsets.UTF_8));
-        return Settings.builder()
-            .put(ThreadContext.PREFIX + ".Authorization", token)
-            .build();
+        final String token = "Basic "
+            + Base64.getEncoder().encodeToString(("x_pack_rest_user:x-pack-test-password").getBytes(StandardCharsets.UTF_8));
+        return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", token).build();
     }
 
     private void createSourceIndex(String index) throws IOException {