Change back to assigned queued

Generation can still be null

Author: ywangd

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

@@ -509,7 +509,7 @@ private static boolean assertShardStateConsistent(
         int shardId,
         ShardSnapshotStatus shardSnapshotStatus
     ) {
-        if (shardSnapshotStatus.isActiveOrQueuedWithGeneration()) {
+        if (shardSnapshotStatus.isActiveOrAssignedQueued()) {
             Tuple<String, Integer> plainShardId = Tuple.tuple(indexName, shardId);
             assert assignedShards.add(plainShardId) : plainShardId + " is assigned twice in " + entries;
             assert queuedShards.contains(plainShardId) == false : plainShardId + " is queued then assigned in " + entries;
@@ -758,12 +758,13 @@ public static ShardSnapshotStatus success(String nodeId, ShardSnapshotResult sha
         }
 
         @SuppressForbidden(reason = "using a private constructor within the same file")
-        public static ShardSnapshotStatus queuedWithGeneration(ShardGeneration generation) {
-            return new ShardSnapshotStatus(null, ShardState.QUEUED, generation, null, null);
+        public static ShardSnapshotStatus assignedQueued(String nodeId, ShardGeneration generation) {
+            return new ShardSnapshotStatus(nodeId, ShardState.QUEUED, generation, null, null);
         }
 
-        public boolean isQueuedWithGeneration() {
-            return state == ShardState.QUEUED && generation != null && nodeId == null;
+        public boolean isAssignedQueued() {
+            // generation can still be null if previous shard snapshots all failed
+            return state == ShardState.QUEUED && nodeId != null;
         }
 
         public boolean isUnassignedQueued() {
@@ -790,7 +791,7 @@ private boolean assertConsistent() {
             assert state.failed() == false || reason != null;
             assert (state != ShardState.INIT && state != ShardState.WAITING && state != ShardState.PAUSED_FOR_NODE_REMOVAL)
                 || nodeId != null : "Null node id for state [" + state + "]";
-            assert state != ShardState.QUEUED || (isUnassignedQueued() || isQueuedWithGeneration())
+            assert state != ShardState.QUEUED || (isUnassignedQueued() || isAssignedQueued())
                 : "Found unexpected shard state=["
                     + state
                     + "], nodeId=["
@@ -853,8 +854,8 @@ public boolean isActive() {
             };
         }
 
-        public boolean isActiveOrQueuedWithGeneration() {
-            return isActive() || isQueuedWithGeneration();
+        public boolean isActiveOrAssignedQueued() {
+            return isActive() || isAssignedQueued();
         }
 
         @Override
@@ -1230,30 +1231,38 @@ public Entry withClones(Map<RepositoryShardId, ShardSnapshotStatus> updatedClone
          * @return aborted snapshot entry or {@code null} if entry can be removed from the cluster state directly
          */
         @Nullable
-        public Entry abort(BiConsumer<ShardId, ShardSnapshotStatus> abortedQueuedWithGenerationShardConsumer) {
+        public Entry abort(BiConsumer<ShardId, ShardSnapshotStatus> abortedAssignedQueuedShardConsumer) {
             final Map<ShardId, ShardSnapshotStatus> shardsBuilder = new HashMap<>();
             boolean completed = true;
             boolean allQueued = true;
             for (Map.Entry<ShardId, ShardSnapshotStatus> shardEntry : shards.entrySet()) {
                 ShardSnapshotStatus status = shardEntry.getValue();
-                final var isQueuedWithGeneration = status.isQueuedWithGeneration();
-                allQueued &= (status.state() == ShardState.QUEUED && isQueuedWithGeneration == false);
+                final var isAssignedQueued = status.isAssignedQueued();
+                allQueued &= (status.state() == ShardState.QUEUED && isAssignedQueued == false);
                 if (status.state().completed() == false) {
                     final String nodeId = status.nodeId();
-                    status = new ShardSnapshotStatus(
-                        nodeId,
-                        // QUEUED with generation transitioned to ABORTED (incomplete) and is completed by a separate cluster state update
-                        (nodeId == null && isQueuedWithGeneration == false) ? ShardState.FAILED : ShardState.ABORTED,
-                        status.generation(),
-                        "aborted by snapshot deletion"
-                    );
-                    if (isQueuedWithGeneration) {
-                        // Accumulate the updates needed to complete the aborted QUEUED with generation shard snapshots
+                    if (isAssignedQueued == false) {
+                        status = new ShardSnapshotStatus(
+                            nodeId,
+                            nodeId == null ? ShardState.FAILED : ShardState.ABORTED,
+                            status.generation(),
+                            "aborted by snapshot deletion"
+                        );
+                    } else {
                         assert isClone() == false
                             : "The state queued with generation should not be possible for a clone entry [" + this + "]";
-                        abortedQueuedWithGenerationShardConsumer.accept(
+                        final String reason = "assigned-queued aborted by snapshot deletion";
+                        status = new ShardSnapshotStatus(
+                            nodeId,
+                            // Assigned QUEUED transitions to ABORTED (incomplete) and is completed by a separate cluster state update
+                            ShardState.ABORTED,
+                            status.generation(),
+                            reason
+                        );
+                        // Accumulate the updates needed to complete the aborted QUEUED with generation shard snapshots
+                        abortedAssignedQueuedShardConsumer.accept(
                             shardEntry.getKey(),
-                            new ShardSnapshotStatus(null, ShardState.FAILED, status.generation, "aborted by snapshot deletion")
+                            new ShardSnapshotStatus(null, ShardState.FAILED, status.generation, reason)
                         );
                     }
                 }

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

@@ -68,7 +68,7 @@ private static void addStateInformation(
         String indexName
     ) {
         // Both active or assigned queued means the shard is meant to be the one with actions if node capacity allows it
-        if (shardState.isActiveOrQueuedWithGeneration()) {
+        if (shardState.isActiveOrAssignedQueued()) {
             busyIds.computeIfAbsent(indexName, k -> new HashSet<>()).add(shardId);
             assert assertGenerationConsistency(generations, indexName, shardId, shardState.generation());
         } else if (shardState.state() == SnapshotsInProgress.ShardState.SUCCESS) {

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

@@ -44,6 +44,7 @@ public PerNodeShardSnapshotCounter(
                     return;
                 }
                 for (var shardSnapshotStatus : entry.shards().values()) {
+                    // TODO: consider more states as active, e.g. Abort on data node?
                     if (shardSnapshotStatus.state() == SnapshotsInProgress.ShardState.INIT) {
                         perNodeCounts.computeIfPresent(shardSnapshotStatus.nodeId(), (nodeId, count) -> count + 1);
                     }

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

@@ -1013,7 +1013,7 @@ private static boolean assertNoDanglingSnapshots(ClusterState state) {
                 if (value.equals(ShardSnapshotStatus.UNASSIGNED_QUEUED)) {
                     assert reposWithRunningDelete.contains(new ProjectRepo(entry.projectId(), entry.repository()))
                         : "Found shard snapshot waiting to be assigned in [" + entry + "] but it is not blocked by any running delete";
-                } else if (value.isActiveOrQueuedWithGeneration()) {
+                } else if (value.isActiveOrAssignedQueued()) {
                     assert reposWithRunningDelete.contains(new ProjectRepo(entry.projectId(), entry.repository())) == false
                         : "Found shard snapshot actively executing in ["
                             + entry
@@ -2296,7 +2296,7 @@ public ClusterState execute(ClusterState currentState) {
                 // Snapshot ids that will have to be physically deleted from the repository
                 final Set<SnapshotId> snapshotIdsRequiringCleanup = new HashSet<>(snapshotIds);
 
-                final List<Runnable> completeAbortedQueuedWithGenerationRunnables = new ArrayList<>();
+                final List<Runnable> completeAbortedAssignedQueuedRunnables = new ArrayList<>();
                 final SnapshotsInProgress updatedSnapshots = snapshotsInProgress.createCopyWithUpdatedEntriesForRepo(
                     projectId,
                     repositoryName,
@@ -2305,7 +2305,7 @@ public ClusterState execute(ClusterState currentState) {
                             && snapshotIdsRequiringCleanup.contains(existing.snapshot().getSnapshotId())) {
                             // snapshot is started - mark every non completed shard as aborted
                             final SnapshotsInProgress.Entry abortedEntry = existing.abort(
-                                ((shardId, shardSnapshotStatus) -> completeAbortedQueuedWithGenerationRunnables.add(
+                                ((shardId, shardSnapshotStatus) -> completeAbortedAssignedQueuedRunnables.add(
                                     () -> innerUpdateSnapshotState(
                                         existing.snapshot(),
                                         shardId,
@@ -2332,7 +2332,7 @@ public ClusterState execute(ClusterState currentState) {
                         return existing;
                     }).filter(Objects::nonNull).toList()
                 );
-                for (var runnable : completeAbortedQueuedWithGenerationRunnables) {
+                for (var runnable : completeAbortedAssignedQueuedRunnables) {
                     runnable.run();
                 }
                 if (snapshotIdsRequiringCleanup.isEmpty()) {
@@ -2457,7 +2457,7 @@ private static boolean isWritingToRepositoryOrQueueWithGeneration(SnapshotsInPro
             return true;
         }
         for (ShardSnapshotStatus value : entry.shardSnapshotStatusByRepoShardId().values()) {
-            if (value.isActiveOrQueuedWithGeneration()) {
+            if (value.isActiveOrAssignedQueued()) {
                 // Entry is writing to the repo because it's writing to a shard on a data node or waiting to do so for a concrete shard
                 return true;
             }
@@ -3059,7 +3059,7 @@ private SnapshotsInProgress updatedSnapshotsInProgress(ClusterState currentState
                                         : "Missing assignment for [" + sid + "]";
                                     updatedAssignmentsBuilder.put(sid, ShardSnapshotStatus.MISSING);
                                 } else {
-                                    if (updated.isActiveOrQueuedWithGeneration()) {
+                                    if (updated.isActiveOrAssignedQueued()) {
                                         markShardReassigned(shardId, reassignedShardIds);
                                     }
                                     updatedAssignmentsBuilder.put(sid, updated);
@@ -3151,7 +3151,7 @@ private static void maybeStartQueuedWithGenerationShardSnapshots(
         if (perNodeShardSnapshotCounter.hasCapacityOnAnyNode()) {
             for (var shardId : shardsBuilder.keys()) {
                 final var existingShardSnapshotStatus = shardsBuilder.get(shardId);
-                if (existingShardSnapshotStatus.isQueuedWithGeneration() == false) {
+                if (existingShardSnapshotStatus.isAssignedQueued() == false) {
                     continue;
                 }
                 final IndexRoutingTable indexRouting = clusterState.routingTable(entry.projectId()).index(shardId.getIndex());
@@ -3336,7 +3336,7 @@ private static ShardSnapshotStatus initShardSnapshotStatus(
             if (perNodeShardSnapshotCounter.tryStartShardSnapshotOnNode(primary.currentNodeId())) {
                 shardSnapshotStatus = new ShardSnapshotStatus(primary.currentNodeId(), shardRepoGeneration);
             } else {
-                shardSnapshotStatus = ShardSnapshotStatus.queuedWithGeneration(shardRepoGeneration);
+                shardSnapshotStatus = ShardSnapshotStatus.assignedQueued(primary.currentNodeId(), shardRepoGeneration);
             }
         }
         return shardSnapshotStatus;
@@ -3821,7 +3821,7 @@ private <T> void applyShardSnapshotUpdate(
                 }
 
                 if (shardSnapshotStatusUpdate.isClone() == false
-                    && existing.state() == ShardState.INIT
+                    && existing.state() == ShardState.INIT // TODO: this is not right, should consider aborted ones on data nodes as well
                     && updatedShardSnapshotStatus.state() != ShardState.INIT) {
                     perNodeShardSnapshotCounter.completeShardSnapshotOnNode(updatedShardSnapshotStatus.nodeId());
                 }
@@ -3906,7 +3906,7 @@ private void startShardSnapshot(RepositoryShardId repoShardId, ShardGeneration g
                 if (shardSnapshotStatus.isActive()) {
                     startShardOperation(shardsBuilder(), routingShardId, shardSnapshotStatus);
                 } else {
-                    if (shardSnapshotStatus.isQueuedWithGeneration()) {
+                    if (shardSnapshotStatus.isAssignedQueued()) {
                         updatesIterator.remove();
                     }
                     // update to queued snapshot did not result in an actual update execution so we just record it but keep applying