[ML] Don't write timing stats on no-op (elastic#43680)

Similar to elastic/ml-cpp#512, if a job opens and closes and
does nothing in between we shouldn't write timing stats to
the results index.

Author: droberts195

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/TimingStatsReporter.java

@@ -42,6 +42,14 @@ public void reportBucketProcessingTime(long bucketProcessingTimeMs) {
         }
     }
 
+    public void finishReporting() {
+        // Don't flush if current timing stats are identical to the persisted ones
+        if (currentTimingStats.equals(persistedTimingStats)) {
+            return;
+        }
+        flush();
+    }
+
     public void flush() {
         persistedTimingStats = new TimingStats(currentTimingStats);
         bulkResultsPersister.persistTimingStats(persistedTimingStats);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/AutodetectResultProcessor.java

@@ -134,7 +134,7 @@ public void process() {
 
             try {
                 if (processKilled == false) {
-                    timingStatsReporter.flush();
+                    timingStatsReporter.finishReporting();
                     bulkResultsPersister.executeRequest();
                 }
             } catch (Exception e) {

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/persistence/TimingStatsReporterTests.java

@@ -14,6 +14,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyZeroInteractions;
 
 public class TimingStatsReporterTests extends ESTestCase {
@@ -76,6 +77,24 @@ public void testFlush() {
         inOrder.verifyNoMoreInteractions();
     }
 
+    public void testFinishReportingNoChange() {
+        TimingStatsReporter reporter = new TimingStatsReporter(new TimingStats(JOB_ID), bulkResultsPersister);
+
+        reporter.finishReporting();
+
+        verifyZeroInteractions(bulkResultsPersister);
+    }
+
+    public void testFinishReportingWithChange() {
+        TimingStatsReporter reporter = new TimingStatsReporter(new TimingStats(JOB_ID), bulkResultsPersister);
+
+        reporter.reportBucketProcessingTime(10);
+
+        reporter.finishReporting();
+
+        verify(bulkResultsPersister).persistTimingStats(new TimingStats(JOB_ID, 1, 10.0, 10.0, 10.0, 10.0));
+    }
+
     public void testTimingStatsDifferSignificantly() {
         assertThat(
             TimingStatsReporter.differSignificantly(

x-pack/plugin/src/test/resources/rest-api-spec/test/ml/index_layout.yml

@@ -124,7 +124,7 @@ setup:
         Authorization: "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" # run as x_pack_rest_user, i.e. the test setup superuser
       count:
         index: .ml-anomalies-shared
-  - match: {count: 8}
+  - match: {count: 6}
 
   - do:
       headers:
@@ -138,7 +138,7 @@ setup:
                 term:
                   job_id: index-layout-job
 
-  - match: {count: 4}
+  - match: {count: 3}
 
   - do:
       headers:
@@ -152,7 +152,7 @@ setup:
                 term:
                   job_id: index-layout-job
 
-  - match: {count: 4}
+  - match: {count: 3}
 
   - do:
       headers:
@@ -166,7 +166,7 @@ setup:
                 term:
                   job_id: index-layout-job2
 
-  - match: {count: 4}
+  - match: {count: 3}
 
   - do:
       headers:
@@ -179,7 +179,7 @@ setup:
               filter:
                 term:
                   job_id: index-layout-job2
-  - match: {count: 4}
+  - match: {count: 3}
 
   - do:
       headers:
@@ -236,7 +236,7 @@ setup:
         Authorization: "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" # run as x_pack_rest_user, i.e. the test setup superuser
       count:
         index: .ml-anomalies-shared
-  - match: {count: 4}
+  - match: {count: 3}
 
 
   - do:
@@ -251,7 +251,7 @@ setup:
                 term:
                   job_id: index-layout-job2
 
-  - match: {count: 4}
+  - match: {count: 3}
 
   - do:
       headers:
@@ -265,7 +265,7 @@ setup:
                 term:
                   job_id: index-layout-job2
 
-  - match: {count: 4}
+  - match: {count: 3}
 
   - do:
       ml.delete_job: