diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDataFrameAnalyticsAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDataFrameAnalyticsAction.java
index 82152de8f69..74497e5e060 100644
--- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDataFrameAnalyticsAction.java
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDataFrameAnalyticsAction.java
@@ -50,6 +50,8 @@ import org.elasticsearch.xpack.ml.process.MlMemoryTracker;
 import org.elasticsearch.xpack.ml.utils.MlIndicesUtils;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
@@ -138,7 +140,7 @@ public class TransportDeleteDataFrameAnalyticsAction
 
         // Step 2. Delete state
         ActionListener<DataFrameAnalyticsConfig> configListener = ActionListener.wrap(
-            config -> deleteState(parentTaskClient, id, deleteStateHandler),
+            config -> deleteState(parentTaskClient, config, deleteStateHandler),
             listener::onFailure
         );
 
@@ -165,11 +167,16 @@ public class TransportDeleteDataFrameAnalyticsAction
         ));
     }
 
-    private void deleteState(ParentTaskAssigningClient parentTaskClient, String analyticsId,
+    private void deleteState(ParentTaskAssigningClient parentTaskClient,
+                             DataFrameAnalyticsConfig config,
                              ActionListener<BulkByScrollResponse> listener) {
+        List<String> ids = new ArrayList<>();
+        ids.add(DataFrameAnalyticsTask.progressDocId(config.getId()));
+        if (config.getAnalysis().persistsState()) {
+            ids.add(config.getAnalysis().getStateDocId(config.getId()));
+        }
         DeleteByQueryRequest request = new DeleteByQueryRequest(AnomalyDetectorsIndex.jobStateIndexPattern());
-        request.setQuery(QueryBuilders.idsQuery().addIds(
-            DataFrameAnalyticsTask.progressDocId(analyticsId)));
+        request.setQuery(QueryBuilders.idsQuery().addIds(ids.toArray(new String[0])));
         request.setIndicesOptions(MlIndicesUtils.addIgnoreUnavailable(IndicesOptions.lenientExpandOpen()));
         request.setSlices(AbstractBulkByScrollRequest.AUTO_SLICES);
         request.setAbortOnVersionConflict(false);
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml
index da00a12ceb9..2e44618cb76 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/ml/data_frame_analytics_crud.yml
@@ -900,6 +900,85 @@ setup:
       ml.delete_data_frame_analytics:
         id: "missing_config"
 
+---
+"Test delete given config and state documents":
+  - skip:
+      features: headers
+  - do:
+      ml.put_data_frame_analytics:
+        id: "delete_foo"
+        body: >
+          {
+            "source": {
+              "index": "index-source"
+            },
+            "dest": {
+              "index": "index-dest"
+            },
+            "analysis": {"regression":{
+                "dependent_variable": "to_predict"
+            }}
+          }
+  - do:
+      headers:
+        Authorization: "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" # run as x_pack_rest_user, i.e. the test setup superuser
+      index:
+        index: .ml-state
+        id: "delete_foo_regression_state#1"
+        body: >
+          {
+          }
+
+  - do:
+      headers:
+        Authorization: "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" # run as x_pack_rest_user, i.e. the test setup superuser
+      index:
+        index: .ml-state
+        id: "data_frame_analytics-delete_foo-progress"
+        body: >
+          {
+          }
+
+  - do:
+      headers:
+        Authorization: "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" # run as x_pack_rest_user, i.e. the test setup superuser
+      indices.refresh:
+        index: .ml-state
+
+  - do:
+      search:
+        index: .ml-state
+        body:
+          size: 0
+          query:
+            bool:
+              filter:
+                ids:
+                  values:
+                    - "data_frame_analytics-delete_foo-progress"
+                    - "delete_foo_regression_state#1"
+
+  - match: { hits.total.value: 2 }
+
+  - do:
+      ml.delete_data_frame_analytics:
+        id: "delete_foo"
+  - match: { acknowledged: true }
+
+  - do:
+      search:
+        index: .ml-state*
+        body:
+          size: 0
+          query:
+            bool:
+              filter:
+                ids:
+                  values:
+                    - "data_frame_analytics-delete_foo-progress"
+                    - "delete_foo_regression_state#1"
+  - match: { hits.total.value: 0 }
+
 ---
 "Test max model memory limit":
   - skip: