diff --git a/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfig.java b/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfig.java
index 97806ac4c3f..3a8d89f5bcb 100644
--- a/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfig.java
+++ b/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfig.java
@@ -119,8 +119,10 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
             // TODO this is to read former _source field. Remove in v7.0.0
             parser.declareBoolean((builder, value) -> {}, SOURCE);
             parser.declareObject(Builder::setChunkingConfig, ChunkingConfig.PARSERS.get(parserType), CHUNKING_CONFIG);
-            parser.declareObject(Builder::setHeaders, (p, c) -> p.mapStrings(), HEADERS);
         }
+        // Headers are only parsed by the metadata parser, so headers supplied in the _body_ of a REST request will be rejected.
+        // (For config headers are explicitly transferred from the auth headers by code in the put/update datafeed actions.)
+        METADATA_PARSER.declareObject(Builder::setHeaders, (p, c) -> p.mapStrings(), HEADERS);
     }
 
     private final String id;
@@ -328,8 +330,7 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
         if (chunkingConfig != null) {
             builder.field(CHUNKING_CONFIG.getPreferredName(), chunkingConfig);
         }
-        if (headers != null && headers.isEmpty() == false
-                && params.paramAsBoolean(ToXContentParams.FOR_CLUSTER_STATE, false) == true) {
+        if (headers.isEmpty() == false && params.paramAsBoolean(ToXContentParams.FOR_CLUSTER_STATE, false) == true) {
             builder.field(HEADERS.getPreferredName(), headers);
         }
         return builder;
@@ -477,7 +478,7 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
         }
 
         public void setHeaders(Map<String, String> headers) {
-            this.headers = headers;
+            this.headers = ExceptionsHelper.requireNonNull(headers, HEADERS.getPreferredName());
         }
 
         public void setIndices(List<String> indices) {
diff --git a/plugin/src/test/resources/rest-api-spec/test/ml/datafeeds_crud.yml b/plugin/src/test/resources/rest-api-spec/test/ml/datafeeds_crud.yml
index b3f5db1884a..a0f79b7caba 100644
--- a/plugin/src/test/resources/rest-api-spec/test/ml/datafeeds_crud.yml
+++ b/plugin/src/test/resources/rest-api-spec/test/ml/datafeeds_crud.yml
@@ -81,6 +81,19 @@ setup:
             "types":["type-bar"]
           }
 
+---
+"Test put datafeed with security headers in the body":
+  - do:
+      catch: /unknown field \[headers\], parser not found/
+      xpack.ml.put_datafeed:
+        datafeed_id: test-datafeed-1
+        body:  >
+          {
+            "job_id":"datafeeds-crud-1",
+            "indices":["index-foo"],
+            "headers":{ "a_security_header" : "secret" }
+          }
+
 ---
 "Test put datafeed referring to existing job_id":
   - do:
diff --git a/qa/smoke-test-ml-with-security/build.gradle b/qa/smoke-test-ml-with-security/build.gradle
index 0ff59a76b6d..1a6cf22fca9 100644
--- a/qa/smoke-test-ml-with-security/build.gradle
+++ b/qa/smoke-test-ml-with-security/build.gradle
@@ -32,6 +32,7 @@ integTestRunner {
     'ml/datafeeds_crud/Test delete datafeed with missing id',
     'ml/datafeeds_crud/Test put datafeed referring to missing job_id',
     'ml/datafeeds_crud/Test put datafeed with invalid query',
+    'ml/datafeeds_crud/Test put datafeed with security headers in the body',
     'ml/datafeeds_crud/Test update datafeed with missing id',
     'ml/delete_job_force/Test cannot force delete a non-existent job',
     'ml/delete_model_snapshot/Test delete snapshot missing snapshotId',