diff --git a/docs/reference/indices/rollover-index.asciidoc b/docs/reference/indices/rollover-index.asciidoc
index f4fdac4e36282..7918234a69eaa 100644
--- a/docs/reference/indices/rollover-index.asciidoc
+++ b/docs/reference/indices/rollover-index.asciidoc
@@ -233,14 +233,14 @@ PUT _index_template/template
   "template": {
     "mappings": {
       "properties": {
-        "@timestamp": {
+        "date": {
           "type": "date"
         }
       }
     }
   },
   "data_stream": {
-    "timestamp_field": "@timestamp"
+    "timestamp_field": "date"
   }
 }
 -----------------------------------
diff --git a/modules/reindex/src/test/resources/rest-api-spec/test/reindex/96_data_streams.yml b/modules/reindex/src/test/resources/rest-api-spec/test/reindex/96_data_streams.yml
index 3880233cf9181..6cfa815159b45 100644
--- a/modules/reindex/src/test/resources/rest-api-spec/test/reindex/96_data_streams.yml
+++ b/modules/reindex/src/test/resources/rest-api-spec/test/reindex/96_data_streams.yml
@@ -34,7 +34,9 @@ teardown:
       index:
         index:  logs-foobar
         refresh: true
-        body:   { foo: bar }
+        body:
+          foo: bar
+          timestamp: '2020-12-12'
 
   - do:
       reindex:
@@ -65,7 +67,9 @@ teardown:
       index:
         index: old-logs-index
         refresh: true
-        body:   { foo: bar }
+        body:
+          foo: bar
+          timestamp: '2020-12-12'
 
   - do:
       reindex:
@@ -96,7 +100,9 @@ teardown:
       index:
         index:  logs-foobar
         refresh: true
-        body:   { foo: bar }
+        body:
+          foo: bar
+          timestamp: '2020-12-12'
 
   - do:
       reindex:
diff --git a/plugins/mapper-murmur3/src/test/java/org/elasticsearch/index/mapper/murmur3/Murmur3FieldMapperTests.java b/plugins/mapper-murmur3/src/test/java/org/elasticsearch/index/mapper/murmur3/Murmur3FieldMapperTests.java
index 0992020e5b751..3be80fbdd7b1f 100644
--- a/plugins/mapper-murmur3/src/test/java/org/elasticsearch/index/mapper/murmur3/Murmur3FieldMapperTests.java
+++ b/plugins/mapper-murmur3/src/test/java/org/elasticsearch/index/mapper/murmur3/Murmur3FieldMapperTests.java
@@ -71,7 +71,7 @@ public void setup() {
             return indexService.newQueryShardContext(0, null, () -> { throw new UnsupportedOperationException(); }, null);
         };
         parser = new DocumentMapperParser(indexService.getIndexSettings(), indexService.mapperService(), indexService.xContentRegistry(),
-                indexService.similarityService(), mapperRegistry, queryShardContext);
+                indexService.similarityService(), mapperRegistry, queryShardContext, null);
     }
 
     @Override
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/10_basic.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/10_basic.yml
index 6186b3e1eb823..9b535dd21c436 100644
--- a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/10_basic.yml
+++ b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/10_basic.yml
@@ -63,7 +63,9 @@ setup:
   - do:
       index:
         index:  simple-data-stream1
-        body:   { foo: bar }
+        body:
+          '@timestamp': '2020-12-12'
+          foo: bar
 
   - do:
       indices.refresh:
@@ -241,27 +243,27 @@ setup:
   - do:
       index:
         index:  logs-foobar
-        body:   { foo: bar }
+        body:   { timestamp: '2020-12-12' }
   - match: { _index: .ds-logs-foobar-000001 }
 
   - do:
       catch: bad_request
       index:
         index:  .ds-logs-foobar-000001
-        body:   { foo: bar }
+        body:   { timestamp: '2020-12-12' }
 
   - do:
       bulk:
         body:
           - create:
               _index: .ds-logs-foobar-000001
-          - foo: bar
+          - timestamp: '2020-12-12'
           - index:
               _index: .ds-logs-foobar-000001
-          - foo: bar
+          - timestamp: '2020-12-12'
           - create:
               _index: logs-foobar
-          - foo: bar
+          - timestamp: '2020-12-12'
   - match: { errors: true }
   - match: { items.0.create.status: 400 }
   - match: { items.0.create.error.type: illegal_argument_exception }
@@ -276,3 +278,58 @@ setup:
       indices.delete_data_stream:
         name: logs-foobar
   - is_true: acknowledged
+
+---
+"Indexing a document into a data stream without a timestamp field":
+  - skip:
+      version: " - 7.9.99"
+      reason: "enable in 7.9+ when backported"
+      features: allowed_warnings
+
+  - do:
+      allowed_warnings:
+        - "index template [generic_logs_template] has index patterns [logs-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [generic_logs_template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: generic_logs_template
+        body:
+          index_patterns: logs-*
+          template:
+            mappings:
+              properties:
+                'timestamp':
+                  type: date
+          data_stream:
+            timestamp_field: timestamp
+
+  - do:
+      catch: bad_request
+      index:
+        index:  logs-foobar
+        body:   { foo: bar }
+
+  - do:
+      bulk:
+        body:
+          - create:
+              _index: logs-foobar
+          - foo: bar
+          - create:
+              _index: logs-foobar
+          - timestamp: '2020-12-12'
+          - create:
+              _index: logs-foobar
+          - timestamp: ['2020-12-12', '2022-12-12']
+  - match: { errors: true }
+  - match: { items.0.create.status: 400 }
+  - match: { items.0.create.error.type: illegal_argument_exception }
+  - match: { items.0.create.error.reason: "data stream timestamp field [timestamp] is missing" }
+  - match: { items.1.create.result: created }
+  - match: { items.1.create._index: .ds-logs-foobar-000001 }
+  - match: { items.2.create.status: 400 }
+  - match: { items.2.create.error.caused_by.type: illegal_argument_exception }
+  - match: { items.2.create.error.caused_by.reason: "data stream timestamp field [timestamp] encountered multiple values" }
+
+  - do:
+      indices.delete_data_stream:
+        name: logs-foobar
+  - is_true: acknowledged
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml
index 32e0590f23973..51b0ec97e5b5d 100644
--- a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml
+++ b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml
@@ -29,7 +29,9 @@
       index:
         index:  logs-foobar
         refresh: true
-        body:   { foo: bar }
+        body:
+          '@timestamp': '2020-12-12'
+          foo: bar
   - match: {_index: .ds-logs-foobar-000001}
 
   - do:
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/30_auto_create_data_stream.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/30_auto_create_data_stream.yml
index 6465988ecbdf2..63d070c4edd49 100644
--- a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/30_auto_create_data_stream.yml
+++ b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/30_auto_create_data_stream.yml
@@ -27,7 +27,9 @@
       index:
         index:  logs-foobar
         refresh: true
-        body:   { foo: bar }
+        body:
+          'timestamp': '2020-12-12'
+          foo: bar
 
   - do:
       search:
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java b/server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java
index d4f4c8680008b..22b2fa2d09aff 100644
--- a/server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java
+++ b/server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java
@@ -233,28 +233,28 @@ public void testMixedAutoCreate() throws Exception {
         client().execute(PutComposableIndexTemplateAction.INSTANCE, createTemplateRequest).actionGet();
 
         BulkRequest bulkRequest = new BulkRequest();
-        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-foobaz").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-barfoo").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
         BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
         assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
 
         bulkRequest = new BulkRequest();
-        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-foobaz2").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-barfoo2").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz2").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo2").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
         bulkResponse = client().bulk(bulkRequest).actionGet();
         assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
 
         bulkRequest = new BulkRequest();
-        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-foobaz2").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-foobaz3").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-barfoo2").opType(CREATE).source("{}", XContentType.JSON));
-        bulkRequest.add(new IndexRequest("logs-barfoo3").opType(CREATE).source("{}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobar").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz2").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-foobaz3").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barbaz").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo2").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
+        bulkRequest.add(new IndexRequest("logs-barfoo3").opType(CREATE).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON));
         bulkResponse = client().bulk(bulkRequest).actionGet();
         assertThat("bulk failures: " + Strings.toString(bulkResponse), bulkResponse.hasFailures(), is(false));
 
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/indices/DataStreamIT.java b/server/src/internalClusterTest/java/org/elasticsearch/indices/DataStreamIT.java
index 3fdedf6bafcc0..7184d09307c40 100644
--- a/server/src/internalClusterTest/java/org/elasticsearch/indices/DataStreamIT.java
+++ b/server/src/internalClusterTest/java/org/elasticsearch/indices/DataStreamIT.java
@@ -54,6 +54,8 @@
 import org.elasticsearch.common.xcontent.ObjectPath;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.mapper.DateFieldMapper;
+import org.elasticsearch.index.mapper.MapperParsingException;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.junit.After;
@@ -63,6 +65,7 @@
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Optional;
 import java.util.Map;
 
@@ -80,6 +83,7 @@
 import static org.elasticsearch.indices.IndicesOptionsIntegrationIT.segments;
 import static org.elasticsearch.indices.IndicesOptionsIntegrationIT.validateQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
