Revert "Fix propagation for clones"

This reverts commit a29ba9b.

Author: ywangd

server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

@@ -1291,14 +1291,8 @@ public Entry abort(Map<ShardId, ShardSnapshotStatus> completedAssignedQueuedShar
                     } else {
                         // Record the deletion of the assigned-queued shard snapshot so that we can kick off the first QUEUED one
                         // in later snapshots.
-                        final ShardId shardId = shardEntry.getKey();
-                        final var old = completedAssignedQueuedShards.put(shardId, status);
-                        assert old == null
-                            : shardId
-                                + " has unexpected previous completed shard snapshot status: "
-                                + old
-                                + " that conflicts with the one from "
-                                + this;
+                        final var old = completedAssignedQueuedShards.put(shardEntry.getKey(), status);
+                        assert old == null : old;
                         assert isClone() == false
                             : "The state queued with generation should not be possible for a clone entry [" + this + "]";
                         final String reason = "assigned-queued aborted by snapshot deletion";

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

@@ -1630,22 +1630,13 @@ public ClusterState execute(ClusterState currentState) {
                             }
                             return abortedEntry;
                         } else {
-                            if (existing.isClone()) {
-                                final var clonesBuilder = maybeUpdateCloneWithCompletedAssignedQueuedShards(
-                                    existing,
-                                    completedAssignedQueuedShards,
-                                    currentState.nodes().getLocalNodeId(),
-                                    () -> ImmutableOpenMap.builder(existing.shardSnapshotStatusByRepoShardId())
-                                );
-                                return clonesBuilder == null ? existing : existing.withClones(clonesBuilder.build());
-                            } else {
-                                final var shardsBuilder = maybeUpdateSnapshotWithCompletedAssignedQueuedShards(
-                                    existing,
-                                    completedAssignedQueuedShards,
-                                    () -> ImmutableOpenMap.builder(existing.shards())
-                                );
-                                return shardsBuilder == null ? existing : existing.withShardStates(shardsBuilder.build());
-                            }
+                            final var shardsBuilder = maybeUpdateWithCompletedAssignedQueuedShards(
+                                existing,
+                                completedAssignedQueuedShards,
+                                true,
+                                () -> ImmutableOpenMap.builder(existing.shards())
+                            );
+                            return shardsBuilder == null ? existing : existing.withShardStates(shardsBuilder.build());
                         }
                     }).filter(Objects::nonNull).toList()
                 );
@@ -1760,86 +1751,43 @@ public String toString() {
     }
 
     /**
-     * Update a snapshot entry with the completed assigned queue shard snapshots accumulated from processing earlier snapshots.
+     * Update the snapshot entry with the completed assigned queue shard snapshots accumulated from processing earlier snapshots.
      * These shard snapshots are completed because either their snapshots are deleted or the associated shards become unassigned.
      * In both cases, the shard snapshots for the same shards must be updated to propagate the changes.
      * @param entry The snapshot entry to update.
      * @param completedAssignedQueuedShards Completed assigned queued shard snapshots accumulated from processing earlier snapshots.
-     *                                      An entry is removed if it is not MISSING state so that it only updates to the first match.
+     * @param removeAfterUpdate If true, the completed assigned queued shards will be removed from completedAssignedQueuedShards
+     *                          after updating the entry.
      * @param shardsBuilderSupplier Supplier to get a builder for new shard snapshots if any update is applicable.
      * @return The updated map of shard snapshots for the snapshot entry, or null if no updates were made.
      */
