[ML] Set df-analytics task state to failed when appropriate (#43880)

This introduces a `failed` state to which the data frame analytics
persistent task is set to when something unexpected fails. It could
be the process crashing, the results processor hitting some error,
etc. The failure message is then captured and set on the task state.
From there, it becomes available via the _stats API as `failure_reason`.

The df-analytics stop API now has a `force` boolean parameter. This allows
the user to call it for a failed task in order to reset it to `stopped` after
we have ensured the failure has been communicated to the user.

This commit also adds the analytics version in the persistent task
params as this allows us to prevent tasks to run on unsuitable nodes in
the future.

client/rest-high-level/src/main/java/org/elasticsearch/client/MLRequestConverters.java

@@ -676,6 +676,9 @@ final class MLRequestConverters {
             params.putParam(
                 StopDataFrameAnalyticsRequest.ALLOW_NO_MATCH.getPreferredName(), Boolean.toString(stopRequest.getAllowNoMatch()));
         }
+        if (stopRequest.getForce() != null) {
+            params.putParam(StopDataFrameAnalyticsRequest.FORCE.getPreferredName(), Boolean.toString(stopRequest.getForce()));
+        }
         request.addParameters(params.asMap());
         return request;
     }

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/StopDataFrameAnalyticsRequest.java

@@ -31,10 +31,12 @@ import java.util.Optional;
 public class StopDataFrameAnalyticsRequest implements Validatable {
 
     public static final ParseField ALLOW_NO_MATCH = new ParseField("allow_no_match");
+    public static final ParseField FORCE = new ParseField("force");
 
     private final String id;
-    private TimeValue timeout;
     private Boolean allowNoMatch;
+    private Boolean force;
+    private TimeValue timeout;
 
     public StopDataFrameAnalyticsRequest(String id) {
         this.id = id;
@@ -62,6 +64,15 @@ public class StopDataFrameAnalyticsRequest implements Validatable {
         return this;
     }
 
+    public Boolean getForce() {
+        return force;
+    }
+
+    public StopDataFrameAnalyticsRequest setForce(boolean force) {
+        this.force = force;
+        return this;
+    }
+
     @Override
     public Optional<ValidationException> validate() {
         if (id == null) {
@@ -78,11 +89,12 @@ public class StopDataFrameAnalyticsRequest implements Validatable {
         StopDataFrameAnalyticsRequest other = (StopDataFrameAnalyticsRequest) o;
         return Objects.equals(id, other.id)
             && Objects.equals(timeout, other.timeout)
-            && Objects.equals(allowNoMatch, other.allowNoMatch);
+            && Objects.equals(allowNoMatch, other.allowNoMatch)
+            && Objects.equals(force, other.force);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, timeout, allowNoMatch);
+        return Objects.hash(id, timeout, allowNoMatch, force);
     }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/dataframe/DataFrameAnalyticsStats.java

@@ -41,6 +41,7 @@ public class DataFrameAnalyticsStats {
 
     static final ParseField ID = new ParseField("id");
     static final ParseField STATE = new ParseField("state");
+    static final ParseField FAILURE_REASON = new ParseField("failure_reason");
     static final ParseField PROGRESS_PERCENT = new ParseField("progress_percent");
     static final ParseField NODE = new ParseField("node");
     static final ParseField ASSIGNMENT_EXPLANATION = new ParseField("assignment_explanation");
@@ -50,9 +51,10 @@ public class DataFrameAnalyticsStats {
             args -> new DataFrameAnalyticsStats(
                 (String) args[0],
                 (DataFrameAnalyticsState) args[1],
-                (Integer) args[2],
-                (NodeAttributes) args[3],
-                (String) args[4]));
+                (String) args[2],
+                (Integer) args[3],
+                (NodeAttributes) args[4],
+                (String) args[5]));
 
     static {
         PARSER.declareString(constructorArg(), ID);
@@ -62,6 +64,7 @@ public class DataFrameAnalyticsStats {
             }
             throw new IllegalArgumentException("Unsupported token [" + p.currentToken() + "]");
         }, STATE, ObjectParser.ValueType.STRING);
+        PARSER.declareString(optionalConstructorArg(), FAILURE_REASON);
         PARSER.declareInt(optionalConstructorArg(), PROGRESS_PERCENT);
         PARSER.declareObject(optionalConstructorArg(), NodeAttributes.PARSER, NODE);
         PARSER.declareString(optionalConstructorArg(), ASSIGNMENT_EXPLANATION);
@@ -69,14 +72,17 @@ public class DataFrameAnalyticsStats {
 
     private final String id;
     private final DataFrameAnalyticsState state;
+    private final String failureReason;
     private final Integer progressPercent;
     private final NodeAttributes node;
     private final String assignmentExplanation;
 
-    public DataFrameAnalyticsStats(String id, DataFrameAnalyticsState state, @Nullable Integer progressPercent,
-                                   @Nullable NodeAttributes node, @Nullable String assignmentExplanation) {
+    public DataFrameAnalyticsStats(String id, DataFrameAnalyticsState state, @Nullable String failureReason,
+                                   @Nullable Integer progressPercent, @Nullable NodeAttributes node,
+                                   @Nullable String assignmentExplanation) {
         this.id = id;
         this.state = state;
+        this.failureReason = failureReason;
         this.progressPercent = progressPercent;
         this.node = node;
         this.assignmentExplanation = assignmentExplanation;
@@ -90,6 +96,10 @@ public class DataFrameAnalyticsStats {
         return state;
     }
 
+    public String getFailureReason() {
+        return failureReason;
+    }
+
     public Integer getProgressPercent() {
         return progressPercent;
     }
@@ -110,6 +120,7 @@ public class DataFrameAnalyticsStats {
         DataFrameAnalyticsStats other = (DataFrameAnalyticsStats) o;
         return Objects.equals(id, other.id)
             && Objects.equals(state, other.state)
+            && Objects.equals(failureReason, other.failureReason)
             && Objects.equals(progressPercent, other.progressPercent)
             && Objects.equals(node, other.node)
             && Objects.equals(assignmentExplanation, other.assignmentExplanation);
@@ -117,7 +128,7 @@ public class DataFrameAnalyticsStats {
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, state, progressPercent, node, assignmentExplanation);
+        return Objects.hash(id, state, failureReason, progressPercent, node, assignmentExplanation);
     }
 
     @Override
@@ -125,6 +136,7 @@ public class DataFrameAnalyticsStats {
         return new ToStringBuilder(getClass())
             .add("id", id)
             .add("state", state)
+            .add("failureReason", failureReason)
             .add("progressPercent", progressPercent)
             .add("node", node)
             .add("assignmentExplanation", assignmentExplanation)

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

@@ -758,11 +758,15 @@ public class MLRequestConvertersTests extends ESTestCase {
     public void testStopDataFrameAnalytics_WithParams() {
         StopDataFrameAnalyticsRequest stopRequest = new StopDataFrameAnalyticsRequest(randomAlphaOfLength(10))
             .setTimeout(TimeValue.timeValueMinutes(1))
-            .setAllowNoMatch(false);
+            .setAllowNoMatch(false)
+            .setForce(true);
         Request request = MLRequestConverters.stopDataFrameAnalytics(stopRequest);
         assertEquals(HttpPost.METHOD_NAME, request.getMethod());
         assertEquals("/_ml/data_frame/analytics/" + stopRequest.getId() + "/_stop", request.getEndpoint());
-        assertThat(request.getParameters(), allOf(hasEntry("timeout", "1m"), hasEntry("allow_no_match", "false")));
+        assertThat(request.getParameters(), allOf(
+            hasEntry("timeout", "1m"),
+            hasEntry("allow_no_match", "false"),
+            hasEntry("force", "true")));
         assertNull(request.getEntity());
     }
 

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

@@ -1359,6 +1359,7 @@ public class MachineLearningIT extends ESRestHighLevelClientTestCase {
         DataFrameAnalyticsStats stats = statsResponse.getAnalyticsStats().get(0);
         assertThat(stats.getId(), equalTo(configId));
         assertThat(stats.getState(), equalTo(DataFrameAnalyticsState.STOPPED));
+        assertNull(stats.getFailureReason());
         assertNull(stats.getProgressPercent());
         assertNull(stats.getNode());
         assertNull(stats.getAssignmentExplanation());

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

@@ -3110,6 +3110,7 @@ public class MlClientDocumentationIT extends ESRestHighLevelClientTestCase {
         {
             // tag::stop-data-frame-analytics-request
             StopDataFrameAnalyticsRequest request = new StopDataFrameAnalyticsRequest("my-analytics-config"); // <1>
+            request.setForce(false); // <2>
             // end::stop-data-frame-analytics-request
 
             // tag::stop-data-frame-analytics-execute

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/dataframe/DataFrameAnalyticsStatsTests.java

@@ -43,6 +43,7 @@ public class DataFrameAnalyticsStatsTests extends ESTestCase {
         return new DataFrameAnalyticsStats(
             randomAlphaOfLengthBetween(1, 10),
             randomFrom(DataFrameAnalyticsState.values()),
+            randomBoolean() ? null : randomAlphaOfLength(10),
             randomBoolean() ? null : randomIntBetween(0, 100),
             randomBoolean() ? null : NodeAttributesTests.createRandom(),
             randomBoolean() ? null : randomAlphaOfLengthBetween(1, 20));
@@ -52,6 +53,9 @@ public class DataFrameAnalyticsStatsTests extends ESTestCase {
         builder.startObject();
         builder.field(DataFrameAnalyticsStats.ID.getPreferredName(), stats.getId());
         builder.field(DataFrameAnalyticsStats.STATE.getPreferredName(), stats.getState().value());
+        if (stats.getFailureReason() != null) {
+            builder.field(DataFrameAnalyticsStats.FAILURE_REASON.getPreferredName(), stats.getFailureReason());
+        }
         if (stats.getProgressPercent() != null) {
             builder.field(DataFrameAnalyticsStats.PROGRESS_PERCENT.getPreferredName(), stats.getProgressPercent());
         }

docs/java-rest/high-level/ml/stop-data-frame-analytics.asciidoc

@@ -19,6 +19,7 @@ A +{request}+ object requires a {dataframe-analytics-config} id.
 include-tagged::{doc-tests-file}[{api}-request]
 ---------------------------------------------------
 <1> Constructing a new stop request referencing an existing {dataframe-analytics-config}
+<2> Optionally used to stop a failed task
 
 include::../execution.asciidoc[]
 

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

@@ -6,9 +6,9 @@
 package org.elasticsearch.xpack.core.ml.action;
 
 import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.TaskOperationFailure;
 import org.elasticsearch.action.support.tasks.BaseTasksRequest;
 import org.elasticsearch.action.support.tasks.BaseTasksResponse;
@@ -158,16 +158,19 @@ public class GetDataFrameAnalyticsStatsAction extends ActionType<GetDataFrameAna
             private final String id;
             private final DataFrameAnalyticsState state;
             @Nullable
+            private final String failureReason;
+            @Nullable
             private final Integer progressPercentage;
             @Nullable
             private final DiscoveryNode node;
             @Nullable
             private final String assignmentExplanation;
 
-            public Stats(String id, DataFrameAnalyticsState state, @Nullable Integer progressPercentage,
+            public Stats(String id, DataFrameAnalyticsState state, @Nullable String failureReason, @Nullable Integer progressPercentage,
                          @Nullable DiscoveryNode node, @Nullable String assignmentExplanation) {
                 this.id = Objects.requireNonNull(id);
                 this.state = Objects.requireNonNull(state);
+                this.failureReason = failureReason;
                 this.progressPercentage = progressPercentage;
                 this.node = node;
                 this.assignmentExplanation = assignmentExplanation;
@@ -176,6 +179,7 @@ public class GetDataFrameAnalyticsStatsAction extends ActionType<GetDataFrameAna
             public Stats(StreamInput in) throws IOException {
                 id = in.readString();
                 state = DataFrameAnalyticsState.fromStream(in);
+                failureReason = in.readOptionalString();
                 progressPercentage = in.readOptionalInt();
                 node = in.readOptionalWriteable(DiscoveryNode::new);
                 assignmentExplanation = in.readOptionalString();
@@ -202,6 +206,9 @@ public class GetDataFrameAnalyticsStatsAction extends ActionType<GetDataFrameAna
             public XContentBuilder toUnwrappedXContent(XContentBuilder builder) throws IOException {
                 builder.field(DataFrameAnalyticsConfig.ID.getPreferredName(), id);
                 builder.field("state", state.toString());
+                if (failureReason != null) {
+                    builder.field("failure_reason", failureReason);
+                }
                 if (progressPercentage != null) {
                     builder.field("progress_percent", progressPercentage);
                 }
@@ -229,6 +236,7 @@ public class GetDataFrameAnalyticsStatsAction extends ActionType<GetDataFrameAna
             public void writeTo(StreamOutput out) throws IOException {
                 out.writeString(id);
                 state.writeTo(out);
+                out.writeOptionalString(failureReason);
                 out.writeOptionalInt(progressPercentage);
                 out.writeOptionalWriteable(node);
                 out.writeOptionalString(assignmentExplanation);
@@ -236,7 +244,7 @@ public class GetDataFrameAnalyticsStatsAction extends ActionType<GetDataFrameAna
 
             @Override
             public int hashCode() {
-                return Objects.hash(id, state, progressPercentage, node, assignmentExplanation);
+                return Objects.hash(id, state, failureReason, progressPercentage, node, assignmentExplanation);
             }
 
             @Override
@@ -250,6 +258,7 @@ public class GetDataFrameAnalyticsStatsAction extends ActionType<GetDataFrameAna
                 Stats other = (Stats) obj;
                 return Objects.equals(id, other.id)
                         && Objects.equals(this.state, other.state)
+                        && Objects.equals(this.failureReason, other.failureReason)
                         && Objects.equals(this.node, other.node)
                         && Objects.equals(this.assignmentExplanation, other.assignmentExplanation);
             }

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

@@ -6,9 +6,9 @@
 package org.elasticsearch.xpack.core.ml.action;
 
 import org.elasticsearch.Version;
-import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.MasterNodeRequest;
 import org.elasticsearch.client.ElasticsearchClient;
@@ -157,20 +157,32 @@ public class StartDataFrameAnalyticsAction extends ActionType<AcknowledgedRespon
         public static final Version VERSION_INTRODUCED = Version.V_7_3_0;
 
         public static ConstructingObjectParser<TaskParams, Void> PARSER = new ConstructingObjectParser<>(
-            MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME, true, a -> new TaskParams((String) a[0]));
+            MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME, true, a -> new TaskParams((String) a[0], (String) a[1]));
+
+        static {
+            PARSER.declareString(ConstructingObjectParser.constructorArg(), DataFrameAnalyticsConfig.ID);
+            PARSER.declareString(ConstructingObjectParser.constructorArg(), DataFrameAnalyticsConfig.VERSION);
+        }
 
         public static TaskParams fromXContent(XContentParser parser) {
             return PARSER.apply(parser, null);
         }
 
-        private String id;
+        private final String id;
+        private final Version version;
 
-        public TaskParams(String id) {
+        public TaskParams(String id, Version version) {
             this.id = Objects.requireNonNull(id);
+            this.version = Objects.requireNonNull(version);
+        }
+
+        private TaskParams(String id, String version) {
+            this(id, Version.fromString(version));
         }
 
         public TaskParams(StreamInput in) throws IOException {
             this.id = in.readString();
+            this.version = Version.readVersion(in);
         }
 
         public String getId() {
@@ -190,15 +202,31 @@ public class StartDataFrameAnalyticsAction extends ActionType<AcknowledgedRespon
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeString(id);
+            Version.writeVersion(version, out);
         }
 
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
             builder.field(DataFrameAnalyticsConfig.ID.getPreferredName(), id);
+            builder.field(DataFrameAnalyticsConfig.VERSION.getPreferredName(), version);
             builder.endObject();
             return builder;
         }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(id, version);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (o == this) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            TaskParams other = (TaskParams) o;
+            return Objects.equals(id, other.id) && Objects.equals(version, other.version);
+        }
     }
 
     public interface TaskMatcher {

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

@@ -5,9 +5,9 @@
  */
 package org.elasticsearch.xpack.core.ml.action;
 
-import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.tasks.BaseTasksRequest;
 import org.elasticsearch.action.support.tasks.BaseTasksResponse;
 import org.elasticsearch.client.ElasticsearchClient;
@@ -49,14 +49,17 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
 
     public static class Request extends BaseTasksRequest<Request> implements ToXContentObject {
 
-        public static final ParseField TIMEOUT = new ParseField("timeout");
         public static final ParseField ALLOW_NO_MATCH = new ParseField("allow_no_match");
+        public static final ParseField FORCE = new ParseField("force");
+        public static final ParseField TIMEOUT = new ParseField("timeout");
 
         private static final ObjectParser<Request, Void> PARSER = new ObjectParser<>(NAME, Request::new);
 
         static {
             PARSER.declareString((request, id) -> request.id = id, DataFrameAnalyticsConfig.ID);
             PARSER.declareString((request, val) -> request.setTimeout(TimeValue.parseTimeValue(val, TIMEOUT.getPreferredName())), TIMEOUT);
+            PARSER.declareBoolean(Request::setAllowNoMatch, ALLOW_NO_MATCH);
+            PARSER.declareBoolean(Request::setForce, FORCE);
         }
 
         public static Request parseRequest(String id, XContentParser parser) {
@@ -71,8 +74,9 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
         }
 
         private String id;
-        private Set<String> expandedIds = Collections.emptySet();
         private boolean allowNoMatch = true;
+        private boolean force;
+        private Set<String> expandedIds = Collections.emptySet();
 
         public Request(String id) {
             setId(id);
@@ -81,8 +85,9 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
         public Request(StreamInput in) throws IOException {
             super(in);
             id = in.readString();
-            expandedIds = new HashSet<>(Arrays.asList(in.readStringArray()));
             allowNoMatch = in.readBoolean();
+            force = in.readBoolean();
+            expandedIds = new HashSet<>(Arrays.asList(in.readStringArray()));
         }
 
         public Request() {}
@@ -95,6 +100,22 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
             return id;
         }
 
+        public boolean allowNoMatch() {
+            return allowNoMatch;
+        }
+
+        public void setAllowNoMatch(boolean allowNoMatch) {
+            this.allowNoMatch = allowNoMatch;
+        }
+
+        public boolean isForce() {
+            return force;
+        }
+
+        public void setForce(boolean force) {
+            this.force = force;
+        }
+
         @Nullable
         public Set<String> getExpandedIds() {
             return expandedIds;
@@ -104,14 +125,6 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
             this.expandedIds = Objects.requireNonNull(expandedIds);
         }
 
-        public boolean allowNoMatch() {
-            return allowNoMatch;
-        }
-
-        public void setAllowNoMatch(boolean allowNoMatch) {
-            this.allowNoMatch = allowNoMatch;
-        }
-
         @Override
         public ActionRequestValidationException validate() {
             return null;
@@ -121,8 +134,9 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeString(id);
-            out.writeStringArray(expandedIds.toArray(new String[0]));
             out.writeBoolean(allowNoMatch);
+            out.writeBoolean(force);
+            out.writeStringArray(expandedIds.toArray(new String[0]));
         }
 
         @Override
@@ -131,12 +145,13 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
                 .startObject()
                 .field(DataFrameAnalyticsConfig.ID.getPreferredName(), id)
                 .field(ALLOW_NO_MATCH.getPreferredName(), allowNoMatch)
+                .field(FORCE.getPreferredName(), force)
                 .endObject();
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(id, getTimeout(), expandedIds, allowNoMatch);
+            return Objects.hash(id, getTimeout(), allowNoMatch, force, expandedIds);
         }
 
         @Override
@@ -150,8 +165,9 @@ public class StopDataFrameAnalyticsAction extends ActionType<StopDataFrameAnalyt
             StopDataFrameAnalyticsAction.Request other = (StopDataFrameAnalyticsAction.Request) obj;
             return Objects.equals(id, other.id)
                 && Objects.equals(getTimeout(), other.getTimeout())
-                && Objects.equals(expandedIds, other.expandedIds)
-                && allowNoMatch == other.allowNoMatch;
+                && allowNoMatch == other.allowNoMatch
+                && force == other.force
+                && Objects.equals(expandedIds, other.expandedIds);
         }
 
         @Override

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

@@ -10,11 +10,12 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Locale;
 
 public enum DataFrameAnalyticsState implements Writeable {
 
-    STARTED, REINDEXING, ANALYZING, STOPPING, STOPPED;
+    STARTED, REINDEXING, ANALYZING, STOPPING, STOPPED, FAILED;
 
     public static DataFrameAnalyticsState fromString(String name) {
         return valueOf(name.trim().toUpperCase(Locale.ROOT));
@@ -33,4 +34,11 @@ public enum DataFrameAnalyticsState implements Writeable {
     public String toString() {
         return name().toLowerCase(Locale.ROOT);
     }
+
+    /**
+     * @return {@code true} if state matches any of the given {@code candidates}
+     */
+    public boolean isAnyOf(DataFrameAnalyticsState... candidates) {
+        return Arrays.stream(candidates).anyMatch(candidate -> this == candidate);
+    }
 }

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

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.core.ml.dataframe;
 
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -25,13 +26,15 @@ public class DataFrameAnalyticsTaskState implements PersistentTaskState {
 
     private static ParseField STATE = new ParseField("state");
     private static ParseField ALLOCATION_ID = new ParseField("allocation_id");
+    private static ParseField REASON = new ParseField("reason");
 
     private final DataFrameAnalyticsState state;
     private final long allocationId;
+    private final String reason;
 
     private static final ConstructingObjectParser<DataFrameAnalyticsTaskState, Void> PARSER =
             new ConstructingObjectParser<>(NAME, true,
-                a -> new DataFrameAnalyticsTaskState((DataFrameAnalyticsState) a[0], (long) a[1]));
+                a -> new DataFrameAnalyticsTaskState((DataFrameAnalyticsState) a[0], (long) a[1], (String) a[2]));
 
     static {
         PARSER.declareField(ConstructingObjectParser.constructorArg(), p -> {
@@ -41,6 +44,7 @@ public class DataFrameAnalyticsTaskState implements PersistentTaskState {
            throw new IllegalArgumentException("Unsupported token [" + p.currentToken() + "]");
         }, STATE, ObjectParser.ValueType.STRING);
         PARSER.declareLong(ConstructingObjectParser.constructorArg(), ALLOCATION_ID);
+        PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), REASON);
     }
 
     public static DataFrameAnalyticsTaskState fromXContent(XContentParser parser) {
@@ -51,20 +55,27 @@ public class DataFrameAnalyticsTaskState implements PersistentTaskState {
         }
     }
 
-    public DataFrameAnalyticsTaskState(DataFrameAnalyticsState state, long allocationId) {
+    public DataFrameAnalyticsTaskState(DataFrameAnalyticsState state, long allocationId, @Nullable String reason) {
         this.state = Objects.requireNonNull(state);
         this.allocationId = allocationId;
+        this.reason = reason;
     }
 
     public DataFrameAnalyticsTaskState(StreamInput in) throws IOException {
         this.state = DataFrameAnalyticsState.fromStream(in);
         this.allocationId = in.readLong();
+        this.reason = in.readOptionalString();
     }
 
     public DataFrameAnalyticsState getState() {
         return state;
     }
 
+    @Nullable
+    public String getReason() {
+        return reason;
+    }
+
     public boolean isStatusStale(PersistentTasksCustomMetaData.PersistentTask<?> task) {
         return allocationId != task.getAllocationId();
     }
@@ -78,6 +89,7 @@ public class DataFrameAnalyticsTaskState implements PersistentTaskState {
     public void writeTo(StreamOutput out) throws IOException {
         state.writeTo(out);
         out.writeLong(allocationId);
+        out.writeOptionalString(reason);
     }
 
     @Override
@@ -85,6 +97,9 @@ public class DataFrameAnalyticsTaskState implements PersistentTaskState {
         builder.startObject();
         builder.field(STATE.getPreferredName(), state.toString());
         builder.field(ALLOCATION_ID.getPreferredName(), allocationId);
+        if (reason != null) {
+            builder.field(REASON.getPreferredName(), reason);
+        }
         builder.endObject();
         return builder;
     }
@@ -95,11 +110,12 @@ public class DataFrameAnalyticsTaskState implements PersistentTaskState {
         if (o == null || getClass() != o.getClass()) return false;
         DataFrameAnalyticsTaskState that = (DataFrameAnalyticsTaskState) o;
         return allocationId == that.allocationId &&
-            state == that.state;
+            state == that.state &&
+            Objects.equals(reason, that.reason);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(state, allocationId);
+        return Objects.hash(state, allocationId, reason);
     }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/action/GetDataFrameAnalyticsStatsActionResponseTests.java

@@ -23,8 +23,9 @@ public class GetDataFrameAnalyticsStatsActionResponseTests extends AbstractWireS
         List<Response.Stats> analytics = new ArrayList<>(listSize);
         for (int j = 0; j < listSize; j++) {
             Integer progressPercentage = randomBoolean() ? null : randomIntBetween(0, 100);
+            String failureReason = randomBoolean() ? null : randomAlphaOfLength(10);
             Response.Stats stats = new Response.Stats(DataFrameAnalyticsConfigTests.randomValidId(),
-                randomFrom(DataFrameAnalyticsState.values()), progressPercentage, null, randomAlphaOfLength(20));
+                randomFrom(DataFrameAnalyticsState.values()), failureReason, progressPercentage, null, randomAlphaOfLength(20));
             analytics.add(stats);
         }
         return new Response(new QueryPage<>(analytics, analytics.size(), GetDataFrameAnalyticsAction.Response.RESULTS_FIELD));

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/action/StartDataFrameAnalyticsActionTaskParamsTests.java

@@ -0,0 +1,37 @@
+
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core.ml.action;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractSerializingTestCase;
+
+import java.io.IOException;
+
+public class StartDataFrameAnalyticsActionTaskParamsTests extends AbstractSerializingTestCase<StartDataFrameAnalyticsAction.TaskParams> {
+
+    @Override
+    protected StartDataFrameAnalyticsAction.TaskParams doParseInstance(XContentParser parser) throws IOException {
+        return StartDataFrameAnalyticsAction.TaskParams.fromXContent(parser);
+    }
+
+    @Override
+    protected StartDataFrameAnalyticsAction.TaskParams createTestInstance() {
+        return new StartDataFrameAnalyticsAction.TaskParams(randomAlphaOfLength(10), Version.CURRENT);
+    }
+
+    @Override
+    protected Writeable.Reader<StartDataFrameAnalyticsAction.TaskParams> instanceReader() {
+        return StartDataFrameAnalyticsAction.TaskParams::new;
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return true;
+    }
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/action/StopDataFrameAnalyticsRequestTests.java

@@ -24,6 +24,9 @@ public class StopDataFrameAnalyticsRequestTests extends AbstractWireSerializingT
         if (randomBoolean()) {
             request.setAllowNoMatch(randomBoolean());
         }
+        if (randomBoolean()) {
+            request.setForce(randomBoolean());
+        }
         int expandedIdsCount = randomIntBetween(0, 10);
         Set<String> expandedIds = new HashSet<>();
         for (int i = 0; i < expandedIdsCount; i++) {

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsStateTests.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core.ml.dataframe;
+
+import org.elasticsearch.test.ESTestCase;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+
+public class DataFrameAnalyticsStateTests extends ESTestCase {
+
+    public void testFromString() {
+        assertThat(DataFrameAnalyticsState.fromString("started"), equalTo(DataFrameAnalyticsState.STARTED));
+        assertThat(DataFrameAnalyticsState.fromString("reindexing"), equalTo(DataFrameAnalyticsState.REINDEXING));
+        assertThat(DataFrameAnalyticsState.fromString("analyzing"), equalTo(DataFrameAnalyticsState.ANALYZING));
+        assertThat(DataFrameAnalyticsState.fromString("stopping"), equalTo(DataFrameAnalyticsState.STOPPING));
+        assertThat(DataFrameAnalyticsState.fromString("stopped"), equalTo(DataFrameAnalyticsState.STOPPED));
+        assertThat(DataFrameAnalyticsState.fromString("failed"), equalTo(DataFrameAnalyticsState.FAILED));
+    }
+
+    public void testToString() {
+        assertThat(DataFrameAnalyticsState.STARTED.toString(), equalTo("started"));
+        assertThat(DataFrameAnalyticsState.REINDEXING.toString(), equalTo("reindexing"));
+        assertThat(DataFrameAnalyticsState.ANALYZING.toString(), equalTo("analyzing"));
+        assertThat(DataFrameAnalyticsState.STOPPING.toString(), equalTo("stopping"));
+        assertThat(DataFrameAnalyticsState.STOPPED.toString(), equalTo("stopped"));
+        assertThat(DataFrameAnalyticsState.FAILED.toString(), equalTo("failed"));
+    }
+
+    public void testIsAnyOf() {
+        assertThat(DataFrameAnalyticsState.STARTED.isAnyOf(), is(false));
+        assertThat(DataFrameAnalyticsState.STARTED.isAnyOf(DataFrameAnalyticsState.STARTED), is(true));
+        assertThat(DataFrameAnalyticsState.STARTED.isAnyOf(DataFrameAnalyticsState.ANALYZING, DataFrameAnalyticsState.STOPPED), is(false));
+        assertThat(DataFrameAnalyticsState.STARTED.isAnyOf(DataFrameAnalyticsState.STARTED, DataFrameAnalyticsState.STOPPED), is(true));
+        assertThat(DataFrameAnalyticsState.ANALYZING.isAnyOf(DataFrameAnalyticsState.STARTED, DataFrameAnalyticsState.STOPPED), is(false));
+        assertThat(DataFrameAnalyticsState.ANALYZING.isAnyOf(DataFrameAnalyticsState.ANALYZING, DataFrameAnalyticsState.FAILED), is(true));
+    }
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/dataframe/DataFrameAnalyticsTaskStateTests.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core.ml.dataframe;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractSerializingTestCase;
+
+import java.io.IOException;
+
+public class DataFrameAnalyticsTaskStateTests extends AbstractSerializingTestCase<DataFrameAnalyticsTaskState> {
+
+    @Override
+    protected DataFrameAnalyticsTaskState createTestInstance() {
+        return new DataFrameAnalyticsTaskState(randomFrom(DataFrameAnalyticsState.values()), randomLong(), randomAlphaOfLength(10));
+    }
+
+    @Override
+    protected Writeable.Reader<DataFrameAnalyticsTaskState> instanceReader() {
+        return DataFrameAnalyticsTaskState::new;
+    }
+
+    @Override
+    protected DataFrameAnalyticsTaskState doParseInstance(XContentParser parser) throws IOException {
+        return DataFrameAnalyticsTaskState.fromXContent(parser);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return true;
+    }
+}

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

@@ -501,7 +501,7 @@ public class MachineLearning extends Plugin implements ActionPlugin, AnalysisPlu
                     new BlackHoleAutodetectProcess(job.getId());
             // factor of 1.0 makes renormalization a no-op
             normalizerProcessFactory = (jobId, quantilesState, bucketSpan, executorService) -> new MultiplyingNormalizerProcess(1.0);
-            analyticsProcessFactory = (jobId, analyticsProcessConfig, executorService) -> null;
+            analyticsProcessFactory = (jobId, analyticsProcessConfig, executorService, onProcessCrash) -> null;
         }
         NormalizerFactory normalizerFactory = new NormalizerFactory(normalizerProcessFactory,
                 threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME));

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

@@ -34,6 +34,7 @@ import org.elasticsearch.xpack.core.ml.action.GetDataFrameAnalyticsStatsAction;
 import org.elasticsearch.xpack.core.ml.action.GetDataFrameAnalyticsStatsAction.Response.Stats;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsState;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsTaskState;
 import org.elasticsearch.xpack.ml.action.TransportStartDataFrameAnalyticsAction.DataFrameAnalyticsTask;
 import org.elasticsearch.xpack.ml.dataframe.process.AnalyticsProcessManager;
 
@@ -178,6 +179,11 @@ public class TransportGetDataFrameAnalyticsStatsAction
         PersistentTasksCustomMetaData tasks = clusterState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
         PersistentTasksCustomMetaData.PersistentTask<?> analyticsTask = MlTasks.getDataFrameAnalyticsTask(concreteAnalyticsId, tasks);
         DataFrameAnalyticsState analyticsState = MlTasks.getDataFrameAnalyticsState(concreteAnalyticsId, tasks);
+        String failureReason = null;
+        if (analyticsState == DataFrameAnalyticsState.FAILED) {
+            DataFrameAnalyticsTaskState taskState = (DataFrameAnalyticsTaskState) analyticsTask.getState();
+            failureReason = taskState.getReason();
+        }
         DiscoveryNode node = null;
         String assignmentExplanation = null;
         if (analyticsTask != null) {
@@ -185,6 +191,6 @@ public class TransportGetDataFrameAnalyticsStatsAction
             assignmentExplanation = analyticsTask.getAssignment().getExplanation();
         }
         return new GetDataFrameAnalyticsStatsAction.Response.Stats(
-            concreteAnalyticsId, analyticsState, progressPercent, node, assignmentExplanation);
+            concreteAnalyticsId, analyticsState, failureReason, progressPercent, node, assignmentExplanation);
     }
 }

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

@@ -62,6 +62,7 @@ import org.elasticsearch.xpack.ml.process.MlMemoryTracker;
 import java.io.IOException;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
 
 import static org.elasticsearch.xpack.core.ml.MlTasks.AWAITING_UPGRADE;
@@ -131,8 +132,6 @@ public class TransportStartDataFrameAnalyticsAction
             return;
         }
 
-        StartDataFrameAnalyticsAction.TaskParams taskParams = new StartDataFrameAnalyticsAction.TaskParams(request.getId());
-
         // Wait for analytics to be started
         ActionListener<PersistentTasksCustomMetaData.PersistentTask<StartDataFrameAnalyticsAction.TaskParams>> waitForAnalyticsToStart =
             new ActionListener<PersistentTasksCustomMetaData.PersistentTask<StartDataFrameAnalyticsAction.TaskParams>>() {
@@ -151,17 +150,26 @@ public class TransportStartDataFrameAnalyticsAction
                 }
             };
 
+        AtomicReference<DataFrameAnalyticsConfig> configHolder = new AtomicReference<>();
+
         // Start persistent task
         ActionListener<Void> memoryRequirementRefreshListener = ActionListener.wrap(
-            validated -> persistentTasksService.sendStartRequest(MlTasks.dataFrameAnalyticsTaskId(request.getId()),
-                MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME, taskParams, waitForAnalyticsToStart),
+            aVoid -> {
+                StartDataFrameAnalyticsAction.TaskParams taskParams = new StartDataFrameAnalyticsAction.TaskParams(
+                    request.getId(), configHolder.get().getVersion());
+                persistentTasksService.sendStartRequest(MlTasks.dataFrameAnalyticsTaskId(request.getId()),
+                    MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME, taskParams, waitForAnalyticsToStart);
+            },
             listener::onFailure
         );
 
         // Tell the job tracker to refresh the memory requirement for this job and all other jobs that have persistent tasks
         ActionListener<DataFrameAnalyticsConfig> configListener = ActionListener.wrap(
-            config -> memoryTracker.addDataFrameAnalyticsJobMemoryAndRefreshAllOthers(
-                request.getId(), config.getModelMemoryLimit().getBytes(), memoryRequirementRefreshListener),
+            config -> {
+                configHolder.set(config);
+                memoryTracker.addDataFrameAnalyticsJobMemoryAndRefreshAllOthers(
+                    request.getId(), config.getModelMemoryLimit().getBytes(), memoryRequirementRefreshListener);
+            },
             listener::onFailure
         );
 
@@ -251,7 +259,21 @@ public class TransportStartDataFrameAnalyticsAction
             }
             DataFrameAnalyticsTaskState taskState = (DataFrameAnalyticsTaskState) persistentTask.getState();
             DataFrameAnalyticsState analyticsState = taskState == null ? DataFrameAnalyticsState.STOPPED : taskState.getState();
-            return analyticsState == DataFrameAnalyticsState.STARTED;
+            switch (analyticsState) {
+                case STARTED:
+                case REINDEXING:
+                case ANALYZING:
+                    return true;
+                case STOPPING:
+                    exception = ExceptionsHelper.conflictStatusException("the task has been stopped while waiting to be started");
+                    return true;
+                case STOPPED:
+                    return false;
+                case FAILED:
+                default:
+                    exception = ExceptionsHelper.serverError("Unexpected task state [" + analyticsState + "] while waiting to be started");
+                    return true;
+            }
         }
     }
 
@@ -425,13 +447,15 @@ public class TransportStartDataFrameAnalyticsAction
             DataFrameAnalyticsTaskState analyticsTaskState = (DataFrameAnalyticsTaskState) state;
 
             // If we are "stopping" there is nothing to do
-            if (analyticsTaskState != null && analyticsTaskState.getState() == DataFrameAnalyticsState.STOPPING) {
+            // If we are "failed" then we should leave the task as is; for recovery it must be force stopped.
+            if (analyticsTaskState != null && analyticsTaskState.getState().isAnyOf(
+                    DataFrameAnalyticsState.STOPPING, DataFrameAnalyticsState.FAILED)) {
                 return;
             }
 
             if (analyticsTaskState == null) {
                 DataFrameAnalyticsTaskState startedState = new DataFrameAnalyticsTaskState(DataFrameAnalyticsState.STARTED,
-                    task.getAllocationId());
+                    task.getAllocationId(), null);
                 task.updatePersistentTaskState(startedState, ActionListener.wrap(
                     response -> manager.execute((DataFrameAnalyticsTask) task, DataFrameAnalyticsState.STARTED),
                     task::markAsFailed));

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

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.ml.action;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
@@ -31,6 +32,7 @@ import org.elasticsearch.xpack.core.ml.action.StopDataFrameAnalyticsAction;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsState;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsTaskState;
+import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.ml.MachineLearning;
 import org.elasticsearch.xpack.ml.dataframe.persistence.DataFrameAnalyticsConfigProvider;
 
@@ -89,13 +91,10 @@ public class TransportStopDataFrameAnalyticsAction
                     return;
                 }
 
-                Set<String> startedAnalytics = new HashSet<>();
-                Set<String> stoppingAnalytics = new HashSet<>();
                 PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
-                sortAnalyticsByTaskState(expandedIds, tasks, startedAnalytics, stoppingAnalytics);
-
-                request.setExpandedIds(startedAnalytics);
-                request.setNodes(findAllocatedNodesAndRemoveUnassignedTasks(startedAnalytics, tasks));
+                Set<String> analyticsToStop = findAnalyticsToStop(tasks, expandedIds, request.isForce());
+                request.setExpandedIds(analyticsToStop);
+                request.setNodes(findAllocatedNodesAndRemoveUnassignedTasks(analyticsToStop, tasks));
 
                 ActionListener<StopDataFrameAnalyticsAction.Response> finalListener = ActionListener.wrap(
                     r -> waitForTaskRemoved(expandedIds, request, r, listener),
@@ -110,8 +109,28 @@ public class TransportStopDataFrameAnalyticsAction
         expandIds(state, request, expandedIdsListener);
     }
 
+    /** Visible for testing */
+    static Set<String> findAnalyticsToStop(PersistentTasksCustomMetaData tasks, Set<String> ids, boolean force) {
+        Set<String> startedAnalytics = new HashSet<>();
+        Set<String> stoppingAnalytics = new HashSet<>();
+        Set<String> failedAnalytics = new HashSet<>();
+        sortAnalyticsByTaskState(ids, tasks, startedAnalytics, stoppingAnalytics, failedAnalytics);
+
+        if (force == false && failedAnalytics.isEmpty() == false) {
+            ElasticsearchStatusException e = failedAnalytics.size() == 1 ? ExceptionsHelper.conflictStatusException(
+                "cannot close data frame analytics [{}] because it failed, use force stop instead", failedAnalytics.iterator().next()) :
+                ExceptionsHelper.conflictStatusException("one or more data frame analytics are in failed state, " +
+                    "use force stop instead");
+            throw e;
+        }
+
+        startedAnalytics.addAll(failedAnalytics);
+        return startedAnalytics;
+    }
+
     private static void sortAnalyticsByTaskState(Set<String> analyticsIds, PersistentTasksCustomMetaData tasks,
-                                                 Set<String> startedAnalytics, Set<String> stoppingAnalytics) {
+                                                 Set<String> startedAnalytics, Set<String> stoppingAnalytics,
+                                                 Set<String> failedAnalytics) {
         for (String analyticsId : analyticsIds) {
             switch (MlTasks.getDataFrameAnalyticsState(analyticsId, tasks)) {
                 case STARTED:
@@ -124,6 +143,9 @@ public class TransportStopDataFrameAnalyticsAction
                     break;
                 case STOPPED:
                     break;
+                case FAILED:
+                    failedAnalytics.add(analyticsId);
+                    break;
                 default:
                     break;
             }
@@ -203,7 +225,7 @@ public class TransportStopDataFrameAnalyticsAction
                                  TransportStartDataFrameAnalyticsAction.DataFrameAnalyticsTask task,
                                  ActionListener<StopDataFrameAnalyticsAction.Response> listener) {
         DataFrameAnalyticsTaskState stoppingState =
-            new DataFrameAnalyticsTaskState(DataFrameAnalyticsState.STOPPING, task.getAllocationId());
+            new DataFrameAnalyticsTaskState(DataFrameAnalyticsState.STOPPING, task.getAllocationId(), null);
         task.updatePersistentTaskState(stoppingState, ActionListener.wrap(pTask -> {
                 threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(new AbstractRunnable() {
                     @Override
@@ -213,6 +235,7 @@ public class TransportStopDataFrameAnalyticsAction
 
                     @Override
                     protected void doRun() {
+                        logger.info("[{}] Stopping task with force [{}]", task.getParams().getId(), request.isForce());
                         task.stop("stop_data_frame_analytics (api)", request.getTimeout());
                         listener.onResponse(new StopDataFrameAnalyticsAction.Response(true));
                     }

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

@@ -70,7 +70,7 @@ public class DataFrameAnalyticsManager {
         ActionListener<DataFrameAnalyticsConfig> configListener = ActionListener.wrap(
             config -> {
                 DataFrameAnalyticsTaskState reindexingState = new DataFrameAnalyticsTaskState(DataFrameAnalyticsState.REINDEXING,
-                    task.getAllocationId());
+                    task.getAllocationId(), null);
                 switch(currentState) {
                     // If we are STARTED, we are right at the beginning of our task, we should indicate that we are entering the
                     // REINDEX state and start reindexing.
@@ -191,7 +191,7 @@ public class DataFrameAnalyticsManager {
         ActionListener<DataFrameDataExtractorFactory> dataExtractorFactoryListener = ActionListener.wrap(
             dataExtractorFactory -> {
                 DataFrameAnalyticsTaskState analyzingState = new DataFrameAnalyticsTaskState(DataFrameAnalyticsState.ANALYZING,
-                    task.getAllocationId());
+                    task.getAllocationId(), null);
                 task.updatePersistentTaskState(analyzingState, ActionListener.wrap(
                     updatedTask -> processManager.runJob(task, config, dataExtractorFactory,
                         error -> {

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

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.ml.dataframe.process;
 
 import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
 
 public interface AnalyticsProcessFactory {
 
@@ -15,7 +16,9 @@ public interface AnalyticsProcessFactory {
      * @param jobId             The job id
      * @param analyticsProcessConfig The process configuration
      * @param executorService   Executor service used to start the async tasks a job needs to operate the analytical process
+     * @param onProcessCrash    Callback to execute if the process stops unexpectedly
      * @return The process
      */
-    AnalyticsProcess createAnalyticsProcess(String jobId, AnalyticsProcessConfig analyticsProcessConfig, ExecutorService executorService);
+    AnalyticsProcess createAnalyticsProcess(String jobId, AnalyticsProcessConfig analyticsProcessConfig, ExecutorService executorService,
+                                            Consumer<String> onProcessCrash);
 }

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

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.ml.dataframe.process;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
 import org.elasticsearch.client.Client;
@@ -15,9 +16,11 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsState;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsTaskState;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.ml.MachineLearning;
-import org.elasticsearch.xpack.ml.action.TransportStartDataFrameAnalyticsAction;
+import org.elasticsearch.xpack.ml.action.TransportStartDataFrameAnalyticsAction.DataFrameAnalyticsTask;
 import org.elasticsearch.xpack.ml.dataframe.extractor.DataFrameDataExtractor;
 import org.elasticsearch.xpack.ml.dataframe.extractor.DataFrameDataExtractorFactory;
 
@@ -46,7 +49,7 @@ public class AnalyticsProcessManager {
         this.processFactory = Objects.requireNonNull(analyticsProcessFactory);
     }
 
-    public void runJob(TransportStartDataFrameAnalyticsAction.DataFrameAnalyticsTask task, DataFrameAnalyticsConfig config,
+    public void runJob(DataFrameAnalyticsTask task, DataFrameAnalyticsConfig config,
                        DataFrameDataExtractorFactory dataExtractorFactory, Consumer<Exception> finishHandler) {
         threadPool.generic().execute(() -> {
             if (task.isStopping()) {
@@ -61,10 +64,10 @@ public class AnalyticsProcessManager {
                     + "] Could not create process as one already exists"));
                 return;
             }
-            if (processContext.startProcess(dataExtractorFactory, config)) {
+            if (processContext.startProcess(dataExtractorFactory, config, task)) {
                 ExecutorService executorService = threadPool.executor(MachineLearning.JOB_COMMS_THREAD_POOL_NAME);
-                executorService.execute(() -> processContext.resultProcessor.process(processContext.process));
-                executorService.execute(() -> processData(task.getAllocationId(), config, processContext.dataExtractor,
+                executorService.execute(() -> processResults(processContext));
+                executorService.execute(() -> processData(task, config, processContext.dataExtractor,
                     processContext.process, processContext.resultProcessor, finishHandler));
             } else {
                 finishHandler.accept(null);
@@ -72,8 +75,17 @@ public class AnalyticsProcessManager {
         });
     }
 
-    private void processData(long taskAllocationId, DataFrameAnalyticsConfig config, DataFrameDataExtractor dataExtractor,
+    private void processResults(ProcessContext processContext) {
+        try {
+            processContext.resultProcessor.process(processContext.process);
+        } catch (Exception e) {
+            processContext.setFailureReason(e.getMessage());
+        }
+    }
+
+    private void processData(DataFrameAnalyticsTask task, DataFrameAnalyticsConfig config, DataFrameDataExtractor dataExtractor,
                              AnalyticsProcess process, AnalyticsResultProcessor resultProcessor, Consumer<Exception> finishHandler) {
+
         try {
             writeHeaderRecord(dataExtractor, process);
             writeDataRows(dataExtractor, process);
@@ -82,26 +94,28 @@ public class AnalyticsProcessManager {
 
             LOGGER.info("[{}] Waiting for result processor to complete", config.getId());
             resultProcessor.awaitForCompletion();
+            processContextByAllocation.get(task.getAllocationId()).setFailureReason(resultProcessor.getFailure());
+
             refreshDest(config);
             LOGGER.info("[{}] Result processor has completed", config.getId());
-        } catch (IOException e) {
-            LOGGER.error(new ParameterizedMessage("[{}] Error writing data to the process", config.getId()), e);
-            // TODO Handle this failure by setting the task state to FAILED
+        } catch (Exception e) {
+            String errorMsg = new ParameterizedMessage("[{}] Error while processing data", config.getId()).getFormattedMessage();
+            processContextByAllocation.get(task.getAllocationId()).setFailureReason(errorMsg);
         } finally {
-            LOGGER.info("[{}] Closing process", config.getId());
-            try {
-                process.close();
-                LOGGER.info("[{}] Closed process", config.getId());
+            closeProcess(task);
 
+            ProcessContext processContext = processContextByAllocation.remove(task.getAllocationId());
+            LOGGER.debug("Removed process context for task [{}]; [{}] processes still running", config.getId(),
+                processContextByAllocation.size());
+
+            if (processContext.getFailureReason() == null) {
                 // This results in marking the persistent task as complete
+                LOGGER.info("[{}] Marking task completed", config.getId());
                 finishHandler.accept(null);
-            } catch (IOException e) {
-                LOGGER.error("[{}] Error closing data frame analyzer process", config.getId());
-                finishHandler.accept(e);
+            } else {
+                LOGGER.error("[{}] Marking task failed; {}", config.getId(), processContext.getFailureReason());
+                updateTaskState(task, DataFrameAnalyticsState.FAILED, processContext.getFailureReason());
             }
-            processContextByAllocation.remove(taskAllocationId);
-            LOGGER.debug("Removed process context for task [{}]; [{}] processes still running", config.getId(),
-                processContextByAllocation.size());
         }
     }
 
@@ -142,15 +156,34 @@ public class AnalyticsProcessManager {
         process.writeRecord(headerRecord);
     }
 
-    private AnalyticsProcess createProcess(String jobId, AnalyticsProcessConfig analyticsProcessConfig) {
+    private AnalyticsProcess createProcess(DataFrameAnalyticsTask task, AnalyticsProcessConfig analyticsProcessConfig) {
         ExecutorService executorService = threadPool.executor(MachineLearning.JOB_COMMS_THREAD_POOL_NAME);
-        AnalyticsProcess process = processFactory.createAnalyticsProcess(jobId, analyticsProcessConfig, executorService);
+        AnalyticsProcess process = processFactory.createAnalyticsProcess(task.getParams().getId(), analyticsProcessConfig,
+            executorService, onProcessCrash(task));
         if (process.isProcessAlive() == false) {
             throw ExceptionsHelper.serverError("Failed to start data frame analytics process");
         }
         return process;
     }
 
+    private Consumer<String> onProcessCrash(DataFrameAnalyticsTask task) {
+        return reason -> {
+            ProcessContext processContext = processContextByAllocation.get(task.getAllocationId());
+            if (processContext != null) {
+                processContext.setFailureReason(reason);
+                processContext.stop();
+            }
+        };
+    }
+
+    private void updateTaskState(DataFrameAnalyticsTask task, DataFrameAnalyticsState state, @Nullable String reason) {
+        DataFrameAnalyticsTaskState newTaskState = new DataFrameAnalyticsTaskState(state, task.getAllocationId(), reason);
+        task.updatePersistentTaskState(newTaskState, ActionListener.wrap(
+            updatedTask -> LOGGER.info("[{}] Successfully update task state to [{}]", task.getParams().getId(), state),
+            e -> LOGGER.error(new ParameterizedMessage("[{}] Could not update task state to [{}]", task.getParams().getId(), state), e)
+        ));
+    }
+
     @Nullable
     public Integer getProgressPercent(long allocationId) {
         ProcessContext processContext = processContextByAllocation.get(allocationId);
@@ -162,13 +195,29 @@ public class AnalyticsProcessManager {
             () -> client.execute(RefreshAction.INSTANCE, new RefreshRequest(config.getDest().getIndex())).actionGet());
     }
 
-    public void stop(TransportStartDataFrameAnalyticsAction.DataFrameAnalyticsTask task) {
+    private void closeProcess(DataFrameAnalyticsTask task) {
+        String configId = task.getParams().getId();
+        LOGGER.info("[{}] Closing process", configId);
+
+        ProcessContext processContext = processContextByAllocation.get(task.getAllocationId());
+        try {
+            processContext.process.close();
+            LOGGER.info("[{}] Closed process", configId);
+        } catch (IOException e) {
+            String errorMsg = new ParameterizedMessage("[{}] Error closing data frame analyzer process [{}]"
+                , configId, e.getMessage()).getFormattedMessage();
+            processContext.setFailureReason(errorMsg);
+        }
+    }
+
+    public void stop(DataFrameAnalyticsTask task) {
         ProcessContext processContext = processContextByAllocation.get(task.getAllocationId());
         if (processContext != null) {
             LOGGER.debug("[{}] Stopping process", task.getParams().getId() );
             processContext.stop();
         } else {
             LOGGER.debug("[{}] No process context to stop", task.getParams().getId() );
+            task.markAsCompleted();
         }
     }
 
@@ -180,6 +229,7 @@ public class AnalyticsProcessManager {
         private volatile AnalyticsResultProcessor resultProcessor;
         private final AtomicInteger progressPercent = new AtomicInteger(0);
         private volatile boolean processKilled;
+        private volatile String failureReason;
 
         ProcessContext(String id) {
             this.id = Objects.requireNonNull(id);
@@ -197,6 +247,17 @@ public class AnalyticsProcessManager {
             this.progressPercent.set(progressPercent);
         }
 
+        private synchronized void setFailureReason(String failureReason) {
+            // Only set the new reason if there isn't one already as we want to keep the first reason
+            if (failureReason != null) {
+                this.failureReason = failureReason;
+            }
+        }
+
+        private String getFailureReason() {
+            return failureReason;
+        }
+
         public synchronized void stop() {
             LOGGER.debug("[{}] Stopping process", id);
             processKilled = true;
@@ -215,14 +276,15 @@ public class AnalyticsProcessManager {
         /**
          * @return {@code true} if the process was started or {@code false} if it was not because it was stopped in the meantime
          */
-        private synchronized boolean startProcess(DataFrameDataExtractorFactory dataExtractorFactory, DataFrameAnalyticsConfig config) {
+        private synchronized boolean startProcess(DataFrameDataExtractorFactory dataExtractorFactory, DataFrameAnalyticsConfig config,
+                                                  DataFrameAnalyticsTask task) {
             if (processKilled) {
                 // The job was stopped before we started the process so no need to start it
                 return false;
             }
 
             dataExtractor = dataExtractorFactory.newExtractor(false);
-            process = createProcess(config.getId(), createProcessConfig(config, dataExtractor));
+            process = createProcess(task, createProcessConfig(config, dataExtractor));
             DataFrameRowsJoiner dataFrameRowsJoiner = new DataFrameRowsJoiner(config.getId(), client,
                 dataExtractorFactory.newExtractor(true));
             resultProcessor = new AnalyticsResultProcessor(id, dataFrameRowsJoiner, this::isProcessKilled, this::setProgressPercent);

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

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.ml.dataframe.process;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.xpack.ml.dataframe.process.results.RowResults;
 
 import java.util.Iterator;
@@ -26,6 +27,7 @@ public class AnalyticsResultProcessor {
     private final Supplier<Boolean> isProcessKilled;
     private final Consumer<Integer> progressConsumer;
     private final CountDownLatch completionLatch = new CountDownLatch(1);
+    private volatile String failure;
 
     public AnalyticsResultProcessor(String dataFrameAnalyticsId, DataFrameRowsJoiner dataFrameRowsJoiner, Supplier<Boolean> isProcessKilled,
                                     Consumer<Integer> progressConsumer) {
@@ -35,6 +37,11 @@ public class AnalyticsResultProcessor {
         this.progressConsumer = Objects.requireNonNull(progressConsumer);
     }
 
+    @Nullable
+    public String getFailure() {
+        return failure == null ? dataFrameRowsJoiner.getFailure() : failure;
+    }
+
     public void awaitForCompletion() {
         try {
             if (completionLatch.await(30, TimeUnit.MINUTES) == false) {
@@ -59,6 +66,7 @@ public class AnalyticsResultProcessor {
                 // No need to log error as it's due to stopping
             } else {
                 LOGGER.error(new ParameterizedMessage("[{}] Error parsing data frame analytics output", dataFrameAnalyticsId), e);
+                failure = "error parsing data frame analytics output: [" + e.getMessage() + "]";
             }
         } finally {
             completionLatch.countDown();

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

@@ -14,6 +14,7 @@ import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
@@ -41,7 +42,7 @@ class DataFrameRowsJoiner implements AutoCloseable {
     private final DataFrameDataExtractor dataExtractor;
     private final Iterator<DataFrameDataExtractor.Row> dataFrameRowsIterator;
     private LinkedList<RowResults> currentResults;
-    private boolean failed;
+    private volatile String failure;
 
     DataFrameRowsJoiner(String analyticsId, Client client, DataFrameDataExtractor dataExtractor) {
         this.analyticsId = Objects.requireNonNull(analyticsId);
@@ -51,8 +52,13 @@ class DataFrameRowsJoiner implements AutoCloseable {
         this.currentResults = new LinkedList<>();
     }
 
+    @Nullable
+    String getFailure() {
+        return failure;
+    }
+
     void processRowResults(RowResults rowResults) {
-        if (failed) {
+        if (failure != null) {
             // If we are in failed state we drop the results but we let the processor
             // parse the output
             return;
@@ -61,8 +67,8 @@ class DataFrameRowsJoiner implements AutoCloseable {
         try {
             addResultAndJoinIfEndOfBatch(rowResults);
         } catch (Exception e) {
-            LOGGER.error(new ParameterizedMessage("[{}] Failed to join results", analyticsId), e);
-            failed = true;
+            LOGGER.error(new ParameterizedMessage("[{}] Failed to join results ", analyticsId), e);
+            failure = "[" + analyticsId + "] Failed to join results: " + e.getMessage();
         }
     }
 
@@ -93,8 +99,7 @@ class DataFrameRowsJoiner implements AutoCloseable {
             msg += "expected [" + row.getChecksum() + "] but result had [" + result.getChecksum() + "]; ";
             msg += "this implies the data frame index [" + row.getHit().getIndex() + "] was modified while the analysis was running. ";
             msg += "We rely on this index being immutable during a running analysis and so the results will be unreliable.";
-            throw new RuntimeException(msg);
-            // TODO Communicate this error to the user as effectively the analytics have failed (e.g. FAILED state, audit error, etc.)
+            throw ExceptionsHelper.serverError(msg);
         }
     }
 
@@ -112,8 +117,7 @@ class DataFrameRowsJoiner implements AutoCloseable {
         BulkResponse bulkResponse = ClientHelper.executeWithHeaders(dataExtractor.getHeaders(), ClientHelper.ML_ORIGIN, client,
                 () -> client.execute(BulkAction.INSTANCE, bulkRequest).actionGet());
         if (bulkResponse.hasFailures()) {
-            LOGGER.error("Failures while writing data frame");
-            // TODO Better error handling
+            throw ExceptionsHelper.serverError("failures while writing results [" + bulkResponse.buildFailureMessage() + "]");
         }
     }
 
@@ -123,7 +127,7 @@ class DataFrameRowsJoiner implements AutoCloseable {
             joinCurrentResults();
         } catch (Exception e) {
             LOGGER.error(new ParameterizedMessage("[{}] Failed to join results", analyticsId), e);
-            failed = true;
+            failure = "[" + analyticsId + "] Failed to join results: " + e.getMessage();
         } finally {
             try {
                 consumeDataExtractor();
@@ -159,7 +163,7 @@ class DataFrameRowsJoiner implements AutoCloseable {
             }
 
             if (row == null || row.shouldSkip()) {
-                throw ExceptionsHelper.serverError("No more data frame rows could be found while joining results");
+                throw ExceptionsHelper.serverError("no more data frame rows could be found while joining results");
             }
             return row;
         }
@@ -175,9 +179,7 @@ class DataFrameRowsJoiner implements AutoCloseable {
             try {
                 return dataExtractor.next();
             } catch (IOException e) {
-                // TODO Implement recovery strategy or better error reporting
-                LOGGER.error("Error reading next batch of data frame rows", e);
-                return Optional.empty();
+                throw ExceptionsHelper.serverError("error reading next batch of data frame rows [" + e.getMessage() + "]");
             }
         }
     }

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

@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
 
 public class NativeAnalyticsProcessFactory implements AnalyticsProcessFactory {
 
@@ -50,7 +51,7 @@ public class NativeAnalyticsProcessFactory implements AnalyticsProcessFactory {
 
     @Override
     public AnalyticsProcess createAnalyticsProcess(String jobId, AnalyticsProcessConfig analyticsProcessConfig,
-                                                   ExecutorService executorService) {
+                                                   ExecutorService executorService, Consumer<String> onProcessCrash) {
         List<Path> filesToDelete = new ArrayList<>();
         ProcessPipes processPipes = new ProcessPipes(env, NAMED_PIPE_HELPER, AnalyticsBuilder.ANALYTICS, jobId,
                 true, false, true, true, false, false);
@@ -62,8 +63,7 @@ public class NativeAnalyticsProcessFactory implements AnalyticsProcessFactory {
 
         NativeAnalyticsProcess analyticsProcess = new NativeAnalyticsProcess(jobId, nativeController, processPipes.getLogStream().get(),
                 processPipes.getProcessInStream().get(), processPipes.getProcessOutStream().get(), null, numberOfFields,
-                filesToDelete, reason -> {});
-
+                filesToDelete, onProcessCrash);
 
         try {
             analyticsProcess.start(executorService);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobNodeSelector.java

@@ -265,8 +265,9 @@ public class JobNodeSelector {
                 MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME, task -> node.getId().equals(task.getExecutorNode()));
             for (PersistentTasksCustomMetaData.PersistentTask<?> assignedTask : assignedAnalyticsTasks) {
                 DataFrameAnalyticsState dataFrameAnalyticsState = ((DataFrameAnalyticsTaskState) assignedTask.getState()).getState();
-                // TODO: skip FAILED here too if such a state is ever added
-                if (dataFrameAnalyticsState != DataFrameAnalyticsState.STOPPED) {
+
+                // Don't count stopped and failed df-analytics tasks as they don't consume native memory
+                if (dataFrameAnalyticsState.isAnyOf(DataFrameAnalyticsState.STOPPED, DataFrameAnalyticsState.FAILED) == false) {
                     // The native process is only running in the ANALYZING and STOPPING states, but in the STARTED
                     // and REINDEXING states we're committed to using the memory soon, so account for it here
                     ++result.numberOfAssignedJobs;

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

@@ -7,7 +7,6 @@ package org.elasticsearch.xpack.ml.rest.dataframe;
 
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
@@ -39,15 +38,11 @@ public class RestStopDataFrameAnalyticsAction extends BaseRestHandler {
             request = StopDataFrameAnalyticsAction.Request.parseRequest(id, restRequest.contentOrSourceParamParser());
         } else {
             request = new StopDataFrameAnalyticsAction.Request(id);
-            if (restRequest.hasParam(StopDataFrameAnalyticsAction.Request.TIMEOUT.getPreferredName())) {
-                TimeValue timeout = restRequest.paramAsTime(StopDataFrameAnalyticsAction.Request.TIMEOUT.getPreferredName(),
-                    request.getTimeout());
-                request.setTimeout(timeout);
-            }
-            if (restRequest.hasParam(StopDataFrameAnalyticsAction.Request.ALLOW_NO_MATCH.getPreferredName())) {
-                request.setAllowNoMatch(restRequest.paramAsBoolean(StopDataFrameAnalyticsAction.Request.ALLOW_NO_MATCH.getPreferredName(),
-                    request.allowNoMatch()));
-            }
+            request.setTimeout(restRequest.paramAsTime(StopDataFrameAnalyticsAction.Request.TIMEOUT.getPreferredName(),
+                request.getTimeout()));
+            request.setAllowNoMatch(restRequest.paramAsBoolean(StopDataFrameAnalyticsAction.Request.ALLOW_NO_MATCH.getPreferredName(),
+                request.allowNoMatch()));
+            request.setForce(restRequest.paramAsBoolean(StopDataFrameAnalyticsAction.Request.FORCE.getPreferredName(), request.isForce()));
         }
         return channel -> client.execute(StopDataFrameAnalyticsAction.INSTANCE, request, new RestToXContentListener<>(channel));
     }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/action/TransportStopDataFrameAnalyticsActionTests.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.ml.action;
+
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.Version;
+import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.ml.MlTasks;
+import org.elasticsearch.xpack.core.ml.action.StartDataFrameAnalyticsAction;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsState;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsTaskState;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.equalTo;
+
+public class TransportStopDataFrameAnalyticsActionTests extends ESTestCase {
+
+    public void testFindAnalyticsToStop_GivenOneFailedTaskAndNotForce() {
+        PersistentTasksCustomMetaData.Builder tasksBuilder =  PersistentTasksCustomMetaData.builder();
+        addAnalyticsTask(tasksBuilder, "started", "foo-node", DataFrameAnalyticsState.STARTED);
+        addAnalyticsTask(tasksBuilder, "reindexing", "foo-node", DataFrameAnalyticsState.REINDEXING);
+        addAnalyticsTask(tasksBuilder, "analyzing", "foo-node", DataFrameAnalyticsState.ANALYZING);
+        addAnalyticsTask(tasksBuilder, "stopping", "foo-node", DataFrameAnalyticsState.STOPPING);
+        addAnalyticsTask(tasksBuilder, "stopped", "foo-node", DataFrameAnalyticsState.STOPPED);
+        addAnalyticsTask(tasksBuilder, "failed", "foo-node", DataFrameAnalyticsState.FAILED);
+
+        Set<String> ids = new HashSet<>(Arrays.asList("started", "reindexing", "analyzing", "stopping", "stopped", "failed"));
+
+        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class,
+            () -> TransportStopDataFrameAnalyticsAction.findAnalyticsToStop(tasksBuilder.build(), ids, false));
+
+        assertThat(e.status(), equalTo(RestStatus.CONFLICT));
+        assertThat(e.getMessage(), equalTo("cannot close data frame analytics [failed] because it failed, use force stop instead"));
+    }
+
+    public void testFindAnalyticsToStop_GivenTwoFailedTasksAndNotForce() {
+        PersistentTasksCustomMetaData.Builder tasksBuilder =  PersistentTasksCustomMetaData.builder();
+        addAnalyticsTask(tasksBuilder, "failed", "foo-node", DataFrameAnalyticsState.FAILED);
+        addAnalyticsTask(tasksBuilder, "another_failed", "foo-node", DataFrameAnalyticsState.FAILED);
+
+        Set<String> ids = new HashSet<>(Arrays.asList("failed", "another_failed"));
+
+        ElasticsearchStatusException e = expectThrows(ElasticsearchStatusException.class,
+            () -> TransportStopDataFrameAnalyticsAction.findAnalyticsToStop(tasksBuilder.build(), ids, false));
+
+        assertThat(e.status(), equalTo(RestStatus.CONFLICT));
+        assertThat(e.getMessage(), equalTo("one or more data frame analytics are in failed state, use force stop instead"));
+    }
+
+    public void testFindAnalyticsToStop_GivenFailedTaskAndForce() {
+        PersistentTasksCustomMetaData.Builder tasksBuilder =  PersistentTasksCustomMetaData.builder();
+        addAnalyticsTask(tasksBuilder, "started", "foo-node", DataFrameAnalyticsState.STARTED);
+        addAnalyticsTask(tasksBuilder, "reindexing", "foo-node", DataFrameAnalyticsState.REINDEXING);
+        addAnalyticsTask(tasksBuilder, "analyzing", "foo-node", DataFrameAnalyticsState.ANALYZING);
+        addAnalyticsTask(tasksBuilder, "stopping", "foo-node", DataFrameAnalyticsState.STOPPING);
+        addAnalyticsTask(tasksBuilder, "stopped", "foo-node", DataFrameAnalyticsState.STOPPED);
+        addAnalyticsTask(tasksBuilder, "failed", "foo-node", DataFrameAnalyticsState.FAILED);
+
+        Set<String> ids = new HashSet<>(Arrays.asList("started", "reindexing", "analyzing", "stopping", "stopped", "failed"));
+
+        Set<String> analyticsToStop = TransportStopDataFrameAnalyticsAction.findAnalyticsToStop(tasksBuilder.build(), ids, true);
+
+        assertThat(analyticsToStop, containsInAnyOrder("started", "reindexing", "analyzing", "failed"));
+    }
+
+    private static void addAnalyticsTask(PersistentTasksCustomMetaData.Builder builder, String analyticsId, String nodeId,
+                                         DataFrameAnalyticsState state) {
+        builder.addTask(MlTasks.dataFrameAnalyticsTaskId(analyticsId), MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME,
+            new StartDataFrameAnalyticsAction.TaskParams(analyticsId, Version.CURRENT),
+            new PersistentTasksCustomMetaData.Assignment(nodeId, "test assignment"));
+
+        builder.updateTaskState(MlTasks.dataFrameAnalyticsTaskId(analyticsId),
+            new DataFrameAnalyticsTaskState(state, builder.getLastAllocationId(), null));
+
+    }
+}

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/JobNodeSelectorTests.java

@@ -566,10 +566,11 @@ public class JobNodeSelectorTests extends ESTestCase {
     static void addDataFrameAnalyticsJobTask(String id, String nodeId, DataFrameAnalyticsState state,
                                              PersistentTasksCustomMetaData.Builder builder, boolean isStale) {
         builder.addTask(MlTasks.dataFrameAnalyticsTaskId(id), MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME,
-            new StartDataFrameAnalyticsAction.TaskParams(id), new PersistentTasksCustomMetaData.Assignment(nodeId, "test assignment"));
+            new StartDataFrameAnalyticsAction.TaskParams(id, Version.CURRENT),
+            new PersistentTasksCustomMetaData.Assignment(nodeId, "test assignment"));
         if (state != null) {
             builder.updateTaskState(MlTasks.dataFrameAnalyticsTaskId(id),
-                new DataFrameAnalyticsTaskState(state, builder.getLastAllocationId() - (isStale ? 1 : 0)));
+                new DataFrameAnalyticsTaskState(state, builder.getLastAllocationId() - (isStale ? 1 : 0), null));
         }
     }
 }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/process/MlMemoryTrackerTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.ml.process;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.ClusterSettings;
@@ -199,7 +200,7 @@ public class MlMemoryTrackerTests extends ESTestCase {
     private
     PersistentTasksCustomMetaData.PersistentTask<StartDataFrameAnalyticsAction.TaskParams> makeTestDataFrameAnalyticsTask(String id) {
         return new PersistentTasksCustomMetaData.PersistentTask<>(MlTasks.dataFrameAnalyticsTaskId(id),
-            MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME, new StartDataFrameAnalyticsAction.TaskParams(id), 0,
+            MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME, new StartDataFrameAnalyticsAction.TaskParams(id, Version.CURRENT), 0,
             PersistentTasksCustomMetaData.INITIAL_ASSIGNMENT);
     }
 }

x-pack/plugin/src/test/resources/rest-api-spec/api/ml.stop_data_frame_analytics.json

@@ -18,6 +18,11 @@
           "required": false,
           "description": "Whether to ignore if a wildcard expression matches no data frame analytics. (This includes `_all` string or when no data frame analytics have been specified)"
         },
+        "force": {
+          "type": "boolean",
+          "required": false,
+          "description": "True if the data frame analytics should be forcefully stopped"
+        },
         "timeout": {
           "type": "time",
           "required": false,