Replaced _data_stream_timestamp meta field's 'path' option with 'enabled' option (#59503)

and adjusted exception messages.

Relates to #59076

server/src/main/java/org/elasticsearch/cluster/metadata/ComposableIndexTemplate.java

@@ -267,8 +267,9 @@ public class ComposableIndexTemplate extends AbstractDiffable<ComposableIndexTem
         /**
          * @return a mapping snippet for a backing index with `_data_stream_timestamp` meta field mapper properly configured.
          */
-        public Map<String, Object> getDataSteamMappingSnippet() {
-            return Map.of(MapperService.SINGLE_MAPPING_NAME, Map.of("_data_stream_timestamp", Map.of("path", FIXED_TIMESTAMP_FIELD)));
+        public Map<String, Object> getDataStreamMappingSnippet() {
+            // _data_stream_timestamp meta fields default to @timestamp:
+            return Map.of(MapperService.SINGLE_MAPPING_NAME, Map.of("_data_stream_timestamp", Map.of("enabled", true)));
         }
 
         @Override

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java

@@ -179,10 +179,11 @@ public class MetadataCreateDataStreamService {
 
         Map<String, Object> parsedTemplateMapping =
             MapperService.parseMapping(NamedXContentRegistry.EMPTY, mapperService.documentMapper().mappingSource().string());
-        String configuredPath = ObjectPath.eval("_doc._data_stream_timestamp.path", parsedTemplateMapping);
-        if (timestampFieldName.equals(configuredPath) == false) {
-            throw new IllegalArgumentException("[_data_stream_timestamp] meta field doesn't point to data stream timestamp field [" +
-                timestampFieldName + "]");
+        Boolean enabled = ObjectPath.eval("_doc._data_stream_timestamp.enabled", parsedTemplateMapping);
+        // Sanity check: if this fails then somehow the mapping for _data_stream_timestamp has been overwritten and
+        // that would be a bug.
+        if (enabled == null || enabled == false) {
+            throw new IllegalStateException("[_data_stream_timestamp] meta field has been disabled");
         }
 
         // Sanity check (this validation logic should already have been executed when merging mappings):

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateService.java

@@ -981,7 +981,7 @@ public class MetadataIndexTemplateService {
             // Only if template has data stream definition this should be added and
             // adding this template last, since _timestamp field should have highest precedence:
             Optional.ofNullable(template.getDataStreamTemplate())
-                .map(ComposableIndexTemplate.DataStreamTemplate::getDataSteamMappingSnippet)
+                .map(ComposableIndexTemplate.DataStreamTemplate::getDataStreamMappingSnippet)
                 .map(mapping -> {
                     try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
                         builder.value(mapping);

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataIndexTemplateServiceTests.java

@@ -1500,11 +1500,11 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                 public MetadataFieldMapper.Builder<?> parse(String name,
                                                             Map<String, Object> node,
                                                             ParserContext parserContext) throws MapperParsingException {
-                    String path = (String) node.remove("path");
+                    Boolean enabled = (Boolean) node.remove("enabled");
                     return new MetadataFieldMapper.Builder(name, new FieldType()) {
                         @Override
                         public MetadataFieldMapper build(Mapper.BuilderContext context) {
-                            return newInstance(path);
+                            return newInstance(enabled);
                         }
                     };
                 }
@@ -1514,7 +1514,7 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
                     return newInstance(null);
                 }
 
-                MetadataFieldMapper newInstance(String path) {
+                MetadataFieldMapper newInstance(Boolean enabled) {
                     FieldType fieldType = new FieldType();
                     fieldType.freeze();
                     MappedFieldType mappedFieldType =
@@ -1547,12 +1547,12 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
 
                         @Override
                         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-                            if (path == null) {
+                            if (enabled == null) {
                                 return builder;
                             }
 
                             builder.startObject(simpleName());
-                            builder.field("path", path);
+                            builder.field("enabled", enabled);
                             return builder.endObject();
                         }
 

test/framework/src/main/java/org/elasticsearch/cluster/DataStreamTestHelper.java

@@ -79,7 +79,7 @@ public final class DataStreamTestHelper {
     public static String generateMapping(String timestampFieldName, String type) {
         return "{\n" +
             "      \"_data_stream_timestamp\": {\n" +
-            "        \"path\": \"" + timestampFieldName + "\"\n" +
+            "        \"enabled\": true\n" +
             "      }," +
             "      \"properties\": {\n" +
             "        \"" + timestampFieldName + "\": {\n" +

x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java

@@ -372,7 +372,7 @@ public class DataStreamIT extends ESIntegTestCase {
         );
         assertThat(
             e.getCause().getCause().getMessage(),
-            equalTo("the configured timestamp field [@timestamp] is of type [keyword], but [date,date_nanos] is expected")
+            equalTo("data stream timestamp field [@timestamp] is of type [keyword], but [date,date_nanos] is expected")
         );
     }
 
@@ -605,7 +605,7 @@ public class DataStreamIT extends ESIntegTestCase {
             "properties",
             Map.of("@timestamp", Map.of("type", "date")),
             "_data_stream_timestamp",
-            Map.of("path", "@timestamp")
+            Map.of("enabled", true)
         );
         GetMappingsResponse getMappingsResponse = client().admin().indices().prepareGetMappings("logs-foobar").get();
         assertThat(getMappingsResponse.getMappings().size(), equalTo(2));
@@ -616,7 +616,7 @@ public class DataStreamIT extends ESIntegTestCase {
             "properties",
             Map.of("@timestamp", Map.of("type", "date"), "my_field", Map.of("type", "keyword")),
             "_data_stream_timestamp",
-            Map.of("path", "@timestamp")
+            Map.of("enabled", true)
         );
         client().admin()
             .indices()

x-pack/plugin/data-streams/src/main/java/org/elasticsearch/xpack/datastreams/mapper/DataStreamTimestampFieldMapper.java

@@ -15,6 +15,7 @@ import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.index.mapper.DocumentFieldMappers;
 import org.elasticsearch.index.mapper.FieldMapper;
@@ -39,6 +40,7 @@ import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
 
     public static final String NAME = "_data_stream_timestamp";
+    private static final String DEFAULT_PATH = "@timestamp";
 
     public static class Defaults {
 
@@ -77,19 +79,19 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
 
     public static class Builder extends MetadataFieldMapper.Builder<Builder> {
 
-        private String path;
+        private boolean enabled;
 
         public Builder() {
             super(NAME, Defaults.TIMESTAMP_FIELD_TYPE);
         }
 
-        public void setPath(String path) {
-            this.path = path;
+        public void setEnabled(boolean enabled) {
+            this.enabled = enabled;
         }
 
         @Override
         public MetadataFieldMapper build(BuilderContext context) {
-            return new DataStreamTimestampFieldMapper(fieldType, new TimestampFieldType(), path);
+            return new DataStreamTimestampFieldMapper(fieldType, new TimestampFieldType(), enabled);
         }
     }
 
@@ -103,8 +105,8 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
                 Map.Entry<String, Object> entry = iterator.next();
                 String fieldName = entry.getKey();
                 Object fieldNode = entry.getValue();
-                if (fieldName.equals("path")) {
-                    builder.setPath((String) fieldNode);
+                if (fieldName.equals("enabled")) {
+                    builder.setEnabled(XContentMapValues.nodeBooleanValue(fieldNode, name + ".enabled"));
                     iterator.remove();
                 }
             }
@@ -113,32 +115,34 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
 
         @Override
         public MetadataFieldMapper getDefault(ParserContext parserContext) {
-            return new DataStreamTimestampFieldMapper(Defaults.TIMESTAMP_FIELD_TYPE, new TimestampFieldType(), null);
+            return new DataStreamTimestampFieldMapper(Defaults.TIMESTAMP_FIELD_TYPE, new TimestampFieldType(), false);
         }
     }
 
     private final String path;
+    private final boolean enabled;
 
-    private DataStreamTimestampFieldMapper(FieldType fieldType, MappedFieldType mappedFieldType, String path) {
+    private DataStreamTimestampFieldMapper(FieldType fieldType, MappedFieldType mappedFieldType, boolean enabled) {
         super(fieldType, mappedFieldType);
-        this.path = path;
+        this.path = DEFAULT_PATH;
+        this.enabled = enabled;
     }
 
     public void validate(DocumentFieldMappers lookup) {
-        if (path == null) {
+        if (enabled == false) {
             // not configured, so skip the validation
             return;
         }
 
         Mapper mapper = lookup.getMapper(path);
         if (mapper == null) {
-            throw new IllegalArgumentException("the configured timestamp field [" + path + "] does not exist");
+            throw new IllegalArgumentException("data stream timestamp field [" + path + "] does not exist");
         }
 
         if (DateFieldMapper.CONTENT_TYPE.equals(mapper.typeName()) == false
             && DateFieldMapper.DATE_NANOS_CONTENT_TYPE.equals(mapper.typeName()) == false) {
             throw new IllegalArgumentException(
-                "the configured timestamp field ["
+                "data stream timestamp field ["
                     + path
                     + "] is of type ["
                     + mapper.typeName()
@@ -152,19 +156,19 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
 
         DateFieldMapper dateFieldMapper = (DateFieldMapper) mapper;
         if (dateFieldMapper.fieldType().isSearchable() == false) {
-            throw new IllegalArgumentException("the configured timestamp field [" + path + "] is not indexed");
+            throw new IllegalArgumentException("data stream timestamp field [" + path + "] is not indexed");
         }
         if (dateFieldMapper.fieldType().hasDocValues() == false) {
-            throw new IllegalArgumentException("the configured timestamp field [" + path + "] doesn't have doc values");
+            throw new IllegalArgumentException("data stream timestamp field [" + path + "] doesn't have doc values");
         }
         if (dateFieldMapper.getNullValue() != null) {
             throw new IllegalArgumentException(
-                "the configured timestamp field [" + path + "] has disallowed [null_value] attribute specified"
+                "data stream timestamp field [" + path + "] has disallowed [null_value] attribute specified"
             );
         }
         if (dateFieldMapper.getIgnoreMalformed()) {
             throw new IllegalArgumentException(
-                "the configured timestamp field [" + path + "] has disallowed [ignore_malformed] attribute specified"
+                "data stream timestamp field [" + path + "] has disallowed [ignore_malformed] attribute specified"
             );
         }
 
@@ -184,7 +188,7 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
             // All other configured attributes are not allowed:
             if (configuredSettings.isEmpty() == false) {
                 throw new IllegalArgumentException(
-                    "the configured timestamp field [@timestamp] has disallowed attributes: " + configuredSettings.keySet()
+                    "data stream timestamp field [@timestamp] has disallowed attributes: " + configuredSettings.keySet()
                 );
             }
         } catch (IOException e) {
@@ -192,10 +196,6 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
         }
     }
 
-    public String getPath() {
-        return path;
-    }
-
     @Override
     public void preParse(ParseContext context) throws IOException {}
 
@@ -207,7 +207,7 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
 
     @Override
     public void postParse(ParseContext context) throws IOException {
-        if (path == null) {
+        if (enabled == false) {
             // not configured, so skip the validation
             return;
         }
@@ -227,12 +227,12 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        if (path == null) {
+        if (enabled == false) {
             return builder;
         }
 
         builder.startObject(simpleName());
-        builder.field("path", path);
+        builder.field("enabled", enabled);
         return builder.endObject();
     }
 
@@ -254,8 +254,8 @@ public class DataStreamTimestampFieldMapper extends MetadataFieldMapper {
     @Override
     protected void mergeOptions(FieldMapper other, List<String> conflicts) {
         DataStreamTimestampFieldMapper otherTimestampFieldMapper = (DataStreamTimestampFieldMapper) other;
-        if (Objects.equals(path, otherTimestampFieldMapper.path) == false) {
-            conflicts.add("cannot update path setting for [_data_stream_timestamp]");
+        if (Objects.equals(enabled, otherTimestampFieldMapper.enabled) == false) {
+            conflicts.add("cannot update enabled setting for [_data_stream_timestamp]");
         }
     }
 }

x-pack/plugin/data-streams/src/test/java/org/elasticsearch/xpack/datastreams/mapper/DataStreamTimestampFieldMapperTests.java

@@ -40,7 +40,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "@timestamp")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
                 .startObject("@timestamp")
@@ -98,10 +98,10 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "non-existing-field")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
-                .startObject("@timestamp")
+                .startObject("my_date_field")
                 .field("type", "date")
                 .endObject()
                 .endObject()
@@ -114,7 +114,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
             () -> createIndex("test").mapperService()
                 .merge("type", new CompressedXContent(mapping), MapperService.MergeReason.MAPPING_UPDATE)
         );
-        assertThat(e.getMessage(), equalTo("the configured timestamp field [non-existing-field] does not exist"));
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] does not exist"));
     }
 
     public void testValidateInvalidFieldType() throws IOException {
@@ -123,7 +123,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "@timestamp")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
                 .startObject("@timestamp")
@@ -141,7 +141,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
         );
         assertThat(
             e.getMessage(),
-            equalTo("the configured timestamp field [@timestamp] is of type [keyword], but [date,date_nanos] is expected")
+            equalTo("data stream timestamp field [@timestamp] is of type [keyword], but [date,date_nanos] is expected")
         );
     }
 
@@ -151,7 +151,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "@timestamp")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
                 .startObject("@timestamp")
@@ -168,7 +168,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
             () -> createIndex("test").mapperService()
                 .merge("type", new CompressedXContent(mapping), MapperService.MergeReason.MAPPING_UPDATE)
         );
-        assertThat(e.getMessage(), equalTo("the configured timestamp field [@timestamp] is not indexed"));
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] is not indexed"));
     }
 
     public void testValidateNotDocValues() throws IOException {
@@ -177,7 +177,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "@timestamp")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
                 .startObject("@timestamp")
@@ -194,7 +194,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
             () -> createIndex("test").mapperService()
                 .merge("type", new CompressedXContent(mapping), MapperService.MergeReason.MAPPING_UPDATE)
         );
-        assertThat(e.getMessage(), equalTo("the configured timestamp field [@timestamp] doesn't have doc values"));
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] doesn't have doc values"));
     }
 
     public void testValidateNullValue() throws IOException {
@@ -203,7 +203,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "@timestamp")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
                 .startObject("@timestamp")
@@ -220,7 +220,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
             () -> createIndex("test").mapperService()
                 .merge("type", new CompressedXContent(mapping), MapperService.MergeReason.MAPPING_UPDATE)
         );