@@ -139,9 +143,9 @@ public void testBasicScenario() throws Exception {
         assertThat(ObjectPath.eval("properties.@timestamp1.type", mappings), is("date"));
 
         int numDocsBar = randomIntBetween(2, 16);
-        indexDocs("metrics-bar", numDocsBar);
+        indexDocs("metrics-bar", "@timestamp2", numDocsBar);
         int numDocsFoo = randomIntBetween(2, 16);
-        indexDocs("metrics-foo", numDocsFoo);
+        indexDocs("metrics-foo", "@timestamp1", numDocsFoo);
 
         verifyDocs("metrics-bar", numDocsBar, 1, 1);
         verifyDocs("metrics-foo", numDocsFoo, 1, 1);
@@ -169,9 +173,9 @@ public void testBasicScenario() throws Exception {
         assertThat(ObjectPath.eval("properties.@timestamp2.type", mappings), is("date"));
 
         int numDocsBar2 = randomIntBetween(2, 16);
-        indexDocs("metrics-bar", numDocsBar2);
+        indexDocs("metrics-bar", "@timestamp2", numDocsBar2);
         int numDocsFoo2 = randomIntBetween(2, 16);
-        indexDocs("metrics-foo", numDocsFoo2);
+        indexDocs("metrics-foo", "@timestamp1", numDocsFoo2);
 
         verifyDocs("metrics-bar", numDocsBar + numDocsBar2, 1, 2);
         verifyDocs("metrics-foo", numDocsFoo + numDocsFoo2, 1, 2);
@@ -203,7 +207,7 @@ public void testOtherWriteOps() throws Exception {
 
         {
             BulkRequest bulkRequest = new BulkRequest()
-                .add(new IndexRequest(dataStreamName).source("{}", XContentType.JSON));
+                .add(new IndexRequest(dataStreamName).source("{\"@timestamp1\": \"2020-12-12\"}", XContentType.JSON));
             expectFailure(dataStreamName, () -> client().bulk(bulkRequest).actionGet());
         }
         {
@@ -213,11 +217,12 @@ public void testOtherWriteOps() throws Exception {
         }
         {
             BulkRequest bulkRequest = new BulkRequest()
-                .add(new UpdateRequest(dataStreamName, "_id").doc("{}", XContentType.JSON));
+                .add(new UpdateRequest(dataStreamName, "_id").doc("{\"@timestamp1\": \"2020-12-12\"}", XContentType.JSON));
             expectFailure(dataStreamName, () -> client().bulk(bulkRequest).actionGet());
         }
         {
-            IndexRequest indexRequest = new IndexRequest(dataStreamName).source("{}", XContentType.JSON);
+            IndexRequest indexRequest = new IndexRequest(dataStreamName)
+                .source("{\"@timestamp1\": \"2020-12-12\"}", XContentType.JSON);
             expectFailure(dataStreamName, () -> client().index(indexRequest).actionGet());
         }
         {
@@ -230,14 +235,15 @@ public void testOtherWriteOps() throws Exception {
             expectFailure(dataStreamName, () -> client().delete(deleteRequest).actionGet());
         }
         {
-            IndexRequest indexRequest = new IndexRequest(dataStreamName).source("{}", XContentType.JSON)
+            IndexRequest indexRequest = new IndexRequest(dataStreamName)
+                .source("{\"@timestamp1\": \"2020-12-12\"}", XContentType.JSON)
                 .opType(DocWriteRequest.OpType.CREATE);
             IndexResponse indexResponse = client().index(indexRequest).actionGet();
             assertThat(indexResponse.getIndex(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 1)));
         }
         {
             BulkRequest bulkRequest = new BulkRequest()
-                .add(new IndexRequest(dataStreamName).source("{}", XContentType.JSON)
+                .add(new IndexRequest(dataStreamName).source("{\"@timestamp1\": \"2020-12-12\"}", XContentType.JSON)
                     .opType(DocWriteRequest.OpType.CREATE));
             BulkResponse bulkItemResponses  = client().bulk(bulkRequest).actionGet();
             assertThat(bulkItemResponses.getItems()[0].getIndex(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 1)));
@@ -274,7 +280,7 @@ public void testComposableTemplateOnlyMatchingWithDataStreamName() throws Except
         client().execute(PutComposableIndexTemplateAction.INSTANCE, request).actionGet();
 
         int numDocs = randomIntBetween(2, 16);
-        indexDocs(dataStreamName, numDocs);
+        indexDocs(dataStreamName, "@timestamp", numDocs);
         verifyDocs(dataStreamName, numDocs, 1, 1);
 
         String backingIndex = DataStream.getDefaultBackingIndexName(dataStreamName, 1);
@@ -305,7 +311,7 @@ public void testComposableTemplateOnlyMatchingWithDataStreamName() throws Except
             getIndexResponse.mappings().get(backingIndex).getSourceAsMap()), equalTo("keyword"));
 
         int numDocs2 = randomIntBetween(2, 16);
-        indexDocs(dataStreamName, numDocs2);
+        indexDocs(dataStreamName, "@timestamp", numDocs2);
         verifyDocs(dataStreamName, numDocs + numDocs2, 1, 2);
 
         DeleteDataStreamAction.Request deleteDataStreamRequest = new DeleteDataStreamAction.Request(dataStreamName);
@@ -368,7 +374,7 @@ public void testResolvabilityOfDataStreamsInAPIs() throws Exception {
         client().admin().indices().createDataStream(request).actionGet();
 
         verifyResolvability(dataStreamName, client().prepareIndex(dataStreamName)
-                .setSource("{}", XContentType.JSON)
+                .setSource("{\"ts\": \"2020-12-12\"}", XContentType.JSON)
                 .setOpType(DocWriteRequest.OpType.CREATE),
             false);
         verifyResolvability(dataStreamName, refreshBuilder(dataStreamName), false);
@@ -395,7 +401,7 @@ public void testResolvabilityOfDataStreamsInAPIs() throws Exception {
         request = new CreateDataStreamAction.Request("logs-barbaz");
         client().admin().indices().createDataStream(request).actionGet();
         verifyResolvability("logs-barbaz", client().prepareIndex("logs-barbaz")
-                .setSource("{}", XContentType.JSON)
+                .setSource("{\"ts\": \"2020-12-12\"}", XContentType.JSON)
                 .setOpType(DocWriteRequest.OpType.CREATE),
             false);
 
@@ -479,7 +485,8 @@ public void testChangeTimestampFieldInComposableTemplatePriorToRollOver() throws
         putComposableIndexTemplate("id1", "@timestamp", List.of("logs-foo*"));
 
         // Index doc that triggers creation of a data stream
-        IndexRequest indexRequest = new IndexRequest("logs-foobar").source("{}", XContentType.JSON).opType("create");
+        IndexRequest indexRequest =
+            new IndexRequest("logs-foobar").source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON).opType("create");
         IndexResponse indexResponse = client().index(indexRequest).actionGet();
         assertThat(indexResponse.getIndex(), equalTo(DataStream.getDefaultBackingIndexName("logs-foobar", 1)));
         assertBackingIndex(DataStream.getDefaultBackingIndexName("logs-foobar", 1), "properties.@timestamp");
@@ -491,7 +498,7 @@ public void testChangeTimestampFieldInComposableTemplatePriorToRollOver() throws
         assertBackingIndex(DataStream.getDefaultBackingIndexName("logs-foobar", 2), "properties.@timestamp");
 
         // Index another doc into a data stream
-        indexRequest = new IndexRequest("logs-foobar").source("{}", XContentType.JSON).opType("create");
+        indexRequest = new IndexRequest("logs-foobar").source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON).opType("create");
         indexResponse = client().index(indexRequest).actionGet();
         assertThat(indexResponse.getIndex(), equalTo(DataStream.getDefaultBackingIndexName("logs-foobar", 2)));
 
@@ -506,7 +513,7 @@ public void testChangeTimestampFieldInComposableTemplatePriorToRollOver() throws
         assertBackingIndex(DataStream.getDefaultBackingIndexName("logs-foobar", 3), "properties.@timestamp");
 
         // Index another doc into a data stream
-        indexRequest = new IndexRequest("logs-foobar").source("{}", XContentType.JSON).opType("create");
+        indexRequest = new IndexRequest("logs-foobar").source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON).opType("create");
         indexResponse = client().index(indexRequest).actionGet();
         assertThat(indexResponse.getIndex(), equalTo(DataStream.getDefaultBackingIndexName("logs-foobar", 3)));
 
@@ -601,6 +608,33 @@ private static void assertBackingIndex(String backingIndex, String timestampFiel
         assertThat(ObjectPath.eval(timestampFieldPathInMapping, mappings), is(expectedMapping));
     }
 
+    public void testNoTimestampInDocument() throws Exception {
+        putComposableIndexTemplate("id", "@timestamp", List.of("logs-foobar*"));
+        String dataStreamName = "logs-foobar";
+        CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
+        client().admin().indices().createDataStream(createDataStreamRequest).get();
+
+        IndexRequest indexRequest = new IndexRequest(dataStreamName)
+            .opType("create")
+            .source("{}", XContentType.JSON);
+        Exception e = expectThrows(IllegalArgumentException.class, () -> client().index(indexRequest).actionGet());
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] is missing"));
+    }
+
+    public void testMultipleTimestampValuesInDocument() throws Exception {
+        putComposableIndexTemplate("id", "@timestamp", List.of("logs-foobar*"));
+        String dataStreamName = "logs-foobar";
+        CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
+        client().admin().indices().createDataStream(createDataStreamRequest).get();
+
+        IndexRequest indexRequest = new IndexRequest(dataStreamName)
+            .opType("create")
+            .source("{\"@timestamp\": [\"2020-12-12\",\"2022-12-12\"]}", XContentType.JSON);
+        Exception e = expectThrows(MapperParsingException.class, () -> client().index(indexRequest).actionGet());
+        assertThat(e.getCause().getMessage(),
+            containsString("data stream timestamp field [@timestamp] encountered multiple values"));
+    }
+
     private static void verifyResolvability(String dataStream, ActionRequestBuilder requestBuilder, boolean fail) {
         verifyResolvability(dataStream, requestBuilder, fail, 0);
     }
@@ -635,12 +669,13 @@ private static void verifyResolvability(String dataStream, ActionRequestBuilder
         }
     }
 
