[ML][Data Frame] cleaning up tests since tasks are cancelled onfinish (#43136)

* [ML][Data Frame] cleaning up usage test since tasks are cancelled onfinish

* Update DataFrameUsageIT.java

* Fixing additional test, waiting for task to complete

* removing unused import

* unmuting test

x-pack/plugin/data-frame/qa/multi-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameTransformIT.java

@@ -9,7 +9,6 @@ package org.elasticsearch.xpack.dataframe.integration;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.core.IndexerState;
 import org.elasticsearch.client.dataframe.transforms.DataFrameTransformConfig;
-import org.elasticsearch.client.dataframe.transforms.DataFrameTransformStateAndStats;
 import org.elasticsearch.client.dataframe.transforms.pivot.SingleGroupSource;
 import org.elasticsearch.client.dataframe.transforms.pivot.TermsGroupSource;
 import org.elasticsearch.search.aggregations.AggregationBuilders;
@@ -30,7 +29,6 @@ public class DataFrameTransformIT extends DataFrameIntegTestCase {
         cleanUp();
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/43139")
     public void testDataFrameTransformCrud() throws Exception {
         createReviewsIndex();
 
@@ -54,9 +52,10 @@ public class DataFrameTransformIT extends DataFrameIntegTestCase {
 
         waitUntilCheckpoint(config.getId(), 1L);
 
-        DataFrameTransformStateAndStats stats = getDataFrameTransformStats(config.getId()).getTransformsStateAndStats().get(0);
-
-        assertThat(stats.getTransformState().getIndexerState(), equalTo(IndexerState.STOPPED));
+        // It will eventually be stopped
+        assertBusy(() ->
+            assertThat(getDataFrameTransformStats(config.getId()).getTransformsStateAndStats().get(0).getTransformState().getIndexerState(),
+                equalTo(IndexerState.STOPPED)));
     }
 
 

x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameGetAndGetStatsIT.java

@@ -62,8 +62,11 @@ public class DataFrameGetAndGetStatsIT extends DataFrameRestTestCase {
         createPivotReviewsTransform("pivot_1", "pivot_reviews_1", null);
         createPivotReviewsTransform("pivot_2", "pivot_reviews_2", null);
 
+        // TODO: adjust when we support continuous
         startAndWaitForTransform("pivot_1", "pivot_reviews_1");
         startAndWaitForTransform("pivot_2", "pivot_reviews_2");
+        stopDataFrameTransform("pivot_1", false);
+        stopDataFrameTransform("pivot_2", false);
 
         // Alternate testing between admin and lowly user, as both should be able to get the configs and stats
         String authHeader = randomFrom(BASIC_AUTH_VALUE_DATA_FRAME_USER, BASIC_AUTH_VALUE_DATA_FRAME_ADMIN);

x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameRestTestCase.java

@@ -239,7 +239,8 @@ public abstract class DataFrameRestTestCase extends ESRestTestCase {
     void waitForDataFrameStopped(String transformId) throws Exception {
         assertBusy(() -> {
             assertEquals("stopped", getDataFrameTaskState(transformId));
-        }, 5, TimeUnit.SECONDS);
+            assertEquals("stopped", getDataFrameIndexerState(transformId));
+        }, 15, TimeUnit.SECONDS);
     }
 
     void waitForDataFrameCheckpoint(String transformId) throws Exception {

x-pack/plugin/data-frame/qa/single-node-tests/src/test/java/org/elasticsearch/xpack/dataframe/integration/DataFrameUsageIT.java

@@ -9,7 +9,7 @@ package org.elasticsearch.xpack.dataframe.integration;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
-import org.elasticsearch.xpack.core.dataframe.DataFrameField;
+
 import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformStateAndStats;
 import org.elasticsearch.xpack.dataframe.persistence.DataFrameInternalIndex;
 import org.junit.Before;
@@ -23,24 +23,10 @@ import static org.elasticsearch.xpack.core.dataframe.DataFrameField.INDEX_DOC_TY
 import static org.elasticsearch.xpack.dataframe.DataFrameFeatureSet.PROVIDED_STATS;
 
 public class DataFrameUsageIT extends DataFrameRestTestCase {
-    private boolean indicesCreated = false;
-
-    // preserve indices in order to reuse source indices in several test cases
-    @Override
-    protected boolean preserveIndicesUponCompletion() {
-        return true;
-    }
 
     @Before
     public void createIndexes() throws IOException {
-
-        // it's not possible to run it as @BeforeClass as clients aren't initialized then, so we need this little hack
-        if (indicesCreated) {
-            return;
-        }
-
         createReviewsIndex();
-        indicesCreated = true;
     }
 
     public void testUsage() throws Exception {
@@ -55,28 +41,23 @@ public class DataFrameUsageIT extends DataFrameRestTestCase {
 
         // create transforms
         createPivotReviewsTransform("test_usage", "pivot_reviews", null);
-        createPivotReviewsTransform("test_usage_no_task", "pivot_reviews_no_task", null);
-        createPivotReviewsTransform("test_usage_no_stats_or_task", "pivot_reviews_no_stats_or_task", null);
+        createPivotReviewsTransform("test_usage_no_stats", "pivot_reviews_no_stats", null);
         usageResponse = client().performRequest(new Request("GET", "_xpack/usage"));
         usageAsMap = entityAsMap(usageResponse);
-        assertEquals(3, XContentMapValues.extractValue("data_frame.transforms._all", usageAsMap));
-        assertEquals(3, XContentMapValues.extractValue("data_frame.transforms.stopped", usageAsMap));
-
-        startAndWaitForTransform("test_usage_no_task", "pivot_reviews_no_task");
-        stopDataFrameTransform("test_usage_no_task", false);
-        // Remove the task, we should still have the transform and its stat doc
-        client().performRequest(new Request("POST", "_tasks/_cancel?actions="+ DataFrameField.TASK_NAME+"*"));
+        assertEquals(2, XContentMapValues.extractValue("data_frame.transforms._all", usageAsMap));
+        assertEquals(2, XContentMapValues.extractValue("data_frame.transforms.stopped", usageAsMap));
 
         startAndWaitForTransform("test_usage", "pivot_reviews");
+        stopDataFrameTransform("test_usage", false);
 
         Request statsExistsRequest = new Request("GET",
             DataFrameInternalIndex.INDEX_NAME+"/_search?q=" +
                 INDEX_DOC_TYPE.getPreferredName() + ":" +
                 DataFrameTransformStateAndStats.NAME);
-        // Verify that we have our two stats documents
+        // Verify that we have one stat document
         assertBusy(() -> {
             Map<String, Object> hasStatsMap = entityAsMap(client().performRequest(statsExistsRequest));
-            assertEquals(2, XContentMapValues.extractValue("hits.total.value", hasStatsMap));
+            assertEquals(1, XContentMapValues.extractValue("hits.total.value", hasStatsMap));
         });
 
         Request getRequest = new Request("GET", DATAFRAME_ENDPOINT + "test_usage/_stats");
@@ -90,23 +71,13 @@ public class DataFrameUsageIT extends DataFrameRestTestCase {
             expectedStats.put(statName, statistic);
         }
 
-        getRequest = new Request("GET", DATAFRAME_ENDPOINT + "test_usage_no_task/_stats");
-        stats = entityAsMap(client().performRequest(getRequest));
-        for(String statName : PROVIDED_STATS) {
-            @SuppressWarnings("unchecked")
-            List<Integer> specificStatistic = ((List<Integer>)XContentMapValues.extractValue("transforms.stats." + statName, stats));
-            assertNotNull(specificStatistic);
-            Integer statistic = (specificStatistic).get(0);
-            expectedStats.merge(statName, statistic, Integer::sum);
-        }
-
         usageResponse = client().performRequest(new Request("GET", "_xpack/usage"));
 
         usageAsMap = entityAsMap(usageResponse);
         // we should see some stats
-        assertEquals(3, XContentMapValues.extractValue("data_frame.transforms._all", usageAsMap));
-        // TODO: due to auto-stop we only see stopped data frames
-        assertEquals(3, XContentMapValues.extractValue("data_frame.transforms.stopped", usageAsMap));
+        assertEquals(2, XContentMapValues.extractValue("data_frame.transforms._all", usageAsMap));
+        // TODO: Adjust when continuous is supported
+        assertEquals(2, XContentMapValues.extractValue("data_frame.transforms.stopped", usageAsMap));
         for(String statName : PROVIDED_STATS) {
             assertEquals("Incorrect stat " +  statName,
                     expectedStats.get(statName), XContentMapValues.extractValue("data_frame.stats." + statName, usageAsMap));