[ML] Fix race condition between job opening and feature reset

There was a point during the job opening sequence where performing
a feature reset could hang.

This happened when the kill request issued by feature reset was
executed after the job's persistent task was assigned but before
the job's native process was started. The persistent task was
incorrectly left running in this situation, yet the job opening
sequence was aborted which meant the subsequent close request
issued by feature reset would wait for a very long time for the
persistent task to disappear.

The fix is to make the kill process request cancel the persistent
task consistently based on its request parameters and not on the
current state of the task.

Fixes elastic#74141

Author: droberts195

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

@@ -52,10 +52,14 @@ boolean shouldBeKilled() {
     }
 
     void killIt() {
+        // This method should only be called for a process that's known to be connected
+        if (autodetectCommunicator == null) {
+            throw new IllegalArgumentException("Unable to kill job as its communicator is not connected");
+        }
         if (latestKillRequest == null) {
             throw new IllegalArgumentException("Unable to kill job as previous request is not completed");
         }
-        latestKillRequest.kill();
+        latestKillRequest.killConnectedProcess(false);
     }
 
     ProcessStateName getState() {
@@ -128,10 +132,27 @@ KillBuilder setShouldFinalizeJob(boolean shouldFinalizeJob) {
         }
 
         void kill() {
-            if (autodetectCommunicator == null) {
+            if (autodetectCommunicator != null) {
+                killConnectedProcess(finish);
+            } else {
                 latestKillRequest = this;
-                return;
+                // Killing a connected process would also complete the persistent task if `finish` was true,
+                // so we should do the same here even though the process wasn't yet connected at the time of
+                // the kill
+                if (finish) {
+                    jobTask.markAsCompleted();
+                }
             }
+        }
+
+        /**
+         * @param finish This argument overrides the member variable of the same name. Sometimes this method is called
+         *               immediately after a process is started, if it was requested to be killed before it started,
+         *               and in this situation the persistent task has already been completed if this was desired.
+         */
+        private void killConnectedProcess(boolean finish) {
+            assert autodetectCommunicator != null;
+
             String jobId = jobTask.getJobId();
 
             if (silent == false) {