diff --git a/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/LogsDataStreamIT.java b/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/LogsDataStreamIT.java
index 31d2f6a8e2171..4c8d1bd41694f 100644
--- a/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/LogsDataStreamIT.java
+++ b/modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/LogsDataStreamIT.java
@@ -20,6 +20,7 @@
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.matchesRegex;
 
 public class LogsDataStreamIT extends DisabledSecurityDataStreamTestCase {
 
@@ -100,6 +101,108 @@ public void testDefaultLogsSettingAndMapping() throws Exception {
         }
     }
 
+    @SuppressWarnings("unchecked")
+    public void testLogsDefaultPipeline() throws Exception {
+        RestClient client = client();
+        waitForLogs(client);
+
+        {
+            Request request = new Request("POST", "/_component_template/logs@custom");
+            request.setJsonEntity("""
+                {
+                  "template": {
+                    "mappings": {
+                      "properties": {
+                        "custom_timestamp": {
+                          "type": "date"
+                        }
+                      }
+                    }
+                  }
+                }
+                """);
+            assertOK(client.performRequest(request));
+        }
+        {
+            Request request = new Request("PUT", "/_ingest/pipeline/logs@custom");
+            request.setJsonEntity("""
+                    {
+                      "processors": [
+                        {
+                          "set" : {
+                            "field": "custom_timestamp",
+                            "copy_from": "_ingest.timestamp"
+                          }
+                        }
+                      ]
+                    }
+                """);
+            assertOK(client.performRequest(request));
+        }
+
+        String dataStreamName = "logs-generic-default";
+        createDataStream(client, dataStreamName);
+        String backingIndex = getWriteBackingIndex(client, dataStreamName);
+
+        // Verify mapping from custom logs
+        Map<String, Object> mappingProperties = getMappingProperties(client, backingIndex);
+        assertThat(((Map<String, Object>) mappingProperties.get("@timestamp")).get("type"), equalTo("date"));
+
+        // no timestamp - testing default pipeline's @timestamp set processor
+        {
+            indexDoc(client, dataStreamName, """
+                {
+                  "message": "no_timestamp"
+                }
+                """);
+            List<Object> results = searchDocs(client, dataStreamName, """
+                {
+                  "query": {
+                    "term": {
+                      "message": {
+                        "value": "no_timestamp"
+                      }
+                    }
+                  },
+                  "fields": ["@timestamp", "custom_timestamp"]
+                }
+                """);
+            Map<String, Object> source = ((Map<String, Map<String, Object>>) results.get(0)).get("_source");
+            String timestamp = (String) source.get("@timestamp");
+            assertThat(timestamp, matchesRegex("[0-9-]+T[0-9:.]+Z"));
+            assertThat(source.get("custom_timestamp"), is(timestamp));
+
+            Map<String, Object> fields = ((Map<String, Map<String, Object>>) results.get(0)).get("fields");
+            timestamp = ((List<String>) fields.get("@timestamp")).get(0);
+            assertThat(timestamp, matchesRegex("[0-9-]+T[0-9:.]+Z"));
+            assertThat(((List<Object>) fields.get("custom_timestamp")).get(0), is(timestamp));
+        }
+
+        // verify that when a document is ingested with a timestamp, it does not get overridden
+        {
+            indexDoc(client, dataStreamName, """
+                {
+                  "message": "with_timestamp",
+                  "@timestamp": "2023-05-10"
+                }
+                """);
+            List<Object> results = searchDocs(client, dataStreamName, """
+                {
+                  "query": {
+                    "term": {
+                      "message": {
+                        "value": "with_timestamp"
+                      }
+                    }
+                  },
+                  "fields": ["@timestamp", "custom_timestamp"]
+                }
+                """);
+            Map<String, Object> fields = ((Map<String, Map<String, Object>>) results.get(0)).get("fields");
+            assertThat(fields.get("@timestamp"), is(List.of("2023-05-10T00:00:00.000Z")));
+        }
+    }
+
     private static void waitForLogs(RestClient client) throws Exception {
         assertBusy(() -> {
             try {
diff --git a/modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/220_logs_default_pipeline.yml b/modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/220_logs_default_pipeline.yml
deleted file mode 100644
index 619df2382d955..0000000000000
--- a/modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/220_logs_default_pipeline.yml
+++ /dev/null
@@ -1,94 +0,0 @@
----
-Test default logs-*-* pipeline:
-  - do:
-      # setting up a custom field mapping, to test custom pipeline
-      cluster.put_component_template:
-        name: logs@custom
-        body:
-          template:
-            mappings:
-              properties:
-                custom_timestamp:
-                  type: date
-
-  - do:
-      ingest.put_pipeline:
-        # testing custom pipeline - setting a custom timestamp with the same value used to set the `@timestamp` field when missing
-        id: "logs@custom"
-        body:  >
-          {
-            "processors": [
-              {
-                "set" : {
-                  "field": "custom_timestamp",
-                  "copy_from": "_ingest.timestamp"
-                }
-              }
-            ]
-          }
-
-  - do:
-      indices.create_data_stream:
-        name: logs-generic-default
-  - is_true: acknowledged
-
-  - do:
-      indices.get_data_stream:
-        name: logs-generic-default
-  - set: { data_streams.0.indices.0.index_name: idx0name }
-
-  - do:
-      indices.get_mapping:
-        index: logs-generic-default
-  - match: { .$idx0name.mappings.properties.@timestamp.type: "date" }
-
-  - do:
-      index:
-        index: logs-generic-default
-        refresh: true
-        body:
-          # no timestamp - testing default pipeline's @timestamp set processor
-          message: 'no_timestamp'
-  - match: {result: "created"}
-
-  - do:
-      search:
-        index: logs-generic-default
-        body:
-          query:
-            term:
-              message:
-                value: 'no_timestamp'
-          fields:
-            - field: '@timestamp'
-            - field: 'custom_timestamp'
-  - length: { hits.hits: 1 }
-  - match: { hits.hits.0._source.@timestamp: '/[0-9-]+T[0-9:.]+Z/' }
-  - set: {hits.hits.0._source.custom_timestamp: custom_timestamp_source }
-  - match: { hits.hits.0._source.@timestamp: $custom_timestamp_source }
-  - match: { hits.hits.0.fields.@timestamp.0: '/[0-9-]+T[0-9:.]+Z/' }
-  - set: {hits.hits.0.fields.custom_timestamp.0: custom_timestamp_field }
-  - match: { hits.hits.0.fields.@timestamp.0: $custom_timestamp_field }
-
-  # verify that when a document is ingested with a timestamp, it does not get overridden
-  - do:
-      index:
-        index: logs-generic-default
-        refresh: true
-        body:
-          '@timestamp': '2023-05-10'
-          message: 'with_timestamp'
-  - match: {result: "created"}
-
-  - do:
-      search:
-        index: logs-generic-default
-        body:
-          query:
-            term:
-              message:
-                value: 'with_timestamp'
-          fields:
-            - field: '@timestamp'
-  - length: { hits.hits: 1 }
-  - match: { hits.hits.0.fields.@timestamp.0: '2023-05-10T00:00:00.000Z' }