[ML] Ensure phase progress may only increase (#56339)

Due to multi-threading it is possible that phase progress
updates written from the c++ process arrive reordered.
We can address this by ensuring that progress may only increase.

Closes #56282

Author: dimitris-athanasiou

x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/RegressionIT.java

@@ -169,8 +169,7 @@ public void testWithOnlyTrainingRowsAndTrainingPercentIsHundred() throws Excepti
             "Started writing results",
             "Finished analysis");
     }
-    
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/56282")
+
     public void testWithOnlyTrainingRowsAndTrainingPercentIsFifty() throws Exception {
         initialize("regression_only_training_data_and_training_percent_is_50");
         String predictedClassField = DEPENDENT_VARIABLE_FIELD + "_prediction";

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/stats/ProgressTracker.java

@@ -53,27 +53,35 @@ public ProgressTracker(List<PhaseProgress> phaseProgresses) {
     }
 
     public void updateReindexingProgress(int progressPercent) {
-        progressPercentPerPhase.put(REINDEXING, progressPercent);
+        updatePhase(REINDEXING, progressPercent);
     }
 
     public int getReindexingProgressPercent() {
         return progressPercentPerPhase.get(REINDEXING);
     }
 
     public void updateLoadingDataProgress(int progressPercent) {
-        progressPercentPerPhase.put(LOADING_DATA, progressPercent);
+        updatePhase(LOADING_DATA, progressPercent);
+    }
+
+    public int getLoadingDataProgressPercent() {
+        return progressPercentPerPhase.get(LOADING_DATA);
     }
 
     public void updateWritingResultsProgress(int progressPercent) {
-        progressPercentPerPhase.put(WRITING_RESULTS, progressPercent);
+        updatePhase(WRITING_RESULTS, progressPercent);
     }
 
     public int getWritingResultsProgressPercent() {
         return progressPercentPerPhase.get(WRITING_RESULTS);
     }
 
     public void updatePhase(PhaseProgress phase) {
-        progressPercentPerPhase.computeIfPresent(phase.getPhase(), (k, v) -> phase.getProgressPercent());
+        updatePhase(phase.getPhase(), phase.getProgressPercent());
+    }
+
+    private void updatePhase(String phase, int progress) {
+        progressPercentPerPhase.computeIfPresent(phase, (k, v) -> Math.max(v, progress));
     }
 
     public List<PhaseProgress> report() {

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/dataframe/stats/ProgressTrackerTests.java

@@ -79,4 +79,56 @@ public void testUpdatePhase_GivenUnknownPhase() {
         assertThat(phases.stream().map(PhaseProgress::getPhase).collect(Collectors.toList()),
             contains("reindexing", "loading_data", "foo", "writing_results"));
     }
+
+    public void testUpdateReindexingProgress_GivenLowerValueThanCurrentProgress() {
+        ProgressTracker progressTracker = ProgressTracker.fromZeroes(Collections.singletonList("foo"));
+
+        progressTracker.updateReindexingProgress(10);
+
+        progressTracker.updateReindexingProgress(11);
+        assertThat(progressTracker.getReindexingProgressPercent(), equalTo(11));
+
+        progressTracker.updateReindexingProgress(10);
+        assertThat(progressTracker.getReindexingProgressPercent(), equalTo(11));
+    }
+
+    public void testUpdateLoadingDataProgress_GivenLowerValueThanCurrentProgress() {
+        ProgressTracker progressTracker = ProgressTracker.fromZeroes(Collections.singletonList("foo"));
+
+        progressTracker.updateLoadingDataProgress(20);
+
+        progressTracker.updateLoadingDataProgress(21);
+        assertThat(progressTracker.getLoadingDataProgressPercent(), equalTo(21));
+
+        progressTracker.updateLoadingDataProgress(20);
+        assertThat(progressTracker.getLoadingDataProgressPercent(), equalTo(21));
+    }
+
+    public void testUpdateWritingResultsProgress_GivenLowerValueThanCurrentProgress() {
+        ProgressTracker progressTracker = ProgressTracker.fromZeroes(Collections.singletonList("foo"));
+
+        progressTracker.updateWritingResultsProgress(30);
+
+        progressTracker.updateWritingResultsProgress(31);
+        assertThat(progressTracker.getWritingResultsProgressPercent(), equalTo(31));
+
+        progressTracker.updateWritingResultsProgress(30);
+        assertThat(progressTracker.getWritingResultsProgressPercent(), equalTo(31));
+    }
+
+    public void testUpdatePhase_GivenLowerValueThanCurrentProgress() {
+        ProgressTracker progressTracker = ProgressTracker.fromZeroes(Collections.singletonList("foo"));
+
+        progressTracker.updatePhase(new PhaseProgress("foo", 40));
+
+        progressTracker.updatePhase(new PhaseProgress("foo", 41));
+        assertThat(getProgressForPhase(progressTracker, "foo"), equalTo(41));
+
+        progressTracker.updatePhase(new PhaseProgress("foo", 40));
+        assertThat(getProgressForPhase(progressTracker, "foo"), equalTo(41));
+    }
+
+    private static int getProgressForPhase(ProgressTracker progressTracker, String phase) {
+        return progressTracker.report().stream().filter(p -> p.getPhase().equals(phase)).findFirst().get().getProgressPercent();
+    }
 }