Synchronize search of existing tasks and mark started

Author: dimitris-athanasiou

client/rest-high-level/src/test/java/org/elasticsearch/client/MLRequestConvertersTests.java

@@ -162,11 +162,18 @@ public void testDeleteJob() {
         Request request = MLRequestConverters.deleteJob(deleteJobRequest);
         assertEquals(HttpDelete.METHOD_NAME, request.getMethod());
         assertEquals("/_xpack/ml/anomaly_detectors/" + jobId, request.getEndpoint());
-        assertEquals(Boolean.toString(false), request.getParameters().get("force"));
+        assertNull(request.getParameters().get("force"));
+        assertNull(request.getParameters().get("wait_for_completion"));
 
+        deleteJobRequest = new DeleteJobRequest(jobId);
         deleteJobRequest.setForce(true);
         request = MLRequestConverters.deleteJob(deleteJobRequest);
         assertEquals(Boolean.toString(true), request.getParameters().get("force"));
+
+        deleteJobRequest = new DeleteJobRequest(jobId);
+        deleteJobRequest.setWaitForCompletion(false);
+        request = MLRequestConverters.deleteJob(deleteJobRequest);
+        assertEquals(Boolean.toString(false), request.getParameters().get("wait_for_completion"));
     }
 
     public void testFlushJob() throws Exception {

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

@@ -12,7 +12,17 @@
 
 public class JobDeletionTask extends Task {
 
+    private volatile boolean started;
+
     public JobDeletionTask(long id, String type, String action, String description, TaskId parentTask, Map<String, String> headers) {
         super(id, type, action, description, parentTask, headers);
     }
+
+    public void start() {
+        started = true;
+    }
+
+    public boolean isStarted() {
+        return started;
+    }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/job/config/MlFilterTests.java

@@ -6,9 +6,7 @@
 package org.elasticsearch.xpack.core.ml.job.config;
 
 import com.carrotsearch.randomizedtesting.generators.CodepointSetGenerator;
-import org.apache.http.util.EntityUtils;
 import org.elasticsearch.ElasticsearchStatusException;
-import org.elasticsearch.client.Response;
 import org.elasticsearch.common.io.stream.Writeable.Reader;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
@@ -19,8 +17,6 @@
 import java.util.List;
 import java.util.SortedSet;
 import java.util.TreeSet;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;

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

@@ -21,7 +21,6 @@
 import org.elasticsearch.xpack.ml.MachineLearning;
 import org.elasticsearch.xpack.test.rest.XPackRestTestHelper;
 import org.junit.After;
-import org.junit.Ignore;
 
 import java.io.IOException;
 import java.util.Locale;
@@ -562,9 +561,8 @@ public void testMultiIndexDelete() throws Exception {
                 client().performRequest(new Request("GET", MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId + "/_stats")));
     }
 
-    @Ignore
     public void testDelete_multipleRequest() throws Exception {
-        String jobId = "delete-job-mulitple-times";
+        String jobId = "delete-job-multiple-times";
         createFarequoteJob(jobId);
 
         ConcurrentMapLong<Response> responses = ConcurrentCollections.newConcurrentMapLong();
@@ -575,8 +573,8 @@ public void testDelete_multipleRequest() throws Exception {
         AtomicReference<ResponseException> recreationException = new AtomicReference<>();
 
         Runnable deleteJob = () -> {
+            boolean forceDelete = randomBoolean();
             try {
-                boolean forceDelete = randomBoolean();
                 String url = MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId;
                 if (forceDelete) {
                     url += "?force=true";
@@ -597,6 +595,7 @@ public void testDelete_multipleRequest() throws Exception {
                 } catch (ResponseException re) {
                     recreationException.set(re);
                 } catch (IOException e) {
+                    logger.error("Error trying to recreate the job", e);
                     ioe.set(e);
                 }
             }
@@ -606,14 +605,14 @@ public void testDelete_multipleRequest() throws Exception {
         // the other to complete. This is difficult to schedule but
         // hopefully it will happen in CI
         int numThreads = 5;
-        Thread [] threads = new Thread[numThreads];
-        for (int i=0; i<numThreads; i++) {
+        Thread[] threads = new Thread[numThreads];
+        for (int i = 0; i < numThreads; i++) {
             threads[i] = new Thread(deleteJob);
         }
-        for (int i=0; i<numThreads; i++) {
+        for (int i = 0; i < numThreads; i++) {
             threads[i].start();
         }
-        for (int i=0; i<numThreads; i++) {
+        for (int i = 0; i < numThreads; i++) {
             threads[i].join();
         }
 

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

@@ -72,6 +72,7 @@
 import org.elasticsearch.xpack.core.ml.job.messages.Messages;
 import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
 import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
+import org.elasticsearch.xpack.core.ml.job.persistence.JobDeletionTask;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.CategorizerState;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSnapshot;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.Quantiles;
@@ -143,11 +144,17 @@ protected void masterOperation(Task task, DeleteJobAction.Request request, Clust
         ParentTaskAssigningClient parentTaskClient = new ParentTaskAssigningClient(client, taskId);
 
         // Check if there is a deletion task for this job already and if yes wait for it to complete
-        Optional<Task> existingTask = taskManager.getTasks().values().stream().filter(searchedTask -> task.getDescription().equals(
-                searchedTask.getDescription()) && task.getId() != searchedTask.getId()).findFirst();
-        if (existingTask.isPresent()) {
-            logger.debug("Found existing deletion task for job [{}]", request.getJobId());
-            TaskId existingTaskId = new TaskId(clusterService.localNode().getId(), existingTask.get().getId());
+        Optional<Task> existingStartedTask;
+        synchronized (this) {
+            existingStartedTask = findExistingStartedTask(task);
+            if (existingStartedTask.isPresent() == false) {
+                ((JobDeletionTask) task).start();
+            }
+        }
+        if (existingStartedTask.isPresent()) {
+            logger.debug("[{}] Deletion task [{}] will wait for existing deletion task [{}]", request.getJobId(), task.getId(),
+                    existingStartedTask.get().getId());
+            TaskId existingTaskId = new TaskId(clusterService.localNode().getId(), existingStartedTask.get().getId());
             waitForExistingTaskToComplete(parentTaskClient, request.getJobId(), existingTaskId, listener);
             return;
         }
@@ -170,6 +177,13 @@ protected void masterOperation(Task task, DeleteJobAction.Request request, Clust
         markJobAsDeleting(request.getJobId(), markAsDeletingListener, request.isForce());
     }
 
+    private Optional<Task> findExistingStartedTask(Task currentTask) {
+        return taskManager.getTasks().values().stream().filter(filteredTask ->
+                currentTask.getDescription().equals(filteredTask.getDescription()) &&
+                        currentTask.getId() != filteredTask.getId()
+                        && ((JobDeletionTask) filteredTask).isStarted()).findFirst();
+    }
+
     private void waitForExistingTaskToComplete(ParentTaskAssigningClient parentTaskClient, String jobId, TaskId taskId,
                                                ActionListener<AcknowledgedResponse> listener) {
         threadPool.generic().execute(new AbstractRunnable() {
@@ -203,6 +217,7 @@ public void onResponse(GetTaskResponse getTaskResponse) {
                                                 + taskErrorBytes.utf8ToString()));
                                     }
                                 } else {
+                                    logger.debug("[{}] Finished waiting for completion of task [{}]", jobId, taskId);
                                     listener.onResponse(new AcknowledgedResponse(true));
                                 }
                             }