-    static ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> maybeUpdateSnapshotWithCompletedAssignedQueuedShards(
+    static ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> maybeUpdateWithCompletedAssignedQueuedShards(
         SnapshotsInProgress.Entry entry,
         Map<ShardId, ShardSnapshotStatus> completedAssignedQueuedShards,
+        boolean removeAfterUpdate,
         Supplier<ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus>> shardsBuilderSupplier
     ) {
-        assert entry.isClone() == false : "unexpected clone " + entry;
-        if (entry.state().completed() || completedAssignedQueuedShards.isEmpty()) {
-            return null;
-        }
-        ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> shardsBuilder = null;
-        final var iterator = completedAssignedQueuedShards.keySet().iterator();
-        while (iterator.hasNext()) {
-            final var shardId = iterator.next();
-            final var shardSnapshotStatus = entry.shards().get(shardId);
-            if (shardSnapshotStatus != null) {
-                assert shardSnapshotStatus.isUnassignedQueued() : shardSnapshotStatus;
-                if (shardsBuilder == null) {
-                    shardsBuilder = shardsBuilderSupplier.get();
-                }
-                final ShardSnapshotStatus completedStatus = completedAssignedQueuedShards.get(shardId);
-                if (completedStatus.state() == ShardState.MISSING) {
-                    shardsBuilder.put(shardId, completedStatus);
-                } else {
-                    shardsBuilder.put(shardId, ShardSnapshotStatus.assignedQueued(completedStatus.nodeId(), completedStatus.generation()));
-                    iterator.remove();
-                }
-            }
-        }
-        return shardsBuilder;
-    }
-
-    /**
-     * Update a clone entry with the completed assigned queue shard snapshots accumulated from processing earlier snapshots.
-     * These shard snapshots are completed because either their snapshots are deleted or the associated shards become unassigned.
-     * In both cases, previous queued clone shard snapshots for the same shards can now start.
-     * @param entry The snapshot entry to update.
-     * @param completedAssignedQueuedShards Completed assigned queued shard snapshots accumulated from processing earlier snapshots.
-     *                                      Once an entry is applied for update, it is removed from the list.
-     * @param clonesBuilderSupplier Supplier to get a builder for new clone shard snapshots if any update is applicable.
-     * @return The updated map of clone shard snapshots for the snapshot entry, or null if no updates were made.
-     */
-    static ImmutableOpenMap.Builder<RepositoryShardId, ShardSnapshotStatus> maybeUpdateCloneWithCompletedAssignedQueuedShards(
-        SnapshotsInProgress.Entry entry,
-        Map<ShardId, ShardSnapshotStatus> completedAssignedQueuedShards,
-        String localNodeId,
-        Supplier<ImmutableOpenMap.Builder<RepositoryShardId, ShardSnapshotStatus>> clonesBuilderSupplier
-    ) {
-        assert entry.isClone() : "expected a clone, but got " + entry;
-        if (entry.state().completed() || completedAssignedQueuedShards.isEmpty()) {
-            return null;
-        }
-        ImmutableOpenMap.Builder<RepositoryShardId, ShardSnapshotStatus> clonesBuilder = null;
-        final var iterator = completedAssignedQueuedShards.keySet().iterator();
-        while (iterator.hasNext()) {
-            final var shardId = iterator.next();
-            final IndexId indexId = entry.indices().get(shardId.getIndexName());
-            if (indexId != null) {
-                final RepositoryShardId repoShardId = new RepositoryShardId(indexId, shardId.id());
-                if (SnapshotsServiceUtils.isQueued(entry.shardSnapshotStatusByRepoShardId().get(repoShardId))) {
-                    if (clonesBuilder == null) {
-                        clonesBuilder = clonesBuilderSupplier.get();
+        if (entry.isClone() == false && completedAssignedQueuedShards.isEmpty() == false) {
+            ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> shardsBuilder = null;
+            final var iterator = completedAssignedQueuedShards.keySet().iterator();
+            while (iterator.hasNext()) {
+                final var shardId = iterator.next();
+                final var shardSnapshotStatus = entry.shards().get(shardId);
+                if (shardSnapshotStatus != null) {
+                    assert shardSnapshotStatus.isUnassignedQueued() : shardSnapshotStatus;
+                    if (shardsBuilder == null) {
+                        shardsBuilder = shardsBuilderSupplier.get();
+                    }
+                    shardsBuilder.put(shardId, completedAssignedQueuedShards.get(shardId));
+                    if (removeAfterUpdate) {
+                        iterator.remove();
                     }
-                    clonesBuilder.put(
-                        repoShardId,
-                        new ShardSnapshotStatus(localNodeId, completedAssignedQueuedShards.get(shardId).generation())
-                    );
-                    iterator.remove();
                 }
             }
+            return shardsBuilder;
+        } else {
+            return null;
         }
-        return clonesBuilder;
     }
 
     private void addDeleteListener(String deleteUUID, ActionListener<Void> listener) {
@@ -2572,49 +2520,36 @@ private SnapshotsInProgress maybeStartAssignedQueuedShardSnapshotsForRepo(
             final List<SnapshotsInProgress.Entry> newEntries = new ArrayList<>(oldEntries.size());
             final Map<ShardId, ShardSnapshotStatus> completedAssignedQueuedShards = new HashMap<>();
             for (SnapshotsInProgress.Entry entry : oldEntries) {
-                if (entry.isClone()) {
-                    final var clonesBuilder = maybeUpdateCloneWithCompletedAssignedQueuedShards(
+                ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> shardsBuilder = maybeUpdateWithCompletedAssignedQueuedShards(
+                    entry,
+                    completedAssignedQueuedShards,
+                    false,
+                    () -> ImmutableOpenMap.builder(entry.shards())
+                );
+                boolean changed = shardsBuilder != null;
+
+                if (entry.hasAssignedQueuedShards() && perNodeShardSnapshotCounter.hasCapacityOnAnyNode()) {
+                    if (shardsBuilder == null) {
+                        shardsBuilder = ImmutableOpenMap.builder(entry.shards());
+                    }
+                    changed |= maybeStartAssignedQueuedShardSnapshots(
+                        clusterState,
                         entry,
-                        completedAssignedQueuedShards,
-                        initialState.nodes().getLocalNodeId(),
-                        () -> ImmutableOpenMap.builder(entry.shardSnapshotStatusByRepoShardId())
+                        snapshotsInProgress::isNodeIdForRemoval,
+                        shardsBuilder,
+                        perNodeShardSnapshotCounter,
+                        completedAssignedQueuedShards
                     );
-                    if (clonesBuilder != null) {
-                        newEntries.add(entry.withClones(clonesBuilder.build()));
-                    } else {
-                        newEntries.add(entry);
-                    }
-                } else {
-                    ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> shardsBuilder =
-                        maybeUpdateSnapshotWithCompletedAssignedQueuedShards(
-                            entry,
-                            completedAssignedQueuedShards,
-                            () -> ImmutableOpenMap.builder(entry.shards())
-                        );
-                    boolean changed = shardsBuilder != null;
+                }
 
-                    if (entry.hasAssignedQueuedShards() && perNodeShardSnapshotCounter.hasCapacityOnAnyNode()) {
-                        if (shardsBuilder == null) {
-                            shardsBuilder = ImmutableOpenMap.builder(entry.shards());
-                        }
-                        changed |= maybeStartAssignedQueuedShardSnapshots(
-                            clusterState,
-                            entry,
-                            snapshotsInProgress::isNodeIdForRemoval,
-                            shardsBuilder,
-                            perNodeShardSnapshotCounter,
-                            completedAssignedQueuedShards
-                        );
-                    }
-                    if (changed) {
-                        final var newEntry = entry.withShardStates(shardsBuilder.build());
-                        newEntries.add(newEntry);
-                        if (newEntry.state().completed()) {
-                            newlyCompletedEntries.add(newEntry);
-                        }
-                    } else {
-                        newEntries.add(entry);
+                if (changed) {
+                    final var newEntry = entry.withShardStates(shardsBuilder.build());
+                    newEntries.add(newEntry);
+                    if (newEntry.state().completed()) {
+                        newlyCompletedEntries.add(newEntry);
                     }
+                } else {
+                    newEntries.add(entry);
                 }
             }
             return snapshotsInProgress.createCopyWithUpdatedEntriesForRepo(projectRepo.projectId(), projectRepo.name(), newEntries);
@@ -2654,17 +2589,10 @@ private boolean maybeStartAssignedQueuedShardSnapshots(
                 );
                 if (newShardSnapshotStatus.equals(existingShardSnapshotStatus) == false) {
                     if (newShardSnapshotStatus.state().completed()) {
-                        assert newShardSnapshotStatus.state() == ShardState.MISSING
-                            : shardId + " has unexpected completion state " + newShardSnapshotStatus + " in snapshot entry " + entry;
+                        assert newShardSnapshotStatus.state() == ShardState.MISSING : newShardSnapshotStatus;
                         // There might be QUEUED shard snapshots of this shard in later snapshots and we need
                         // to propagate the completion to them.
-                        final var old = completedAssignedQueuedShards.put(shardId, newShardSnapshotStatus);
-                        assert old == null
-                            : shardId
-                                + " has unexpected previous completed shard snapshot status: "
-                                + old
-                                + " that conflicts with the one from "
-                                + entry;
+                        completedAssignedQueuedShards.put(shardId, newShardSnapshotStatus);
                     }
                     changedCount++;
                     if (newShardSnapshotStatus.state() == ShardState.INIT) {
@@ -2721,7 +2649,7 @@ private SnapshotsInProgress.Entry applyUpdatesToEntry(
         ) {
             // Completed snapshots do not require any updates so we just add them to the output list and keep going.
             // Also we short circuit if there are no more unconsumed updates to apply.
-            if (entry.state().completed() || (shardUpdates.isEmpty() && completedAssignedQueuedShards.isEmpty())) {
+            if (entry.state().completed() || shardUpdates.isEmpty()) {
                 return entry;
             }
             return new EntryContext(entry, shardUpdates, completedAssignedQueuedShards).computeUpdatedSnapshotEntryFromShardUpdates();
@@ -2789,16 +2717,7 @@ SnapshotsInProgress.Entry computeUpdatedSnapshotEntryFromShardUpdates() {
                     }
                 }
 
-                if (entry.isClone()) {
-                    maybeUpdateCloneWithCompletedAssignedQueuedShards(
-                        entry,
-                        completedAssignedQueuedShards,
-                        initialState.nodes().getLocalNodeId(),
-                        this::clonesBuilder
-                    );
-                } else {
-                    maybeUpdateSnapshotWithCompletedAssignedQueuedShards(entry, completedAssignedQueuedShards, this::shardsBuilder);
-                }
+                maybeUpdateWithCompletedAssignedQueuedShards(entry, completedAssignedQueuedShards, false, this::shardsBuilder);
 
                 if (shardsBuilder != null) {
                     assert clonesBuilder == null

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

@@ -413,7 +413,7 @@ public static ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus>
             .entrySet()) {
             SnapshotsInProgress.ShardSnapshotStatus shardStatus = shardSnapshotEntry.getValue();
             ShardId shardId = snapshotEntry.shardId(shardSnapshotEntry.getKey());
-            if (shardStatus.isUnassignedQueued() || shardStatus.isAssignedQueued()) {
+            if (shardStatus.equals(SnapshotsInProgress.ShardSnapshotStatus.UNASSIGNED_QUEUED)) {
                 // this shard snapshot is waiting for a previous snapshot to finish execution for this shard
                 final SnapshotsInProgress.ShardSnapshotStatus knownFailure = knownFailures.get(shardSnapshotEntry.getKey());
                 if (knownFailure == null) {
@@ -422,30 +422,14 @@ public static ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus>
                         // shard became unassigned while queued after a delete or clone operation so we can fail as missing here
                         assert snapshotEntry.partial();
                         snapshotChanged = true;
-                        logger.debug("failing snapshot of shard [{}] with status [{}] because index got deleted", shardId, shardStatus);
-                        final SnapshotsInProgress.ShardSnapshotStatus newShardStatus = shardStatus.isAssignedQueued()
-                            ? new SnapshotsInProgress.ShardSnapshotStatus(
-                                shardStatus.nodeId(),
-                                SnapshotsInProgress.ShardState.FAILED,
-                                shardStatus.generation(),
-                                "shard is deleted"
-                            )
-                            : SnapshotsInProgress.ShardSnapshotStatus.MISSING;
-                        shards.put(shardId, newShardStatus);
-                        knownFailures.put(shardSnapshotEntry.getKey(), newShardStatus);
+                        logger.debug("failing snapshot of shard [{}] because index got deleted", shardId);
+                        shards.put(shardId, SnapshotsInProgress.ShardSnapshotStatus.MISSING);
+                        knownFailures.put(shardSnapshotEntry.getKey(), SnapshotsInProgress.ShardSnapshotStatus.MISSING);
                     } else {
                         // if no failure is known for the shard we keep waiting
                         shards.put(shardId, shardStatus);
                     }
                 } else {
-                    assert shardStatus.isAssignedQueued() == false
-                        : shardId
-                            + " with status "
-                            + shardStatus
-                            + " has unexpected known failure "
-                            + knownFailure
-                            + " in snapshot entry "
-                            + snapshotEntry;
                     // If a failure is known for an execution we waited on for this shard then we fail with the same exception here
                     // as well
                     snapshotChanged = true;