[ML] Improve reason why datafeed is stopped.

Original commit: elastic/x-pack-elasticsearch@2055a64689

plugin/src/main/java/org/elasticsearch/xpack/ml/action/StartDatafeedAction.java

@@ -262,18 +262,18 @@ public class StartDatafeedAction
 
         @Override
         protected void onCancelled() {
-            stop();
+            stop(getReasonCancelled());
         }
 
-        public void stop() {
-            stop(TimeValue.timeValueSeconds(20));
+        public void stop(String reason) {
+            stop(reason, TimeValue.timeValueSeconds(20));
         }
 
-        public void stop(TimeValue timeout) {
+        public void stop(String reason, TimeValue timeout) {
             if (holder == null) {
                 throw new IllegalStateException("task cancel ran before datafeed runner assigned the holder");
             }
-            holder.stop("cancel", timeout, null);
+            holder.stop(reason, timeout, null);
         }
     }
 

plugin/src/main/java/org/elasticsearch/xpack/ml/action/StopDatafeedAction.java

@@ -252,7 +252,7 @@ public class StopDatafeedAction
 
         @Override
         protected void taskOperation(Request request, StartDatafeedAction.DatafeedTask task, ActionListener<Response> listener) {
-            task.stop(request.getTimeout());
+            task.stop("stop_datafeed_api", request.getTimeout());
             listener.onResponse(new Response(true));
         }
 

plugin/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedJobRunner.java

@@ -271,18 +271,21 @@ public class DatafeedJobRunner extends AbstractComponent {
             if (datafeedJob.stop()) {
                 boolean acquired = false;
                 try {
+                    logger.info("[{}] try lock [{}] to stop datafeed [{}] for job [{}]...", source, timeout, datafeed.getId(),
+                            datafeed.getJobId());
                     acquired = datafeedJobLock.tryLock(timeout.millis(), TimeUnit.MILLISECONDS);
-                    logger.info("[{}] stopping datafeed [{}] for job [{}]...", source, datafeed.getId(), datafeed.getJobId());
+                } catch (InterruptedException e1) {
+                    Thread.currentThread().interrupt();
+                } finally {
+                    logger.info("[{}] stopping datafeed [{}] for job [{}], acquired [{}]...", source, datafeed.getId(),
+                            datafeed.getJobId(), acquired);
                     FutureUtils.cancel(future);
                     auditor.info(datafeed.getJobId(), Messages.getMessage(Messages.JOB_AUDIT_DATAFEED_STOPPED));
+                    handler.accept(e);
                     logger.info("[{}] datafeed [{}] for job [{}] has been stopped", source, datafeed.getId(), datafeed.getJobId());
                     if (autoCloseJob) {
                         closeJob();
                     }
-                } catch (InterruptedException e1) {
-                    Thread.currentThread().interrupt();
-                } finally {
-                    handler.accept(e);
                     if (acquired) {
                         datafeedJobLock.unlock();
                     }

plugin/src/test/java/org/elasticsearch/xpack/ml/datafeed/DatafeedJobRunnerTests.java

@@ -273,7 +273,7 @@ public class DatafeedJobRunnerTests extends ESTestCase {
 
         verify(threadPool, times(1)).executor(MachineLearning.DATAFEED_RUNNER_THREAD_POOL_NAME);
         if (cancelled) {
-            task.stop();
+            task.stop("test");
             verify(handler).accept(null);
         } else {
             verify(client).execute(same(PostDataAction.INSTANCE), eq(createExpectedPostDataRequest("job_id")));