diff --git a/server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java b/server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
index 3010ff0bea0..d63eafaeed5 100644
--- a/server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
+++ b/server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
@@ -446,6 +446,18 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
                 }
                 Index concreteIndex = concreteIndices.resolveIfAbsent(docWriteRequest);
                 try {
+                    // The ConcreteIndices#resolveIfAbsent(...) method validates via IndexNameExpressionResolver whether
+                    // an operation is allowed in index into a data stream, but this isn't done when resolve call is cached, so
+                    // the validation needs to be performed here too.
+                    IndexAbstraction indexAbstraction = clusterState.getMetadata().getIndicesLookup().get(concreteIndex.getName());
+                    if (indexAbstraction.getParentDataStream() != null &&
+                        // avoid valid cases when directly indexing into a backing index
+                        // (for example when directly indexing into .ds-logs-foobar-000001)
+                        concreteIndex.getName().equals(docWriteRequest.index()) == false &&
+                        docWriteRequest.opType() != DocWriteRequest.OpType.CREATE) {
+                        throw new IllegalArgumentException("only write ops with an op_type of create are allowed in data streams");
+                    }
+
                     switch (docWriteRequest.opType()) {
                         case CREATE:
                         case INDEX:
diff --git a/x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java b/x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java
index 3361d790a21..d54ce554996 100644
--- a/x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java
+++ b/x-pack/plugin/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamIT.java
@@ -284,6 +284,118 @@ public class DataStreamIT extends ESIntegTestCase {
             BulkResponse bulkItemResponses = client().bulk(bulkRequest).actionGet();
             assertThat(bulkItemResponses.getItems()[0].getIndex(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 1)));
         }
+
+        {
+            // TODO: remove when fixing the bug when an index matching a backing index name is created before the data stream is created
+            createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName + "-baz");
+            client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
+
+            BulkRequest bulkRequest = new BulkRequest().add(
+                new IndexRequest(dataStreamName).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON),
+                new IndexRequest(dataStreamName).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON).create(true),
+                new IndexRequest(dataStreamName).source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON),
+                new UpdateRequest(dataStreamName, "_id").doc("{\"@timestamp1\": \"2020-12-12\"}", XContentType.JSON),
+                new DeleteRequest(dataStreamName, "_id"),
+                new IndexRequest(dataStreamName + "-baz").source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON).create(true),
+                new DeleteRequest(dataStreamName + "-baz", "_id"),
+                new IndexRequest(dataStreamName + "-baz").source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON),
+                new IndexRequest(dataStreamName + "-baz").source("{\"@timestamp\": \"2020-12-12\"}", XContentType.JSON).create(true),
+                // Non create ops directly against backing indices are allowed:
+                new DeleteRequest(DataStream.getDefaultBackingIndexName(dataStreamName + "-baz", 1), "_id"),
+                new IndexRequest(DataStream.getDefaultBackingIndexName(dataStreamName + "-baz", 1)).source(
+                    "{\"@timestamp\": \"2020-12-12\"}",
+                    XContentType.JSON
+                ).id("_id").setIfSeqNo(1).setIfPrimaryTerm(1)
+            );
+            BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
+            assertThat(bulkResponse.getItems(), arrayWithSize(11));
+            {
+                assertThat(bulkResponse.getItems()[0].getFailure(), notNullValue());
+                assertThat(bulkResponse.getItems()[0].getResponse(), nullValue());
+                assertThat(
+                    bulkResponse.getItems()[0].getFailure().getMessage(),
+                    containsString("only write ops with an op_type of create are allowed in data streams")
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[1].getFailure(), nullValue());
+                assertThat(bulkResponse.getItems()[1].getResponse(), notNullValue());
+                assertThat(bulkResponse.getItems()[1].getIndex(), equalTo(DataStream.getDefaultBackingIndexName(dataStreamName, 1)));
+            }
+            {
+                assertThat(bulkResponse.getItems()[2].getFailure(), notNullValue());
+                assertThat(bulkResponse.getItems()[2].getResponse(), nullValue());
+                assertThat(
+                    bulkResponse.getItems()[2].getFailure().getMessage(),
+                    containsString("only write ops with an op_type of create are allowed in data streams")
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[3].getFailure(), notNullValue());
+                assertThat(bulkResponse.getItems()[3].getResponse(), nullValue());
+                assertThat(
+                    bulkResponse.getItems()[3].getFailure().getMessage(),
+                    containsString("only write ops with an op_type of create are allowed in data streams")
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[4].getFailure(), notNullValue());
+                assertThat(bulkResponse.getItems()[4].getResponse(), nullValue());
+                assertThat(
+                    bulkResponse.getItems()[4].getFailure().getMessage(),
+                    containsString("only write ops with an op_type of create are allowed in data streams")
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[5].getFailure(), nullValue());
+                assertThat(bulkResponse.getItems()[5].getResponse(), notNullValue());
+                assertThat(
+                    bulkResponse.getItems()[5].getIndex(),
+                    equalTo(DataStream.getDefaultBackingIndexName(dataStreamName + "-baz", 1))
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[6].getFailure(), notNullValue());
+                assertThat(bulkResponse.getItems()[6].getResponse(), nullValue());
+                assertThat(
+                    bulkResponse.getItems()[6].getFailure().getMessage(),
+                    containsString("only write ops with an op_type of create are allowed in data streams")
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[7].getFailure(), notNullValue());
+                assertThat(bulkResponse.getItems()[7].getResponse(), nullValue());
+                assertThat(
+                    bulkResponse.getItems()[7].getFailure().getMessage(),
+                    containsString("only write ops with an op_type of create are allowed in data streams")
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[8].getFailure(), nullValue());
+                assertThat(bulkResponse.getItems()[8].getResponse(), notNullValue());
+                assertThat(
+                    bulkResponse.getItems()[8].getIndex(),
+                    equalTo(DataStream.getDefaultBackingIndexName(dataStreamName + "-baz", 1))
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[9].getFailure(), nullValue());
+                assertThat(bulkResponse.getItems()[9].getResponse(), notNullValue());
+                assertThat(
+                    bulkResponse.getItems()[9].getIndex(),
+                    equalTo(DataStream.getDefaultBackingIndexName(dataStreamName + "-baz", 1))
+                );
+            }
+            {
+                assertThat(bulkResponse.getItems()[10].getResponse(), nullValue());
+                assertThat(bulkResponse.getItems()[10].getFailure(), notNullValue());
+                assertThat(bulkResponse.getItems()[10].status(), equalTo(RestStatus.CONFLICT));
+                assertThat(
+                    bulkResponse.getItems()[10].getIndex(),
+                    equalTo(DataStream.getDefaultBackingIndexName(dataStreamName + "-baz", 1))
+                );
+            }
+        }
     }
 
     /**
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/20_unsupported_apis.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/20_unsupported_apis.yml
index 06a800b061f..bba2a7f4881 100644
--- a/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/20_unsupported_apis.yml
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/20_unsupported_apis.yml
@@ -200,8 +200,8 @@
 ---
 "Non append-only writes into a data stream":
   - skip:
-      version: " - 7.8.99"
-      reason: "data streams only supported in 7.9+"
+      version: " - 7.99.99"
+      reason: "set to 7.9.3 after backporting #62766"
       features: allowed_warnings
 
   - do:
@@ -232,12 +232,25 @@
               _index: logs-foobar
               _id: "1"
           - '@timestamp': '2020-12-12'
+          - index:
+              _index: logs-foobar
+              _id: "1"
+          - '@timestamp': '2020-12-12'
+          - delete:
+              _index: logs-foobar
+              _id: 10
   - match: { errors: true }
   - match: { items.0.index.status: 400 }
   - match: { items.0.index.error.type: illegal_argument_exception }
   - match: { items.0.index.error.reason: "only write ops with an op_type of create are allowed in data streams" }
   - match: { items.1.create.result: created }
   - match: { items.1.create._index: .ds-logs-foobar-000001 }
+  - match: { items.2.index.status: 400 }
+  - match: { items.2.index.error.type: illegal_argument_exception }
+  - match: { items.2.index.error.reason: "only write ops with an op_type of create are allowed in data streams" }
+  - match: { items.3.delete.status: 400 }
+  - match: { items.3.delete.error.type: illegal_argument_exception }
+  - match: { items.3.delete.error.reason: "only write ops with an op_type of create are allowed in data streams" }
 
   - do:
       indices.delete_data_stream: