Validate if job index already exists after job validation.

Closes elastic/elasticsearch#594

Original commit: elastic/x-pack-elasticsearch@b3b7b086a7

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/manager/JobManager.java

@@ -194,10 +194,11 @@ public class JobManager extends AbstractComponent {
 
             @Override
             public ClusterState execute(ClusterState currentState) throws Exception {
+                ClusterState cs = updateClusterState(job, request.isOverwrite(), currentState);
                 if (currentState.metaData().index(AnomalyDetectorsIndex.getJobIndexName(job.getIndexName())) != null) {
                     throw new ResourceAlreadyExistsException(Messages.getMessage(Messages.JOB_INDEX_ALREADY_EXISTS, job.getIndexName()));
                 }
-                return updateClusterState(job, request.isOverwrite(), currentState);
+                return cs;
             }
 
         });

elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/manager/JobManagerTests.java

@@ -29,8 +29,6 @@ import org.elasticsearch.xpack.prelert.job.persistence.JobProvider;
 import org.elasticsearch.xpack.prelert.job.persistence.JobResultsPersister;
 import org.elasticsearch.xpack.prelert.job.persistence.QueryPage;
 import org.junit.Before;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
 
 import java.util.Arrays;
 import java.util.HashSet;
@@ -167,15 +165,12 @@ public class JobManagerTests extends ESTestCase {
                 .fPut(AnomalyDetectorsIndex.getJobIndexName("my-special-place"), indexMetaData).build();
 
         ClusterState cs = ClusterState.builder(new ClusterName("_name"))
-                .metaData(MetaData.builder().indices(indexMap)).build();
+                .metaData(MetaData.builder().putCustom(PrelertMetadata.TYPE, PrelertMetadata.PROTO).indices(indexMap)).build();
 
-        doAnswer(new Answer<Void>() {
-            @Override
-            public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
+        doAnswer(invocationOnMock -> {
             AckedClusterStateUpdateTask<Boolean> task = (AckedClusterStateUpdateTask<Boolean>) invocationOnMock.getArguments()[1];
             task.execute(cs);
             return null;
-            }
         }).when(clusterService).submitStateUpdateTask(eq("put-job-foo"), any(AckedClusterStateUpdateTask.class));
 
         ResourceAlreadyExistsException e = expectThrows(ResourceAlreadyExistsException.class, () -> jobManager.putJob(request,

elasticsearch/src/test/resources/rest-api-spec/test/jobs_crud.yaml

@@ -83,6 +83,40 @@
                 "time_format":"yyyy-MM-dd HH:mm:ssX"
             }
           }
+  - do:
+      catch: /The job cannot be created with the Id 'farequote'. The Id is already used./
+      xpack.prelert.put_job:
+        body:  >
+          {
+            "job_id":"farequote",
+            "description":"Analysis of response time by airline",
+            "analysis_config" : {
+                "bucket_span":3600,
+                "detectors" :[{"function":"metric","field_name":"responsetime","by_field_name":"airline"}]
+            },
+            "data_description" : {
+                "field_delimiter":",",
+                "time_field":"time",
+                "time_format":"yyyy-MM-dd HH:mm:ssX"
+            }
+          }
+  - do:
+      catch: param
+      xpack.prelert.put_job:
+        body:  >
+          {
+            "job_id":"farequote",
+            "description":"Analysis of response time by airline",
+            "analysis_config" : {
+                "bucket_span":3600,
+                "detectors" :[{"function":"metric","field_name":"responsetime","by_field_name":"airline"}]
+            },
+            "data_description" : {
+                "field_delimiter":",",
+                "time_field":"time",
+                "time_format":"yyyy-MM-dd HH:mm:ssX"
+            }
+          }
 
 ---
 "Test delete job that is referred by a scheduler":