[7.4][ML] Avoid marking data frame analytics task completed twice (elastic#46721) (elastic#46725)

When the stop API is called while the task is running there is
a chance the task gets marked completed twice. This may cause
undesired side effects, like indexing the progress document a second
time after the stop API has returned (the cause for elastic#46705).

This commit adds a check that the task has not been completed before
proceeding to mark it so. In addition, when we update the task's state
we could get some warnings that the task was missing if the stop API
has been called in the meantime. We now check the errors are
`ResourceNotFoundException` and ignore them if so.

Closes elastic#46705

Backports elastic#46721

Author: dimitris-athanasiou

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStartDataFrameAnalyticsAction.java

@@ -458,7 +458,12 @@ protected void onCancelled() {
 
         @Override
         public void markAsCompleted() {
-            persistProgress(() -> super.markAsCompleted());
+            // It is possible that the stop API has been called in the meantime and that
+            // may also cause this method to be called. We check whether we have already
+            // been marked completed to avoid doing it twice.
+            if (isCompleted() == false) {
+                persistProgress(() -> super.markAsCompleted());
+            }
         }
 
         @Override

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

@@ -7,6 +7,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
@@ -79,7 +80,13 @@ public void execute(DataFrameAnalyticsTask task, DataFrameAnalyticsState current
                     case STARTED:
                         task.updatePersistentTaskState(reindexingState, ActionListener.wrap(
                             updatedTask -> reindexingStateListener.onResponse(config),
-                            reindexingStateListener::onFailure));
+                            error -> {
+                                if (error instanceof ResourceNotFoundException) {
+                                    // The task has been stopped
+                                } else {
+                                    reindexingStateListener.onFailure(error);
+                                }
+                            }));
                         break;
                     // The task has fully reindexed the documents and we should continue on with our analyses
                     case ANALYZING:
@@ -222,7 +229,13 @@ private void startAnalytics(DataFrameAnalyticsTask task, DataFrameAnalyticsConfi
                                 task.markAsCompleted();
                             }
                         }),
-                    error -> task.updateState(DataFrameAnalyticsState.FAILED, error.getMessage())
+                    error -> {
+                        if (error instanceof ResourceNotFoundException) {
+                            // Task has stopped
+                        } else {
+                            task.updateState(DataFrameAnalyticsState.FAILED, error.getMessage());
+                        }
+                    }
                 ));
             },
             error -> task.updateState(DataFrameAnalyticsState.FAILED, error.getMessage())