-    private static void indexDocs(String dataStream, int numDocs) {
+    private static void indexDocs(String dataStream, String timestampField, int numDocs) {
         BulkRequest bulkRequest = new BulkRequest();
         for (int i = 0; i < numDocs; i++) {
+            String value = DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.formatMillis(System.currentTimeMillis());
             bulkRequest.add(new IndexRequest(dataStream)
                 .opType(DocWriteRequest.OpType.CREATE)
-                .source("{}", XContentType.JSON));
+                .source(String.format(Locale.ROOT, "{\"%s\":\"%s\"}", timestampField, value), XContentType.JSON));
         }
         BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
         assertThat(bulkResponse.getItems().length, equalTo(numDocs));
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java
index a2018f8d7e619..28e2968d6f5a9 100644
--- a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java
+++ b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java
@@ -2529,7 +2529,7 @@ public void testDeleteDataStreamDuringSnapshot() throws Exception {
             client.prepareIndex(dataStream)
                 .setOpType(DocWriteRequest.OpType.CREATE)
                 .setId(Integer.toString(i))
-                .setSource(Collections.singletonMap("k", "v"))
+                .setSource(Collections.singletonMap("@timestamp", "2020-12-12"))
                 .execute().actionGet();
         }
         refresh();
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexAliasesService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexAliasesService.java
index 2ea44d0440119..b4b3c8f996464 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexAliasesService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexAliasesService.java
@@ -142,7 +142,7 @@ public ClusterState applyAliasActions(ClusterState currentState, Iterable<AliasA
                             if (indexService == null) {
                                 // temporarily create the index and add mappings so we can parse the filter
                                 try {
-                                    indexService = indicesService.createIndex(index, emptyList(), false);
+                                    indexService = indicesService.createIndex(index, emptyList(), false, null);
                                     indicesToClose.add(index.getIndex());
                                 } catch (IOException e) {
                                     throw new ElasticsearchException("Failed to create temporary index for parsing the alias", e);
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateService.java
index d8b577bbdc914..52378b0d6edca 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateService.java
@@ -1097,7 +1097,7 @@ private static void validateTemplate(Settings validateSettings, String mappings,
                 .build();
 
             final IndexMetadata tmpIndexMetadata = IndexMetadata.builder(temporaryIndexName).settings(dummySettings).build();
-            IndexService dummyIndexService = indicesService.createIndex(tmpIndexMetadata, Collections.emptyList(), false);
+            IndexService dummyIndexService = indicesService.createIndex(tmpIndexMetadata, Collections.emptyList(), false, null);
             createdIndex = dummyIndexService.index();
 
             if (mappings != null) {
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexUpgradeService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexUpgradeService.java
index 58b820beb9eb2..3babc5a324e2e 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexUpgradeService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexUpgradeService.java
@@ -181,7 +181,7 @@ public Set<Entry<String, NamedAnalyzer>> entrySet() {
             try (IndexAnalyzers fakeIndexAnalzyers =
                      new IndexAnalyzers(analyzerMap, analyzerMap, analyzerMap)) {
                 MapperService mapperService = new MapperService(indexSettings, fakeIndexAnalzyers, xContentRegistry, similarityService,
-                        mapperRegistry, () -> null, () -> false);
+                        mapperRegistry, () -> null, () -> false, null);
                 mapperService.merge(indexMetadata, MapperService.MergeReason.MAPPING_RECOVERY);
             }
         } catch (Exception ex) {
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMappingService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMappingService.java
index 067c6c85cb954..50fcbeb2caf8a 100644
--- a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMappingService.java
+++ b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMappingService.java
@@ -144,7 +144,7 @@ ClusterState executeRefresh(final ClusterState currentState, final List<RefreshT
             IndexService indexService = indicesService.indexService(indexMetadata.getIndex());
             if (indexService == null) {
                 // we need to create the index here, and add the current mapping to it, so we can merge
-                indexService = indicesService.createIndex(indexMetadata, Collections.emptyList(), false);
+                indexService = indicesService.createIndex(indexMetadata, Collections.emptyList(), false, null);
                 removeIndex = true;
                 indexService.mapperService().merge(indexMetadata, MergeReason.MAPPING_RECOVERY);
             }
@@ -198,7 +198,7 @@ private boolean refreshIndexMapping(IndexService indexService, IndexMetadata.Bui
      */
     public void refreshMapping(final String index, final String indexUUID) {
         final RefreshTask refreshTask = new RefreshTask(index, indexUUID);
-        clusterService.submitStateUpdateTask("refresh-mapping [" + index + "]",  
+        clusterService.submitStateUpdateTask("refresh-mapping [" + index + "]",
             refreshTask,
             ClusterStateTaskConfig.build(Priority.HIGH),
             refreshExecutor,
diff --git a/server/src/main/java/org/elasticsearch/index/IndexModule.java b/server/src/main/java/org/elasticsearch/index/IndexModule.java
index 30bd74e3edd2c..9cf9ee06210bc 100644
--- a/server/src/main/java/org/elasticsearch/index/IndexModule.java
+++ b/server/src/main/java/org/elasticsearch/index/IndexModule.java
@@ -134,11 +134,11 @@ public final class IndexModule {
     private final IndexNameExpressionResolver expressionResolver;
     private final AtomicBoolean frozen = new AtomicBoolean(false);
     private final BooleanSupplier allowExpensiveQueries;
+    private final String dataStreamTimestampField;
 
     /**
      * Construct the index module for the index with the specified index settings. The index module contains extension points for plugins
      * via {@link org.elasticsearch.plugins.PluginsService#onIndexModule(IndexModule)}.
-     *
      * @param indexSettings       the index settings
      * @param analysisRegistry    the analysis registry
      * @param engineFactory       the engine factory
@@ -150,7 +150,8 @@ public IndexModule(
             final EngineFactory engineFactory,
             final Map<String, IndexStorePlugin.DirectoryFactory> directoryFactories,
             final BooleanSupplier allowExpensiveQueries,
-            final IndexNameExpressionResolver expressionResolver) {
+            final IndexNameExpressionResolver expressionResolver,
+            final String dataStreamTimestampField) {
         this.indexSettings = indexSettings;
         this.analysisRegistry = analysisRegistry;
         this.engineFactory = Objects.requireNonNull(engineFactory);
@@ -159,6 +160,7 @@ public IndexModule(
         this.directoryFactories = Collections.unmodifiableMap(directoryFactories);
         this.allowExpensiveQueries = allowExpensiveQueries;
         this.expressionResolver = expressionResolver;
+        this.dataStreamTimestampField = dataStreamTimestampField;
     }
 
     /**
@@ -427,12 +429,12 @@ public IndexService newIndexService(IndexService.IndexCreationContext indexCreat
             if (IndexService.needsMapperService(indexSettings, indexCreationContext)) {
                 indexAnalyzers = analysisRegistry.build(indexSettings);
             }
-            final IndexService indexService = new IndexService(indexSettings, indexCreationContext, environment, xContentRegistry,
-                new SimilarityService(indexSettings, scriptService, similarities), shardStoreDeleter, indexAnalyzers,
+            final IndexService indexService = new IndexService(indexSettings, indexCreationContext, environment,
+                xContentRegistry, new SimilarityService(indexSettings, scriptService, similarities), shardStoreDeleter, indexAnalyzers,
                 engineFactory, circuitBreakerService, bigArrays, threadPool, scriptService, clusterService, client, queryCache,
                 directoryFactory, eventListener, readerWrapperFactory, mapperRegistry, indicesFieldDataCache, searchOperationListeners,
                 indexOperationListeners, namedWriteableRegistry, idFieldDataEnabled, allowExpensiveQueries, expressionResolver,
-                valuesSourceRegistry);
+                valuesSourceRegistry, dataStreamTimestampField);
             success = true;
             return indexService;
         } finally {
@@ -476,10 +478,12 @@ private static IndexStorePlugin.DirectoryFactory getDirectoryFactory(
      * doing so will result in an exception.
      */
     public MapperService newIndexMapperService(NamedXContentRegistry xContentRegistry, MapperRegistry mapperRegistry,
-            ScriptService scriptService) throws IOException {
+                                               ScriptService scriptService) throws IOException {
+
         return new MapperService(indexSettings, analysisRegistry.build(indexSettings), xContentRegistry,
             new SimilarityService(indexSettings, scriptService, similarities), mapperRegistry,
-            () -> { throw new UnsupportedOperationException("no index query shard context available"); }, () -> false);
+            () -> { throw new UnsupportedOperationException("no index query shard context available"); }, () -> false,
+            dataStreamTimestampField);
     }
 
     /**
diff --git a/server/src/main/java/org/elasticsearch/index/IndexService.java b/server/src/main/java/org/elasticsearch/index/IndexService.java
index f83ad21d5562e..c233c38168781 100644
--- a/server/src/main/java/org/elasticsearch/index/IndexService.java
+++ b/server/src/main/java/org/elasticsearch/index/IndexService.java
@@ -147,6 +147,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final IndexNameExpressionResolver expressionResolver;
     private Supplier<Sort> indexSortSupplier;
     private ValuesSourceRegistry valuesSourceRegistry;
+    private final String dataStreamTimestampField;
 
     public IndexService(
             IndexSettings indexSettings,
@@ -174,7 +175,8 @@ public IndexService(
             BooleanSupplier idFieldDataEnabled,
             BooleanSupplier allowExpensiveQueries,
             IndexNameExpressionResolver expressionResolver,
-            ValuesSourceRegistry valuesSourceRegistry) {
+            ValuesSourceRegistry valuesSourceRegistry,
+            String dataStreamTimestampField) {
         super(indexSettings);
         this.allowExpensiveQueries = allowExpensiveQueries;
         this.indexSettings = indexSettings;
@@ -188,7 +190,7 @@ public IndexService(
             assert indexAnalyzers != null;
             this.mapperService = new MapperService(indexSettings, indexAnalyzers, xContentRegistry, similarityService, mapperRegistry,
                 // we parse all percolator queries as they would be parsed on shard 0
-                () -> newQueryShardContext(0, null, System::currentTimeMillis, null), idFieldDataEnabled);
+                () -> newQueryShardContext(0, null, System::currentTimeMillis, null), idFieldDataEnabled, dataStreamTimestampField);
             this.indexFieldData = new IndexFieldDataService(indexSettings, indicesFieldDataCache, circuitBreakerService, mapperService);
             if (indexSettings.getIndexSortConfig().hasIndexSort()) {
                 // we delay the actual creation of the sort order for this index because the mapping has not been merged yet.
@@ -233,6 +235,7 @@ public IndexService(
         this.trimTranslogTask = new AsyncTrimTranslogTask(this);
         this.globalCheckpointTask = new AsyncGlobalCheckpointTask(this);
         this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
+        this.dataStreamTimestampField = dataStreamTimestampField;
         updateFsyncTaskIfNecessary();
     }
 
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/DateFieldMapper.java b/server/src/main/java/org/elasticsearch/index/mapper/DateFieldMapper.java
index 98f222e822477..3ef205b1b91b5 100644
--- a/server/src/main/java/org/elasticsearch/index/mapper/DateFieldMapper.java
+++ b/server/src/main/java/org/elasticsearch/index/mapper/DateFieldMapper.java
@@ -186,6 +186,7 @@ public static class Builder extends FieldMapper.Builder<Builder> {
         private Locale locale;
         private Resolution resolution = Resolution.MILLISECONDS;
         String nullValue;
+        private boolean singletonDataStreamTimestamp;
 
         public Builder(String name) {
             super(name, Defaults.FIELD_TYPE);
@@ -240,6 +241,10 @@ public boolean isFormatterSet() {
             return format.explicit();
         }
 
+        public void setSingletonDataStreamTimestamp(boolean singletonDataStreamTimestamp) {
+            this.singletonDataStreamTimestamp = singletonDataStreamTimestamp;
+        }
+
         protected DateFieldType setupFieldType(BuilderContext context) {
             String pattern = this.format.value();
             DateFormatter dateTimeFormatter = DateFormatter.forPattern(pattern).withLocale(locale);
@@ -251,7 +256,7 @@ public DateFieldMapper build(BuilderContext context) {
             DateFieldType ft = setupFieldType(context);
             Long nullTimestamp = nullValue == null ? null : ft.dateTimeFormatter.parseMillis(nullValue);
             return new DateFieldMapper(name, fieldType, ft, ignoreMalformed(context), nullTimestamp, nullValue,
-                multiFieldsBuilder.build(this, context), copyTo);
+                multiFieldsBuilder.build(this, context), copyTo, singletonDataStreamTimestamp);
         }
     }
 
@@ -291,6 +296,9 @@ public Mapper.Builder<?> parse(String name, Map<String, Object> node, ParserCont
                     iterator.remove();
                 }
             }
+            if (name.equals(parserContext.getDataStreamTimestampField())) {
+                builder.setSingletonDataStreamTimestamp(true);
+            }
             return builder;
         }
     }
@@ -550,6 +558,13 @@ public DocValueFormat docValueFormat(@Nullable String format, ZoneId timeZone) {
     private final Long nullValue;
     private final String nullValueAsString;
 
+    /**
+     * If this is set to true then that means this field mapper is used by a data stream's
+     * timestamp field. In this case the expectation is that exactly one date value is provided.
+     * No or more date values will result that indexing a document fails.
+     */
+    private final boolean singletonDataStreamTimestamp;
+
     private DateFieldMapper(
             String simpleName,
             FieldType fieldType,
@@ -557,11 +572,13 @@ private DateFieldMapper(
             Explicit<Boolean> ignoreMalformed,
             Long nullValue, String nullValueAsString,
             MultiFields multiFields,
-            CopyTo copyTo) {
+            CopyTo copyTo,
+            boolean singletonDataStreamTimestamp) {
         super(simpleName, fieldType, mappedFieldType, multiFields, copyTo);
         this.ignoreMalformed = ignoreMalformed;
         this.nullValue = nullValue;
         this.nullValueAsString = nullValueAsString;
+        this.singletonDataStreamTimestamp = singletonDataStreamTimestamp;
     }
 
     @Override
@@ -581,6 +598,13 @@ protected DateFieldMapper clone() {
 
     @Override
     protected void parseCreateField(ParseContext context) throws IOException {
+        if (singletonDataStreamTimestamp) {
+            if (context.isDataStreamTimestampParsed()) {
+                throw new IllegalArgumentException("data stream timestamp field [" + name() + "] encountered multiple values");
+            }
+            context.setDataStreamTimestampParsed(true);
+        }
+
         String dateAsString;
         if (context.externalValueSet()) {
             Object dateAsObject = context.externalValue();
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapper.java b/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapper.java
index 01d25da00685c..a13642cce61ce 100644
--- a/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapper.java
+++ b/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapper.java
@@ -53,7 +53,6 @@
 import java.util.Objects;
 import java.util.stream.Stream;
 
-
 public class DocumentMapper implements ToXContentFragment {
 
     public static class Builder {
@@ -66,7 +65,13 @@ public static class Builder {
 
         private final Mapper.BuilderContext builderContext;
 
+        private final String dataStreamTimestampField;
+
         public Builder(RootObjectMapper.Builder builder, MapperService mapperService) {
+            this(builder, mapperService, null);
+        }
+
+        public Builder(RootObjectMapper.Builder builder, MapperService mapperService, String dataStreamTimestampField) {
             final Settings indexSettings = mapperService.getIndexSettings().getSettings();
             this.builderContext = new Mapper.BuilderContext(indexSettings, new ContentPath(1));
             this.rootObjectMapper = builder.build(builderContext);
@@ -89,6 +94,7 @@ public Builder(RootObjectMapper.Builder builder, MapperService mapperService) {
                 }
                 metadataMappers.put(metadataMapper.getClass(), metadataMapper);
             }
+            this.dataStreamTimestampField = dataStreamTimestampField;
         }
 
         public Builder meta(Map<String, Object> meta) {
@@ -109,7 +115,7 @@ public DocumentMapper build(MapperService mapperService) {
                     rootObjectMapper,
                     metadataMappers.values().toArray(new MetadataFieldMapper[metadataMappers.values().size()]),
                     meta);
-            return new DocumentMapper(mapperService, mapping);
+            return new DocumentMapper(mapperService, mapping, dataStreamTimestampField);
         }
     }
 
@@ -122,6 +128,8 @@ public DocumentMapper build(MapperService mapperService) {
 
     private final Mapping mapping;
 
+    private final String dataStreamTimestampField;
+
     private final DocumentParser documentParser;
 
     private final DocumentFieldMappers fieldMappers;
@@ -132,12 +140,13 @@ public DocumentMapper build(MapperService mapperService) {
     private final MetadataFieldMapper[] deleteTombstoneMetadataFieldMappers;
     private final MetadataFieldMapper[] noopTombstoneMetadataFieldMappers;
 
-    public DocumentMapper(MapperService mapperService, Mapping mapping) {
+    public DocumentMapper(MapperService mapperService, Mapping mapping, String dataStreamTimestampField) {
         this.mapperService = mapperService;
         this.type = mapping.root().name();
         this.typeText = new Text(this.type);
         final IndexSettings indexSettings = mapperService.getIndexSettings();
         this.mapping = mapping;
+        this.dataStreamTimestampField = dataStreamTimestampField;
         this.documentParser = new DocumentParser(indexSettings, mapperService.documentMapperParser(), this);
 
         // collect all the mappers for this type
@@ -249,18 +258,22 @@ public Map<String, ObjectMapper> objectMappers() {
     }
 
     public ParsedDocument parse(SourceToParse source) throws MapperParsingException {
-        return documentParser.parseDocument(source, mapping.metadataMappers);
+        return documentParser.parseDocument(source, mapping.metadataMappers, dataStreamTimestampField);
     }
 
     public ParsedDocument createDeleteTombstoneDoc(String index, String id) throws MapperParsingException {
         final SourceToParse emptySource = new SourceToParse(index, id, new BytesArray("{}"), XContentType.JSON);
-        return documentParser.parseDocument(emptySource, deleteTombstoneMetadataFieldMappers).toTombstone();
+        // Supply null as dataStreamTimestampField, to avoid data stream timestamp field validation to fail
+        // (tombstone docs have an empty source)
+        return documentParser.parseDocument(emptySource, deleteTombstoneMetadataFieldMappers, null).toTombstone();
     }
 
     public ParsedDocument createNoopTombstoneDoc(String index, String reason) throws MapperParsingException {
         final String id = ""; // _id won't be used.
         final SourceToParse sourceToParse = new SourceToParse(index, id, new BytesArray("{}"), XContentType.JSON);
-        final ParsedDocument parsedDoc = documentParser.parseDocument(sourceToParse, noopTombstoneMetadataFieldMappers).toTombstone();
+        // Supply null as dataStreamTimestampField, to avoid data stream timestamp field validation to fail
+        // (tombstone docs have an empty source)
+        final ParsedDocument parsedDoc = documentParser.parseDocument(sourceToParse, noopTombstoneMetadataFieldMappers, null).toTombstone();
         // Store the reason of a noop as a raw string in the _source field
         final BytesRef byteRef = new BytesRef(reason);
         parsedDoc.rootDoc().add(new StoredField(SourceFieldMapper.NAME, byteRef.bytes, byteRef.offset, byteRef.length));
@@ -304,7 +317,7 @@ public ObjectMapper findNestedObjectMapper(int nestedDocId, SearchContext sc, Le
 
     public DocumentMapper merge(Mapping mapping) {
         Mapping merged = this.mapping.merge(mapping);
-        return new DocumentMapper(mapperService, merged);
+        return new DocumentMapper(mapperService, merged, dataStreamTimestampField);
     }
 
     @Override
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapperParser.java b/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapperParser.java
index d4f16bcaf8458..83a50cc9c437a 100644
--- a/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapperParser.java
+++ b/server/src/main/java/org/elasticsearch/index/mapper/DocumentMapperParser.java
@@ -52,9 +52,11 @@ public class DocumentMapperParser {
 
     private final Map<String, Mapper.TypeParser> typeParsers;
     private final Map<String, MetadataFieldMapper.TypeParser> rootTypeParsers;
+    private final String dataStreamTimestampField;
 
     public DocumentMapperParser(IndexSettings indexSettings, MapperService mapperService, NamedXContentRegistry xContentRegistry,
-            SimilarityService similarityService, MapperRegistry mapperRegistry, Supplier<QueryShardContext> queryShardContextSupplier) {
+                                SimilarityService similarityService, MapperRegistry mapperRegistry,
+                                Supplier<QueryShardContext> queryShardContextSupplier, String dataStreamTimestampField) {
         this.mapperService = mapperService;
         this.xContentRegistry = xContentRegistry;
         this.similarityService = similarityService;
@@ -62,11 +64,12 @@ public DocumentMapperParser(IndexSettings indexSettings, MapperService mapperSer
         this.typeParsers = mapperRegistry.getMapperParsers();
         this.indexVersionCreated = indexSettings.getIndexVersionCreated();
         this.rootTypeParsers = mapperRegistry.getMetadataMapperParsers(indexVersionCreated);
+        this.dataStreamTimestampField = dataStreamTimestampField;
     }
 
     public Mapper.TypeParser.ParserContext parserContext() {
         return new Mapper.TypeParser.ParserContext(similarityService::getSimilarity, mapperService,
-                typeParsers::get, indexVersionCreated, queryShardContextSupplier);
+                typeParsers::get, indexVersionCreated, queryShardContextSupplier, dataStreamTimestampField);
     }
 
     public DocumentMapper parse(@Nullable String type, CompressedXContent source) throws MapperParsingException {
@@ -91,8 +94,8 @@ private DocumentMapper parse(String type, Map<String, Object> mapping) throws Ma
 
         Mapper.TypeParser.ParserContext parserContext = parserContext();
         // parse RootObjectMapper
-        DocumentMapper.Builder docBuilder = new DocumentMapper.Builder(
-                (RootObjectMapper.Builder) rootObjectTypeParser.parse(type, mapping, parserContext), mapperService);
+        DocumentMapper.Builder docBuilder = new DocumentMapper.Builder((RootObjectMapper.Builder)
+            rootObjectTypeParser.parse(type, mapping, parserContext), mapperService, dataStreamTimestampField);
         Iterator<Map.Entry<String, Object>> iterator = mapping.entrySet().iterator();
         // parse DocumentMapper
         while(iterator.hasNext()) {
@@ -203,4 +206,8 @@ private Tuple<String, Map<String, Object>> extractMapping(String type, Map<Strin
     NamedXContentRegistry getXContentRegistry() {
         return xContentRegistry;
     }
+
+    MapperService getMapperService() {
+        return mapperService;
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/DocumentParser.java b/server/src/main/java/org/elasticsearch/index/mapper/DocumentParser.java
index fd23a0e661398..7d28362568c86 100644
--- a/server/src/main/java/org/elasticsearch/index/mapper/DocumentParser.java
+++ b/server/src/main/java/org/elasticsearch/index/mapper/DocumentParser.java
@@ -47,13 +47,17 @@ final class DocumentParser {
     private final DocumentMapperParser docMapperParser;
     private final DocumentMapper docMapper;
 
-    DocumentParser(IndexSettings indexSettings, DocumentMapperParser docMapperParser, DocumentMapper docMapper) {
+    DocumentParser(IndexSettings indexSettings,
+                   DocumentMapperParser docMapperParser,
+                   DocumentMapper docMapper) {
         this.indexSettings = indexSettings;
         this.docMapperParser = docMapperParser;
         this.docMapper = docMapper;
     }
 
-    ParsedDocument parseDocument(SourceToParse source, MetadataFieldMapper[] metadataFieldsMappers) throws MapperParsingException {
+    ParsedDocument parseDocument(SourceToParse source,
+                                 MetadataFieldMapper[] metadataFieldsMappers,
+                                 String dataStreamTimestampField) throws MapperParsingException {
 
         final Mapping mapping = docMapper.mapping();
         final ParseContext.InternalParseContext context;
@@ -62,6 +66,7 @@ ParsedDocument parseDocument(SourceToParse source, MetadataFieldMapper[] metadat
         try (XContentParser parser = XContentHelper.createParser(docMapperParser.getXContentRegistry(),
             LoggingDeprecationHandler.INSTANCE, source.source(), xContentType)) {
             context = new ParseContext.InternalParseContext(indexSettings, docMapperParser, docMapper, source, parser);
+            context.setDataStreamTimestampField(dataStreamTimestampField);
             validateStart(parser);
             internalParseDocument(mapping, metadataFieldsMappers, context, parser);
             validateEnd(parser);
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/Mapper.java b/server/src/main/java/org/elasticsearch/index/mapper/Mapper.java
index 7b432031238c6..c0af4bde7857b 100644
--- a/server/src/main/java/org/elasticsearch/index/mapper/Mapper.java
+++ b/server/src/main/java/org/elasticsearch/index/mapper/Mapper.java
@@ -88,14 +88,18 @@ class ParserContext {
 
             private final Supplier<QueryShardContext> queryShardContextSupplier;
 
+            private final String dataStreamTimestampField;
+
             public ParserContext(Function<String, SimilarityProvider> similarityLookupService,
                                  MapperService mapperService, Function<String, TypeParser> typeParsers,
-                                 Version indexVersionCreated, Supplier<QueryShardContext> queryShardContextSupplier) {
+                                 Version indexVersionCreated, Supplier<QueryShardContext> queryShardContextSupplier,
+                                 String dataStreamTimestampField) {
                 this.similarityLookupService = similarityLookupService;
                 this.mapperService = mapperService;
                 this.typeParsers = typeParsers;
                 this.indexVersionCreated = indexVersionCreated;
                 this.queryShardContextSupplier = queryShardContextSupplier;
+                this.dataStreamTimestampField = dataStreamTimestampField;
             }
 
             public IndexAnalyzers getIndexAnalyzers() {
@@ -132,10 +136,14 @@ public ParserContext createMultiFieldContext(ParserContext in) {
                 return new MultiFieldParserContext(in);
             }
 
+            public String getDataStreamTimestampField() {
+                return dataStreamTimestampField;
+            }
+
             static class MultiFieldParserContext extends ParserContext {
                 MultiFieldParserContext(ParserContext in) {
                     super(in.similarityLookupService(), in.mapperService(), in.typeParsers(),
-                            in.indexVersionCreated(), in.queryShardContextSupplier());
+                            in.indexVersionCreated(), in.queryShardContextSupplier(), in.getDataStreamTimestampField());
                 }
 
                 @Override
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/MapperService.java b/server/src/main/java/org/elasticsearch/index/mapper/MapperService.java
index 205f9eed2ca59..ce58437362c03 100644
--- a/server/src/main/java/org/elasticsearch/index/mapper/MapperService.java
+++ b/server/src/main/java/org/elasticsearch/index/mapper/MapperService.java
@@ -130,13 +130,14 @@ public enum MergeReason {
 
     public MapperService(IndexSettings indexSettings, IndexAnalyzers indexAnalyzers, NamedXContentRegistry xContentRegistry,
                          SimilarityService similarityService, MapperRegistry mapperRegistry,
-                         Supplier<QueryShardContext> queryShardContextSupplier, BooleanSupplier idFieldDataEnabled) {
+                         Supplier<QueryShardContext> queryShardContextSupplier, BooleanSupplier idFieldDataEnabled,
+                         String dataStreamTimestampField) {
         super(indexSettings);
         this.indexVersionCreated = indexSettings.getIndexVersionCreated();
         this.indexAnalyzers = indexAnalyzers;
         this.fieldTypes = new FieldTypeLookup();
         this.documentParser = new DocumentMapperParser(indexSettings, this, xContentRegistry, similarityService, mapperRegistry,
-                queryShardContextSupplier);
+                queryShardContextSupplier, dataStreamTimestampField);
         this.indexAnalyzer = new MapperAnalyzerWrapper(indexAnalyzers.getDefaultIndexAnalyzer(), p -> p.indexAnalyzer());
         this.searchAnalyzer = new MapperAnalyzerWrapper(indexAnalyzers.getDefaultSearchAnalyzer(), p -> p.searchAnalyzer());
         this.searchQuoteAnalyzer = new MapperAnalyzerWrapper(indexAnalyzers.getDefaultSearchQuoteAnalyzer(), p -> p.searchQuoteAnalyzer());
diff --git a/server/src/main/java/org/elasticsearch/index/mapper/ParseContext.java b/server/src/main/java/org/elasticsearch/index/mapper/ParseContext.java
index 03fa7f1f2ec62..e5bde877d95a3 100644
--- a/server/src/main/java/org/elasticsearch/index/mapper/ParseContext.java
+++ b/server/src/main/java/org/elasticsearch/index/mapper/ParseContext.java
@@ -285,6 +285,16 @@ public void addIgnoredField(String field) {
         public Collection<String> getIgnoredFields() {
             return in.getIgnoredFields();
         }
+
+        @Override
+        public boolean isDataStreamTimestampParsed() {
+            return in.isDataStreamTimestampParsed();
+        }
+
+        @Override
+        public void setDataStreamTimestampParsed(boolean dataStreamTimestampParsed) {
+            in.setDataStreamTimestampParsed(dataStreamTimestampParsed);
+        }
     }
 
     public static class InternalParseContext extends ParseContext {
@@ -319,6 +329,10 @@ public static class InternalParseContext extends ParseContext {
 
         private final Set<String> ignoredFields = new HashSet<>();
 
+        private String dataStreamTimestampField;
+
+        private boolean dataStreamTimestampParsed;
+
         public InternalParseContext(IndexSettings indexSettings, DocumentMapperParser docMapperParser, DocumentMapper docMapper,
                                     SourceToParse source, XContentParser parser) {
             this.indexSettings = indexSettings;
@@ -399,7 +413,7 @@ public DocumentMapper docMapper() {
 
         @Override
         public MapperService mapperService() {
-            return docMapperParser.mapperService;
+            return docMapperParser.getMapperService();
         }
 
         @Override
@@ -441,6 +455,9 @@ public Iterable<Document> nonRootDocuments() {
         }
 
         void postParse() {
+            if (dataStreamTimestampField != null && dataStreamTimestampParsed == false) {
+                throw new IllegalArgumentException("data stream timestamp field [" + dataStreamTimestampField + "] is missing");
+            }
             if (documents.size() > 1) {
                 docsReversed = true;
                 // We preserve the order of the children while ensuring that parents appear after them.
@@ -482,6 +499,18 @@ public void addIgnoredField(String field) {
         public Collection<String> getIgnoredFields() {
             return Collections.unmodifiableCollection(ignoredFields);
         }
+
+        public void setDataStreamTimestampField(String dataStreamTimestampField) {
+            this.dataStreamTimestampField = dataStreamTimestampField;
+        }
+
+        public boolean isDataStreamTimestampParsed() {
+            return dataStreamTimestampParsed;
+        }
+
+        public void setDataStreamTimestampParsed(boolean dataStreamTimestampParsed) {
+            this.dataStreamTimestampParsed = dataStreamTimestampParsed;
+        }
     }
 
     /**
@@ -646,4 +675,8 @@ public final <T> T parseExternalValue(Class<T> clazz) {
      * Get dynamic mappers created while parsing.
      */
     public abstract List<Mapper> getDynamicMappers();
+
+    public abstract boolean isDataStreamTimestampParsed();
+
+    public abstract void setDataStreamTimestampParsed(boolean dataStreamTimestampParsed);
 }
diff --git a/server/src/main/java/org/elasticsearch/indices/IndicesService.java b/server/src/main/java/org/elasticsearch/indices/IndicesService.java
index 6adb4e60dbcac..9b51f7906dc2b 100644
--- a/server/src/main/java/org/elasticsearch/indices/IndicesService.java
+++ b/server/src/main/java/org/elasticsearch/indices/IndicesService.java
@@ -38,6 +38,7 @@
 import org.elasticsearch.action.search.SearchType;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -150,6 +151,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.SortedMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -527,8 +529,8 @@ public IndexService indexServiceSafe(Index index) {
      */
     @Override
     public synchronized IndexService createIndex(
-            final IndexMetadata indexMetadata, final List<IndexEventListener> builtInListeners,
-            final boolean writeDanglingIndices) throws IOException {
+        final IndexMetadata indexMetadata, final List<IndexEventListener> builtInListeners,
+        final boolean writeDanglingIndices, String dataStreamTimestampField) throws IOException {
         ensureChangesAllowed();
         if (indexMetadata.getIndexUUID().equals(IndexMetadata.INDEX_UUID_NA_VALUE)) {
             throw new IllegalArgumentException("index must have a real UUID found value: [" + indexMetadata.getIndexUUID() + "]");
@@ -558,6 +560,7 @@ public void onStoreClosed(ShardId shardId) {
                 createIndexService(
                         CREATE_INDEX,
                         indexMetadata,
+                        dataStreamTimestampField,
                         indicesQueryCache,
                         indicesFieldDataCache,
                         finalListeners,
@@ -611,7 +614,7 @@ public void onStoreCreated(ShardId shardId) {
             createIndexService(
                 CREATE_INDEX,
                 indexMetadata,
-                indicesQueryCache,
+                null, indicesQueryCache,
                 indicesFieldDataCache,
                 finalListeners,
                 indexingMemoryController);
@@ -625,6 +628,7 @@ public void onStoreCreated(ShardId shardId) {
      */
     private synchronized IndexService createIndexService(IndexService.IndexCreationContext indexCreationContext,
                                                          IndexMetadata indexMetadata,
+                                                         String dataStreamTimestampField,
                                                          IndicesQueryCache indicesQueryCache,
                                                          IndicesFieldDataCache indicesFieldDataCache,
                                                          List<IndexEventListener> builtInListeners,
@@ -639,7 +643,7 @@ private synchronized IndexService createIndexService(IndexService.IndexCreationC
             indexCreationContext);
 
         final IndexModule indexModule = new IndexModule(idxSettings, analysisRegistry, getEngineFactory(idxSettings),
-                directoryFactories, () -> allowExpensiveQueries, indexNameExpressionResolver);
+                directoryFactories, () -> allowExpensiveQueries, indexNameExpressionResolver, dataStreamTimestampField);
         for (IndexingOperationListener operationListener : indexingOperationListeners) {
             indexModule.addIndexOperationListener(operationListener);
         }
@@ -708,13 +712,25 @@ private EngineFactory getEngineFactory(final IndexSettings idxSettings) {
      * Note: the returned {@link MapperService} should be closed when unneeded.
      */
     public synchronized MapperService createIndexMapperService(IndexMetadata indexMetadata) throws IOException {
+        String dataStreamTimestampField =
+            lookupDataStreamTimestampField(clusterService.state().metadata().getIndicesLookup(), indexMetadata.getIndex());
         final IndexSettings idxSettings = new IndexSettings(indexMetadata, this.settings, indexScopedSettings);
         final IndexModule indexModule = new IndexModule(idxSettings, analysisRegistry, getEngineFactory(idxSettings),
-                directoryFactories, () -> allowExpensiveQueries, indexNameExpressionResolver);
+                directoryFactories, () -> allowExpensiveQueries, indexNameExpressionResolver, null);
         pluginsService.onIndexModule(indexModule);
         return indexModule.newIndexMapperService(xContentRegistry, mapperRegistry, scriptService);
     }
 
+    public static String lookupDataStreamTimestampField(SortedMap<String, IndexAbstraction> indicesLookup, Index index) {
+        IndexAbstraction indexAbstraction = indicesLookup.get(index.getName());
+        if (indexAbstraction != null &&
+            indexAbstraction.getParentDataStream() != null) {
+            return indexAbstraction.getParentDataStream().getDataStream().getTimeStampField().getName();
+        } else {
+            return null;
+        }
+    }
+
     /**
      * This method verifies that the given {@code metadata} holds sane values to create an {@link IndexService}.
      * This method tries to update the meta data of the created {@link IndexService} if the given {@code metadataUpdate}
@@ -731,7 +747,7 @@ public synchronized void verifyIndexMetadata(IndexMetadata metadata, IndexMetada
             closeables.add(indicesQueryCache);
             // this will also fail if some plugin fails etc. which is nice since we can verify that early
             final IndexService service =
-                createIndexService(METADATA_VERIFICATION, metadata, indicesQueryCache, indicesFieldDataCache, emptyList());
+                createIndexService(METADATA_VERIFICATION, metadata, null, indicesQueryCache, indicesFieldDataCache, emptyList());
             closeables.add(() -> service.close("metadata verification", false));
             service.mapperService().merge(metadata, MapperService.MergeReason.MAPPING_RECOVERY);
             if (metadata.equals(metadataUpdate) == false) {
diff --git a/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java b/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
index 26370878492c6..f8dd5f3d7e763 100644
--- a/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
+++ b/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
@@ -91,6 +91,7 @@
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
+import static org.elasticsearch.indices.IndicesService.lookupDataStreamTimestampField;
 import static org.elasticsearch.indices.cluster.IndicesClusterStateService.AllocatedIndices.IndexRemovalReason.CLOSED;
 import static org.elasticsearch.indices.cluster.IndicesClusterStateService.AllocatedIndices.IndexRemovalReason.DELETED;
 import static org.elasticsearch.indices.cluster.IndicesClusterStateService.AllocatedIndices.IndexRemovalReason.FAILURE;
@@ -495,9 +496,10 @@ private void createIndices(final ClusterState state) {
             final IndexMetadata indexMetadata = state.metadata().index(index);
             logger.debug("[{}] creating index", index);
 
+            final String dataStreamTimestampField = lookupDataStreamTimestampField(state.metadata().getIndicesLookup(), index);
             AllocatedIndex<? extends Shard> indexService = null;
             try {
-                indexService = indicesService.createIndex(indexMetadata, buildInIndexListener, true);
+                indexService = indicesService.createIndex(indexMetadata, buildInIndexListener, true, dataStreamTimestampField);
                 if (indexService.updateMapping(null, indexMetadata) && sendRefreshMapping) {
                     nodeMappingRefreshAction.nodeMappingRefresh(state.nodes().getMasterNode(),
                         new NodeMappingRefreshAction.NodeMappingRefreshRequest(indexMetadata.getIndex().getName(),
@@ -866,7 +868,7 @@ public interface AllocatedIndices<T extends Shard, U extends AllocatedIndex<T>>
          */
         U createIndex(IndexMetadata indexMetadata,
                       List<IndexEventListener> builtInIndexListener,
-                      boolean writeDanglingIndices) throws IOException;
+                      boolean writeDanglingIndices, String dataStreamTimestampField) throws IOException;
 
         /**
          * Verify that the contents on disk for the given index is deleted; if not, delete the contents.
diff --git a/server/src/test/java/org/elasticsearch/index/IndexModuleTests.java b/server/src/test/java/org/elasticsearch/index/IndexModuleTests.java
index 58d3802cdec6a..66169830c63d1 100644
--- a/server/src/test/java/org/elasticsearch/index/IndexModuleTests.java
+++ b/server/src/test/java/org/elasticsearch/index/IndexModuleTests.java
@@ -167,8 +167,8 @@ private IndexService newIndexService(IndexModule module) throws IOException {
 
     public void testWrapperIsBound() throws IOException {
         final MockEngineFactory engineFactory = new MockEngineFactory(AssertingDirectoryReader.class);
-        IndexModule module = new IndexModule(
-                indexSettings, emptyAnalysisRegistry, engineFactory, Collections.emptyMap(), () -> true, new IndexNameExpressionResolver());
+        IndexModule module = new IndexModule(indexSettings, emptyAnalysisRegistry, engineFactory, Collections.emptyMap(),
+            () -> true, new IndexNameExpressionResolver(), null);
         module.setReaderWrapper(s -> new Wrapper());
 
         IndexService indexService = newIndexService(module);
@@ -188,8 +188,8 @@ public void testRegisterIndexStore() throws IOException {
         final IndexSettings indexSettings = IndexSettingsModule.newIndexSettings(index, settings);
         final Map<String, IndexStorePlugin.DirectoryFactory> indexStoreFactories = singletonMap(
             "foo_store", new FooFunction());
-        final IndexModule module = new IndexModule(indexSettings, emptyAnalysisRegistry, new InternalEngineFactory(), indexStoreFactories,
-            () -> true, new IndexNameExpressionResolver());
+        final IndexModule module = new IndexModule(indexSettings, emptyAnalysisRegistry, new InternalEngineFactory(),
+            indexStoreFactories, () -> true, new IndexNameExpressionResolver(), null);
 
         final IndexService indexService = newIndexService(module);
         assertThat(indexService.getDirectoryFactory(), instanceOf(FooFunction.class));
@@ -487,7 +487,7 @@ public void testMmapNotAllowed() {
 
     private static IndexModule createIndexModule(IndexSettings indexSettings, AnalysisRegistry emptyAnalysisRegistry) {
         return new IndexModule(indexSettings, emptyAnalysisRegistry, new InternalEngineFactory(), Collections.emptyMap(), () -> true,
-            new IndexNameExpressionResolver());
+            new IndexNameExpressionResolver(), null);
     }
 
     class CustomQueryCache implements QueryCache {
diff --git a/server/src/test/java/org/elasticsearch/index/codec/CodecTests.java b/server/src/test/java/org/elasticsearch/index/codec/CodecTests.java
index 331a2c138f6f2..47cfea54c65a0 100644
--- a/server/src/test/java/org/elasticsearch/index/codec/CodecTests.java
+++ b/server/src/test/java/org/elasticsearch/index/codec/CodecTests.java
@@ -94,7 +94,7 @@ private CodecService createCodecService() throws IOException {
         IndexAnalyzers indexAnalyzers = createTestAnalysis(settings, nodeSettings).indexAnalyzers;
         MapperRegistry mapperRegistry = new MapperRegistry(Collections.emptyMap(), Collections.emptyMap(), MapperPlugin.NOOP_FIELD_FILTER);
         MapperService service = new MapperService(settings, indexAnalyzers, xContentRegistry(), similarityService, mapperRegistry,
-                () -> null, () -> false);
+                () -> null, () -> false, null);
         return new CodecService(service, LogManager.getLogger("test"));
     }
 
diff --git a/server/src/test/java/org/elasticsearch/index/mapper/DateFieldMapperTests.java b/server/src/test/java/org/elasticsearch/index/mapper/DateFieldMapperTests.java
index 73a679ad863af..56fd1b7aa2709 100644
--- a/server/src/test/java/org/elasticsearch/index/mapper/DateFieldMapperTests.java
+++ b/server/src/test/java/org/elasticsearch/index/mapper/DateFieldMapperTests.java
@@ -21,19 +21,26 @@
 
 import org.apache.lucene.index.DocValuesType;
 import org.apache.lucene.index.IndexableField;
+import org.elasticsearch.Version;
 import org.elasticsearch.bootstrap.JavaVersion;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.compress.CompressedXContent;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.IndexService;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.mapper.MapperService.MergeReason;
 import org.elasticsearch.index.termvectors.TermVectorsService;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.InternalSettingsPlugin;
 import org.junit.Before;
+import org.mockito.Mockito;
 
 import java.io.IOException;
 import java.time.ZoneId;
@@ -45,7 +52,9 @@
 import java.util.Set;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 
 public class DateFieldMapperTests extends FieldMapperTestCase<DateFieldMapper.Builder> {
 
@@ -464,4 +473,26 @@ public void testMeta() throws Exception {
         assertEquals(mapping3, mapper.mappingSource().toString());
     }
 
+    public void testSingletonDataStreamTimestamp() throws Exception {
+        DateFieldMapper.Builder builder = new DateFieldMapper.Builder("@timestamp");
+        builder.setSingletonDataStreamTimestamp(true);
+        Mapper.BuilderContext builderContext = new Mapper.BuilderContext(Settings.EMPTY, new ContentPath(0));
+        DateFieldMapper dateFieldMapper = builder.build(builderContext);
+
+        IndexMetadata build = IndexMetadata.builder("")
+            .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
+            .numberOfShards(1).numberOfReplicas(0).build();
+        IndexSettings settings = new IndexSettings(build, Settings.EMPTY);
+        XContentParser mockParser = Mockito.mock(XContentParser.class);
+        Mockito.when(mockParser.textOrNull()).thenReturn("2020-12-12");
+        SourceToParse sourceToParse = new SourceToParse("_index", "_id", new BytesArray("{}"), XContentType.JSON);
+        ParseContext context = new ParseContext.InternalParseContext(settings, null, null, sourceToParse, mockParser);
+
+        assertThat(context.doc().getField("@timestamp"), nullValue());
+        dateFieldMapper.parse(context);
+        assertThat(context.doc().getField("@timestamp"), notNullValue());
+        MapperParsingException e = expectThrows(MapperParsingException.class, () -> dateFieldMapper.parse(context));
+        assertThat(e.getCause().getMessage(), equalTo("data stream timestamp field [@timestamp] encountered multiple values"));
+    }
+
 }
diff --git a/server/src/test/java/org/elasticsearch/index/mapper/DocumentParserTests.java b/server/src/test/java/org/elasticsearch/index/mapper/DocumentParserTests.java
index b467032c88d63..2a005187bcad4 100644
--- a/server/src/test/java/org/elasticsearch/index/mapper/DocumentParserTests.java
+++ b/server/src/test/java/org/elasticsearch/index/mapper/DocumentParserTests.java
@@ -38,6 +38,7 @@
 import org.elasticsearch.test.ESSingleNodeTestCase;
 import org.elasticsearch.test.TestGeoShapeFieldMapperPlugin;
 import org.elasticsearch.test.InternalSettingsPlugin;
+import org.mockito.Mockito;
 
 import java.io.IOException;
 import java.math.BigDecimal;
@@ -47,6 +48,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.test.StreamsUtils.copyToBytesFromClasspath;
@@ -56,6 +58,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 
 // TODO: make this a real unit test
 public class DocumentParserTests extends ESSingleNodeTestCase {
@@ -1683,4 +1686,45 @@ public void testTypeless() throws IOException {
         ParsedDocument doc = mapper.parse(new SourceToParse("test", "1", bytes, XContentType.JSON));
         assertNull(doc.dynamicMappingsUpdate()); // no update since we reused the existing type
     }
+
+    public void testDataStreamTimestampFieldRequired() {
+        RootObjectMapper rootObjectMapper = Mockito.mock(RootObjectMapper.class);
+        {
+            DateFieldMapper.Builder builder = new DateFieldMapper.Builder("@timestamp");
+            builder.setSingletonDataStreamTimestamp(true);
+            Mapper.BuilderContext builderContext = new Mapper.BuilderContext(Settings.EMPTY, new ContentPath(0));
+
+            DateFieldMapper dateFieldMapper = builder.build(builderContext);
+            Mockito.when(rootObjectMapper.isEnabled()).thenReturn(true);
+            Mockito.when(rootObjectMapper.nested()).thenReturn(ObjectMapper.Nested.NO);
+            Mockito.when(rootObjectMapper.getMapper(Mockito.eq("@timestamp"))).thenReturn(dateFieldMapper);
+        }
+
+        IndexMetadata build = IndexMetadata.builder("")
+            .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
+            .numberOfShards(1).numberOfReplicas(0).build();
+        IndexSettings settings = new IndexSettings(build, Settings.EMPTY);
+
+        Mapping mapping = new Mapping(Version.CURRENT, rootObjectMapper, new MetadataFieldMapper[0], Map.of());
+        DocumentMapper documentMapper = Mockito.mock(DocumentMapper.class);
+        Mockito.when(documentMapper.mapping()).thenReturn(mapping);
+
+        MapperService mapperService = Mockito.mock(MapperService.class);
+        Mockito.when(mapperService.isMetadataField(Mockito.anyString())).thenReturn(false);
+        DocumentMapperParser documentMapperParser = Mockito.mock(DocumentMapperParser.class);
+        Mockito.when(documentMapperParser.getMapperService()).thenReturn(mapperService);
+        DocumentParser documentParser = new DocumentParser(settings, documentMapperParser, documentMapper);
+
+        SourceToParse sourceToParse = new SourceToParse("_index", "_id", new BytesArray("{}"), XContentType.JSON);
+        ParsedDocument parsedDocument = documentParser.parseDocument(sourceToParse, new MetadataFieldMapper[0], null);
+        assertThat(parsedDocument.rootDoc().getField("@timestamp"), nullValue());
+
+        sourceToParse =  new SourceToParse("_index", "_id", new BytesArray("{\"@timestamp\": \"2020-12-12\"}"), XContentType.JSON);
+        parsedDocument = documentParser.parseDocument(sourceToParse, new MetadataFieldMapper[0], "@timestamp");
+        assertThat(parsedDocument.rootDoc().getField("@timestamp"), notNullValue());
+
+        Exception e = expectThrows(IllegalArgumentException.class, () -> documentParser.parseDocument(new SourceToParse("_index", "_id",
+            new BytesArray("{}"), XContentType.JSON), new MetadataFieldMapper[0], "@timestamp"));
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] is missing"));
+    }
 }
diff --git a/server/src/test/java/org/elasticsearch/index/mapper/ExternalFieldMapperTests.java b/server/src/test/java/org/elasticsearch/index/mapper/ExternalFieldMapperTests.java
index 80ca2d6096f26..d43c4af011779 100644
--- a/server/src/test/java/org/elasticsearch/index/mapper/ExternalFieldMapperTests.java
+++ b/server/src/test/java/org/elasticsearch/index/mapper/ExternalFieldMapperTests.java
@@ -73,7 +73,7 @@ public void testExternalValues() throws Exception {
             return indexService.newQueryShardContext(0, null, () -> { throw new UnsupportedOperationException(); }, null);
         };
         DocumentMapperParser parser = new DocumentMapperParser(indexService.getIndexSettings(), indexService.mapperService(),
-                indexService.xContentRegistry(), indexService.similarityService(), mapperRegistry, queryShardContext);
+                indexService.xContentRegistry(), indexService.similarityService(), mapperRegistry, queryShardContext, null);
         DocumentMapper documentMapper = parser.parse("type", new CompressedXContent(
                 Strings.toString(XContentFactory.jsonBuilder().startObject().startObject("type")
                 .startObject(ExternalMetadataMapper.CONTENT_TYPE)
@@ -120,7 +120,7 @@ public void testExternalValuesWithMultifield() throws Exception {
             return indexService.newQueryShardContext(0, null, () -> { throw new UnsupportedOperationException(); }, null);
         };
         DocumentMapperParser parser = new DocumentMapperParser(indexService.getIndexSettings(), indexService.mapperService(),
-                indexService.xContentRegistry(), indexService.similarityService(), mapperRegistry, queryShardContext);
+                indexService.xContentRegistry(), indexService.similarityService(), mapperRegistry, queryShardContext, null);
 
         DocumentMapper documentMapper = parser.parse("type", new CompressedXContent(
                 Strings
diff --git a/server/src/test/java/org/elasticsearch/index/mapper/TypeParsersTests.java b/server/src/test/java/org/elasticsearch/index/mapper/TypeParsersTests.java
index b4074833ce9dc..04bb7ac618d39 100644
--- a/server/src/test/java/org/elasticsearch/index/mapper/TypeParsersTests.java
+++ b/server/src/test/java/org/elasticsearch/index/mapper/TypeParsersTests.java
@@ -192,7 +192,7 @@ public void testMultiFieldWithinMultiField() throws IOException {
 
         Version olderVersion = VersionUtils.randomPreviousCompatibleVersion(random(), Version.V_8_0_0);
         Mapper.TypeParser.ParserContext olderContext = new Mapper.TypeParser.ParserContext(
-            null, null, type -> typeParser, olderVersion, null);
+            null, null, type -> typeParser, olderVersion, null, null);
 
         TypeParsers.parseField(builder, "some-field", fieldNode, olderContext);
         assertWarnings("At least one multi-field, [sub-field], " +
@@ -207,7 +207,7 @@ public void testMultiFieldWithinMultiField() throws IOException {
 
         Version version = VersionUtils.randomVersionBetween(random(), Version.V_8_0_0, Version.CURRENT);
         Mapper.TypeParser.ParserContext context = new Mapper.TypeParser.ParserContext(
-            null, null, type -> typeParser, version, null);
+            null, null, type -> typeParser, version, null, null);
 
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
             () -> TypeParsers.parseField(builder, "some-field", fieldNodeCopy, context));
@@ -233,7 +233,7 @@ public TokenStream create(TokenStream tokenStream) {
 
     public void testParseMeta() {
         FieldMapper.Builder<?> builder = new KeywordFieldMapper.Builder("foo");
-        Mapper.TypeParser.ParserContext parserContext = new Mapper.TypeParser.ParserContext(null, null, null, null, null);
+        Mapper.TypeParser.ParserContext parserContext = new Mapper.TypeParser.ParserContext(null, null, null, null, null, null);
 
         {
             Map<String, Object> mapping = new HashMap<>(Map.of("meta", 3));
diff --git a/server/src/test/java/org/elasticsearch/indices/IndicesLifecycleListenerSingleNodeTests.java b/server/src/test/java/org/elasticsearch/indices/IndicesLifecycleListenerSingleNodeTests.java
index f9ce842584e5f..aba5cd765d52e 100644
--- a/server/src/test/java/org/elasticsearch/indices/IndicesLifecycleListenerSingleNodeTests.java
+++ b/server/src/test/java/org/elasticsearch/indices/IndicesLifecycleListenerSingleNodeTests.java
@@ -122,7 +122,7 @@ public void afterIndexRemoved(Index index, IndexSettings indexSettings, IndexRem
         };
         indicesService.removeIndex(idx, DELETED, "simon says");
         try {
-            IndexService index = indicesService.createIndex(metadata, Arrays.asList(countingListener), false);
+            IndexService index = indicesService.createIndex(metadata, Arrays.asList(countingListener), false, null);
             assertEquals(3, counter.get());
             idx = index.index();
             ShardRouting newRouting = shardRouting;
diff --git a/server/src/test/java/org/elasticsearch/indices/IndicesServiceTests.java b/server/src/test/java/org/elasticsearch/indices/IndicesServiceTests.java
index d814b033c3275..a95e85dc19a8a 100644
--- a/server/src/test/java/org/elasticsearch/indices/IndicesServiceTests.java
+++ b/server/src/test/java/org/elasticsearch/indices/IndicesServiceTests.java
@@ -549,7 +549,7 @@ public void testGetEngineFactory() throws IOException {
                     .numberOfShards(1)
                     .numberOfReplicas(0)
                     .build();
-            final IndexService indexService = indicesService.createIndex(indexMetadata, Collections.emptyList(), false);
+            final IndexService indexService = indicesService.createIndex(indexMetadata, Collections.emptyList(), false, null);
             if (value != null && value) {
                 assertThat(indexService.getEngineFactory(), instanceOf(FooEnginePlugin.FooEngineFactory.class));
             } else {
@@ -574,8 +574,8 @@ public void testConflictingEngineFactories() {
                 .build();
 
         final IndicesService indicesService = getIndicesService();
-        final IllegalStateException e =
-                expectThrows(IllegalStateException.class, () -> indicesService.createIndex(indexMetadata, Collections.emptyList(), false));
+        final IllegalStateException e = expectThrows(IllegalStateException.class,
+            () -> indicesService.createIndex(indexMetadata, Collections.emptyList(), false, null));
         final String pattern =
                 ".*multiple engine factories provided for \\[foobar/.*\\]: \\[.*FooEngineFactory\\],\\[.*BarEngineFactory\\].*";
         assertThat(e, hasToString(new RegexMatcher(pattern)));
diff --git a/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java b/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java
index 6b0df8c422445..582c544d345c2 100644
--- a/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java
+++ b/server/src/test/java/org/elasticsearch/indices/cluster/AbstractIndicesClusterStateServiceTestCase.java
@@ -196,7 +196,7 @@ protected class MockIndicesService implements AllocatedIndices<MockIndexShard, M
         public synchronized MockIndexService createIndex(
                 IndexMetadata indexMetadata,
                 List<IndexEventListener> buildInIndexListener,
-                boolean writeDanglingIndices) throws IOException {
+                boolean writeDanglingIndices, String dataStreamTimestampField) throws IOException {
             MockIndexService indexService = new MockIndexService(new IndexSettings(indexMetadata, Settings.EMPTY));
             indices = Maps.copyMapWithAddedEntry(indices, indexMetadata.getIndexUUID(), indexService);
             return indexService;
diff --git a/test/framework/src/main/java/org/elasticsearch/index/MapperTestUtils.java b/test/framework/src/main/java/org/elasticsearch/index/MapperTestUtils.java
index 04a066e3361ad..2f3fb419a893a 100644
--- a/test/framework/src/main/java/org/elasticsearch/index/MapperTestUtils.java
+++ b/test/framework/src/main/java/org/elasticsearch/index/MapperTestUtils.java
@@ -66,6 +66,8 @@ public static MapperService newMapperService(NamedXContentRegistry xContentRegis
             xContentRegistry,
             similarityService,
             mapperRegistry,
-            () -> null, () -> false);
+            () -> null,
+            () -> false,
+            null);
     }
 }
diff --git a/test/framework/src/main/java/org/elasticsearch/index/engine/TranslogHandler.java b/test/framework/src/main/java/org/elasticsearch/index/engine/TranslogHandler.java
index 5adc5f86699f9..884a802cc77eb 100644
--- a/test/framework/src/main/java/org/elasticsearch/index/engine/TranslogHandler.java
+++ b/test/framework/src/main/java/org/elasticsearch/index/engine/TranslogHandler.java
@@ -65,7 +65,7 @@ public TranslogHandler(NamedXContentRegistry xContentRegistry, IndexSettings ind
         SimilarityService similarityService = new SimilarityService(indexSettings, null, emptyMap());
         MapperRegistry mapperRegistry = new IndicesModule(emptyList()).getMapperRegistry();
         mapperService = new MapperService(indexSettings, indexAnalyzers, xContentRegistry, similarityService, mapperRegistry,
-                () -> null, () -> false);
+                () -> null, () -> false, null);
     }
 
     private DocumentMapperForType docMapper(String type) {
diff --git a/test/framework/src/main/java/org/elasticsearch/search/aggregations/AggregatorTestCase.java b/test/framework/src/main/java/org/elasticsearch/search/aggregations/AggregatorTestCase.java
index 65d7b5e89697b..26ca4ddbfad46 100644
--- a/test/framework/src/main/java/org/elasticsearch/search/aggregations/AggregatorTestCase.java
+++ b/test/framework/src/main/java/org/elasticsearch/search/aggregations/AggregatorTestCase.java
@@ -855,7 +855,7 @@ private void writeTestDoc(MappedFieldType fieldType, String fieldName, RandomInd
 
     private class MockParserContext extends Mapper.TypeParser.ParserContext {
         MockParserContext() {
-            super(null, null, null, null, null);
+            super(null, null, null, null, null, null);
         }
 
         @Override
diff --git a/test/framework/src/main/java/org/elasticsearch/test/AbstractBuilderTestCase.java b/test/framework/src/main/java/org/elasticsearch/test/AbstractBuilderTestCase.java
index aa41bcad47c32..a5efdae46d58f 100644
--- a/test/framework/src/main/java/org/elasticsearch/test/AbstractBuilderTestCase.java
+++ b/test/framework/src/main/java/org/elasticsearch/test/AbstractBuilderTestCase.java
@@ -359,7 +359,7 @@ private static class ServiceHolder implements Closeable {
             similarityService = new SimilarityService(idxSettings, null, Collections.emptyMap());
             MapperRegistry mapperRegistry = indicesModule.getMapperRegistry();
             mapperService = new MapperService(idxSettings, indexAnalyzers, xContentRegistry, similarityService, mapperRegistry,
-                    () -> createShardContext(null), () -> false);
+                    () -> createShardContext(null), () -> false, null);
             IndicesFieldDataCache indicesFieldDataCache = new IndicesFieldDataCache(nodeSettings, new IndexFieldDataCache.Listener() {
             });
             indexFieldDataService = new IndexFieldDataService(idxSettings, indicesFieldDataCache,
diff --git a/x-pack/plugin/ilm/qa/multi-node/src/test/java/org/elasticsearch/xpack/TimeSeriesRestDriver.java b/x-pack/plugin/ilm/qa/multi-node/src/test/java/org/elasticsearch/xpack/TimeSeriesRestDriver.java
index a8512250d3fb7..9f633d6e5c06d 100644
--- a/x-pack/plugin/ilm/qa/multi-node/src/test/java/org/elasticsearch/xpack/TimeSeriesRestDriver.java
+++ b/x-pack/plugin/ilm/qa/multi-node/src/test/java/org/elasticsearch/xpack/TimeSeriesRestDriver.java
@@ -95,7 +95,7 @@ public static void indexDocument(RestClient client, String indexAbstractionName)
 
     public static void indexDocument(RestClient client, String indexAbstractionName, boolean refresh) throws IOException {
         Request indexRequest = new Request("POST", indexAbstractionName + "/_doc" + (refresh ? "?refresh" : ""));
-        indexRequest.setEntity(new StringEntity("{\"a\": \"test\"}", ContentType.APPLICATION_JSON));
+        indexRequest.setEntity(new StringEntity("{\"@timestamp\": \"2020-12-12\"}", ContentType.APPLICATION_JSON));
         Response response = client.performRequest(indexRequest);
         logger.info(response.getStatusLine());
     }
diff --git a/x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/ClassificationIT.java b/x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/ClassificationIT.java
index 1cfa7de7e5907..6ee56b6d5e473 100644
--- a/x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/ClassificationIT.java
+++ b/x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/ClassificationIT.java
@@ -717,6 +717,7 @@ private static void indexData(String sourceIndex, int numTrainingRows, int numNo
             .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
         for (int i = 0; i < numTrainingRows; i++) {
             List<Object> source = List.of(
+                "time", "2020-12-12",
                 BOOLEAN_FIELD, BOOLEAN_FIELD_VALUES.get(i % BOOLEAN_FIELD_VALUES.size()),
                 NUMERICAL_FIELD, NUMERICAL_FIELD_VALUES.get(i % NUMERICAL_FIELD_VALUES.size()),
                 DISCRETE_NUMERICAL_FIELD, DISCRETE_NUMERICAL_FIELD_VALUES.get(i % DISCRETE_NUMERICAL_FIELD_VALUES.size()),
@@ -747,6 +748,7 @@ private static void indexData(String sourceIndex, int numTrainingRows, int numNo
             if (NESTED_FIELD.equals(dependentVariable) == false) {
                 source.addAll(List.of(NESTED_FIELD, KEYWORD_FIELD_VALUES.get(i % KEYWORD_FIELD_VALUES.size())));
             }
+            source.addAll(List.of("time", "2020-12-12"));
             IndexRequest indexRequest = new IndexRequest(sourceIndex).source(source.toArray()).opType(DocWriteRequest.OpType.CREATE);
             bulkRequestBuilder.add(indexRequest);
         }
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_data_stream_resolvability.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_data_stream_resolvability.yml
index 2d76cc0b779cc..1a3381055c34f 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_data_stream_resolvability.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_data_stream_resolvability.yml
@@ -102,7 +102,9 @@
       index:
         index:  logs-foobar
         refresh: true
-        body:   { foo: bar }
+        body:
+          foo: bar
+          '@timestamp': '2020-12-12'
 
   - do:
       ilm.explain_lifecycle:
diff --git a/x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherPluginTests.java b/x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherPluginTests.java
index 4eaeb2a8ea65e..c1139fa80869e 100644
--- a/x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherPluginTests.java
+++ b/x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherPluginTests.java
@@ -76,7 +76,7 @@ public void testWatcherDisabledTests() throws Exception {
         AnalysisRegistry registry = new AnalysisRegistry(TestEnvironment.newEnvironment(settings), emptyMap(), emptyMap(), emptyMap(),
                 emptyMap(), emptyMap(), emptyMap(), emptyMap(), emptyMap(), emptyMap());
         IndexModule indexModule = new IndexModule(indexSettings, registry, new InternalEngineFactory(), Collections.emptyMap(),
-                () -> true, new IndexNameExpressionResolver());
+                () -> true, new IndexNameExpressionResolver(), null);
         // this will trip an assertion if the watcher indexing operation listener is null (which it is) but we try to add it
         watcher.onIndexModule(indexModule);