Fix Edge Case Datastream Snapshot Create Bug (#72747) (#72749)

We always clean up the list of datastreams depending on the indices acutally
in the snapshot. Even for non-partial snapshots datastream indices could be
excluded from the snapshot by index exlusions that remove all the datastream's
indices from the snapshot.

Author: original-brownbear

server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

@@ -1088,8 +1088,6 @@ private static Metadata metadataForSnapshot(SnapshotsInProgress.Entry snapshot,
                 for (Index index : dataStream.getIndices()) {
                     final String indexName = index.getName();
                     if (builder.get(indexName) == null || indicesInSnapshot.contains(indexName) == false) {
-                        assert snapshot.partial() : "Data stream [" + dataStreamName +
-                                "] is missing index [" + index + "] but snapshot was not partial.";
                         missingIndex = true;
                         break;
                     }
@@ -1691,9 +1689,8 @@ private void finalizeSnapshotEntry(SnapshotsInProgress.Entry entry, Metadata met
                         final SnapshotInfo snapshotInfo = new SnapshotInfo(
                                 snapshot.getSnapshotId(),
                                 finalIndices,
-                                entry.partial() ? entry.dataStreams().stream()
-                                        .filter(metaForSnapshot.dataStreams()::containsKey)
-                                        .collect(Collectors.toList()) : entry.dataStreams(),
+                                entry.dataStreams().stream().filter(metaForSnapshot.dataStreams()::containsKey)
+                                    .collect(Collectors.toList()),
                                 entry.partial() ? onlySuccessfulFeatureStates(entry, finalIndices) : entry.featureStates(),
                                 failure,
                                 threadPool.absoluteTimeInMillis(),

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

@@ -715,4 +715,22 @@ public void testSnapshotDSDuringRolloverAndDeleteOldIndex() throws Exception {
         assertEquals(restoreSnapshotResponse.successfulShards(), restoreSnapshotResponse.totalShards());
         assertEquals(restoreSnapshotResponse.failedShards(), 0);
     }
+
+    public void testExcludeDSFromSnapshotWhenExcludingItsIndices() {
+        final String snapshot = "test-snapshot";
+        final String indexWithoutDataStream = "test-idx-no-ds";
+        createIndexWithContent(indexWithoutDataStream);
+        final SnapshotInfo snapshotInfo = createSnapshot(REPO, snapshot, List.of("*", "-.*"));
+        assertThat(snapshotInfo.dataStreams(), empty());
+        assertAcked(client.admin().indices().prepareDelete(indexWithoutDataStream));
+        RestoreInfo restoreInfo = client.admin()
+            .cluster()
+            .prepareRestoreSnapshot(REPO, snapshot)
+            .setWaitForCompletion(true)
+            .setRestoreGlobalState(randomBoolean())
+            .get()
+            .getRestoreInfo();
+        assertThat(restoreInfo.failedShards(), is(0));
+        assertThat(restoreInfo.successfulShards(), is(1));
+    }
 }