-        assertThat(e.getMessage(), equalTo("the configured timestamp field [@timestamp] has disallowed [null_value] attribute specified"));
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] has disallowed [null_value] attribute specified"));
     }
 
     public void testValidateIgnoreMalformed() throws IOException {
@@ -229,7 +229,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "@timestamp")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
                 .startObject("@timestamp")
@@ -248,7 +248,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
         );
         assertThat(
             e.getMessage(),
-            equalTo("the configured timestamp field [@timestamp] has disallowed [ignore_malformed] attribute specified")
+            equalTo("data stream timestamp field [@timestamp] has disallowed [ignore_malformed] attribute specified")
         );
     }
 
@@ -258,7 +258,7 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
                 .startObject()
                 .startObject("type")
                 .startObject("_data_stream_timestamp")
-                .field("path", "@timestamp")
+                .field("enabled", true)
                 .endObject()
                 .startObject("properties")
                 .startObject("@timestamp")
@@ -275,63 +275,21 @@ public class DataStreamTimestampFieldMapperTests extends ESSingleNodeTestCase {
             () -> createIndex("test").mapperService()
                 .merge("type", new CompressedXContent(mapping), MapperService.MergeReason.MAPPING_UPDATE)
         );
-        assertThat(e.getMessage(), equalTo("the configured timestamp field [@timestamp] has disallowed attributes: [store]"));
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] has disallowed attributes: [store]"));
     }
 
     public void testCannotUpdateTimestampField() throws IOException {
         DocumentMapperParser parser = createIndex("test").mapperService().documentMapperParser();
         String mapping1 =
-            "{\"type\":{\"_data_stream_timestamp\":{\"path\":\"@timestamp\"}, \"properties\": {\"@timestamp\": {\"type\": \"date\"}}}}}";
-        String mapping2 = "{\"type\":{\"_data_stream_timestamp\":{\"path\":\"@timestamp2\"}, \"properties\": {\"@timestamp2\": "
+            "{\"type\":{\"_data_stream_timestamp\":{\"enabled\":false}, \"properties\": {\"@timestamp\": {\"type\": \"date\"}}}}}";
+        String mapping2 = "{\"type\":{\"_data_stream_timestamp\":{\"enabled\":true}, \"properties\": {\"@timestamp2\": "
             + "{\"type\": \"date\"},\"@timestamp\": {\"type\": \"date\"}}}})";
-        assertConflicts(mapping1, mapping2, parser, "cannot update path setting for [_data_stream_timestamp]");
+        assertConflicts(mapping1, mapping2, parser, "cannot update enabled setting for [_data_stream_timestamp]");
 
         mapping1 = "{\"type\":{\"properties\":{\"@timestamp\": {\"type\": \"date\"}}}}}";
-        mapping2 = "{\"type\":{\"_data_stream_timestamp\":{\"path\":\"@timestamp2\"}, \"properties\": "
+        mapping2 = "{\"type\":{\"_data_stream_timestamp\":{\"enabled\":true}, \"properties\": "
             + "{\"@timestamp2\": {\"type\": \"date\"},\"@timestamp\": {\"type\": \"date\"}}}})";
-        assertConflicts(mapping1, mapping2, parser, "cannot update path setting for [_data_stream_timestamp]");
-    }
-
-    public void testDifferentTSField() throws IOException {
-        String mapping = "{\n"
-            + "      \"_data_stream_timestamp\": {\n"
-            + "        \"path\": \"event.my_timestamp\"\n"
-            + "      },\n"
-            + "      \"properties\": {\n"
-            + "        \"event\": {\n"
-            + "          \"properties\": {\n"
-            + "            \"my_timestamp\": {\n"
-            + "              \"type\": \"date\""
-            + "            }\n"
-            + "          }\n"
-            + "        }\n"
-            + "      }\n"
-            + "    }";
-        DocumentMapper docMapper = createIndex("test").mapperService()
-            .merge("type", new CompressedXContent(mapping), MapperService.MergeReason.MAPPING_UPDATE);
-
-        ParsedDocument doc = docMapper.parse(
-            new SourceToParse(
-                "test",
-                "1",
-                BytesReference.bytes(XContentFactory.jsonBuilder().startObject().field("event.my_timestamp", "2020-12-12").endObject()),
-                XContentType.JSON
-            )
-        );
-        assertThat(doc.rootDoc().getFields("event.my_timestamp").length, equalTo(2));
-
-        Exception e = expectThrows(
-            MapperException.class,
-            () -> docMapper.parse(
-                new SourceToParse(
-                    "test",
-                    "1",
-                    BytesReference.bytes(XContentFactory.jsonBuilder().startObject().field("event.timestamp", "2020-12-12").endObject()),
-                    XContentType.JSON
-                )
-            )
-        );
-        assertThat(e.getCause().getMessage(), equalTo("data stream timestamp field [event.my_timestamp] is missing"));
+        assertConflicts(mapping1, mapping2, parser, "cannot update enabled setting for [_data_stream_timestamp]");
     }
 
 }

x-pack/plugin/data-streams/src/test/java/org/elasticsearch/xpack/datastreams/mapper/MetadataCreateDataStreamServiceTests.java

@@ -33,20 +33,26 @@ public class MetadataCreateDataStreamServiceTests extends ESTestCase {
 
     public void testValidateTimestampFieldMappingNoFieldMapping() {
         Exception e = expectThrows(
-            IllegalArgumentException.class,
+            IllegalStateException.class,
             () -> validateTimestampFieldMapping("@timestamp", createMapperService("{}"))
         );
-        assertThat(
-            e.getMessage(),
-            equalTo("[_data_stream_timestamp] meta field doesn't point to data stream timestamp field [@timestamp]")
-        );
+        assertThat(e.getMessage(), equalTo("[_data_stream_timestamp] meta field has been disabled"));
+        String mapping1 = "{\n"
+            + "      \"_data_stream_timestamp\": {\n"
+            + "        \"enabled\": false\n"
+            + "      },"
+            + "      \"properties\": {\n"
+            + "        \"@timestamp\": {\n"
+            + "          \"type\": \"date\"\n"
+            + "        }\n"
+            + "      }\n"
+            + "    }";
+        e = expectThrows(IllegalStateException.class, () -> validateTimestampFieldMapping("@timestamp", createMapperService(mapping1)));
+        assertThat(e.getMessage(), equalTo("[_data_stream_timestamp] meta field has been disabled"));
 
-        String mapping = generateMapping("@timestamp2", "date");
-        e = expectThrows(IllegalArgumentException.class, () -> validateTimestampFieldMapping("@timestamp", createMapperService(mapping)));
-        assertThat(
-            e.getMessage(),
-            equalTo("[_data_stream_timestamp] meta field doesn't point to data stream timestamp field [@timestamp]")
-        );
+        String mapping2 = generateMapping("@timestamp2", "date");
+        e = expectThrows(IllegalArgumentException.class, () -> validateTimestampFieldMapping("@timestamp", createMapperService(mapping2)));
+        assertThat(e.getMessage(), equalTo("data stream timestamp field [@timestamp] does not exist"));
     }
 
     public void testValidateTimestampFieldMappingInvalidFieldType() {
@@ -57,7 +63,7 @@ public class MetadataCreateDataStreamServiceTests extends ESTestCase {
         );
         assertThat(
             e.getMessage(),
-            equalTo("the configured timestamp field [@timestamp] is of type [keyword], " + "but [date,date_nanos] is expected")
+            equalTo("data stream timestamp field [@timestamp] is of type [keyword], " + "but [date,date_nanos] is expected")
         );
     }
 

x-pack/qa/full-cluster-restart/src/test/java/org/elasticsearch/xpack/restart/FullClusterRestartIT.java

@@ -636,6 +636,7 @@ public class FullClusterRestartIT extends AbstractFullClusterRestartTestCase {
 
     @SuppressWarnings("unchecked")
     public void testDataStreams() throws Exception {
+        assumeTrue("fail until #59503 is backported to 7.x and 7.9", false);
         assumeTrue("no data streams in versions before " + Version.V_7_9_0, getOldClusterVersion().onOrAfter(Version.V_7_9_0));
         if (isRunningAgainstOldCluster()) {
             createComposableTemplate(client(), "dst", "ds");

x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/DataStreamsUpgradeIT.java

@@ -18,6 +18,7 @@ import static org.elasticsearch.upgrades.IndexingIT.assertCount;
 public class DataStreamsUpgradeIT extends AbstractUpgradeTestCase {
 
     public void testDataStreams() throws IOException {
+        assumeTrue("fail until #59503 is backported to 7.x and 7.9", false);
         if (CLUSTER_TYPE == ClusterType.OLD) {
             String requestBody = "{\n" +
                 "      \"index_patterns\":[\"logs-*\"],\n" +