Snapshot update batching code renames and comments (elastic#127356)

A collection of miscellaneous changes I made while working on
understanding the code.

Author: DiannaHohensee

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

@@ -135,14 +135,16 @@ private SnapshotsInProgress(Map<ProjectRepo, ByRepo> entries, Set<String> nodesI
 
     @FixForMultiProject
     @Deprecated(forRemoval = true)
-    public SnapshotsInProgress withUpdatedEntriesForRepo(String repository, List<Entry> updatedEntries) {
-        return withUpdatedEntriesForRepo(Metadata.DEFAULT_PROJECT_ID, repository, updatedEntries);
+    public SnapshotsInProgress createCopyWithUpdatedEntriesForRepo(String repository, List<Entry> updatedEntries) {
+        return createCopyWithUpdatedEntriesForRepo(Metadata.DEFAULT_PROJECT_ID, repository, updatedEntries);
     }
 
-    public SnapshotsInProgress withUpdatedEntriesForRepo(ProjectId projectId, String repository, List<Entry> updatedEntries) {
+    public SnapshotsInProgress createCopyWithUpdatedEntriesForRepo(ProjectId projectId, String repository, List<Entry> updatedEntries) {
         if (updatedEntries.equals(forRepo(projectId, repository))) {
+            // No changes to apply, return the current object.
             return this;
         }
+
         final Map<ProjectRepo, ByRepo> copy = new HashMap<>(this.entries);
         final var projectRepo = new ProjectRepo(projectId, repository);
         if (updatedEntries.isEmpty()) {
@@ -153,13 +155,14 @@ public SnapshotsInProgress withUpdatedEntriesForRepo(ProjectId projectId, String
         } else {
             copy.put(projectRepo, new ByRepo(updatedEntries));
         }
+
         return new SnapshotsInProgress(copy, nodesIdsForRemoval);
     }
 
     public SnapshotsInProgress withAddedEntry(Entry entry) {
         final List<Entry> forRepo = new ArrayList<>(forRepo(entry.projectId(), entry.repository()));
         forRepo.add(entry);
-        return withUpdatedEntriesForRepo(entry.projectId(), entry.repository(), forRepo);
+        return createCopyWithUpdatedEntriesForRepo(entry.projectId(), entry.repository(), forRepo);
     }
 
     /**

server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.ShardRoutingState;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.FailedShard;
 import org.elasticsearch.cluster.routing.allocation.StaleShard;
@@ -634,10 +635,11 @@ public ClusterState execute(BatchExecutionContext<StartedShardUpdateTask> batchE
             List<ShardRouting> shardRoutingsToBeApplied = new ArrayList<>(batchExecutionContext.taskContexts().size());
             Set<ShardRouting> seenShardRoutings = new HashSet<>(); // to prevent duplicates
             final Map<Index, ClusterStateTimeRanges> updatedTimestampRanges = new HashMap<>();
+
             final ClusterState initialState = batchExecutionContext.initialState();
             for (var taskContext : batchExecutionContext.taskContexts()) {
-                final var task = taskContext.getTask();
-                final StartedShardEntry startedShardEntry = task.getEntry();
+                final ShardStateAction.StartedShardUpdateTask task = taskContext.getTask();
+                final StartedShardEntry startedShardEntry = task.getStartedShardEntry();
                 final Optional<ProjectMetadata> project = initialState.metadata().lookupProject(startedShardEntry.shardId.getIndex());
                 final ShardRouting matched = project.map(ProjectMetadata::id)
                     .map(id -> initialState.routingTable(id).getByAllocationId(startedShardEntry.shardId, startedShardEntry.allocationId))
@@ -917,9 +919,16 @@ public int hashCode() {
         }
     }
 
+    /**
+     * Task that runs on the master node. Handles responding to the request listener with the result of the update request.
+     * Task is created when the master node receives a data node request to mark a shard as {@link ShardRoutingState#STARTED}.
+     *
+     * @param entry Information about the newly sharted shard.
+     * @param listener Channel listener with which to respond to the data node.
+     */
     public record StartedShardUpdateTask(StartedShardEntry entry, ActionListener<Void> listener) implements ClusterStateTaskListener {
 
-        public StartedShardEntry getEntry() {
+        public StartedShardEntry getStartedShardEntry() {
             return entry;
         }
 

server/src/main/java/org/elasticsearch/cluster/service/MasterService.java

@@ -1437,7 +1437,7 @@ private interface Batch {
      *
      * @param name The name of the queue, which is mostly useful for debugging.
      *
-     * @param priority The priority at which tasks submitted to the queue are executed. Avoid priorites other than {@link Priority#NORMAL}
+     * @param priority The priority at which tasks submitted to the queue are executed. Avoid priorities other than {@link Priority#NORMAL}
      *                 where possible. A stream of higher-priority tasks can starve lower-priority ones from running. Higher-priority tasks
      *                 should definitely re-use the same {@link MasterServiceTaskQueue} so that they are executed in batches.
      *

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -3076,7 +3076,9 @@ private ClusterState updateRepositoryGenerationsIfNecessary(ClusterState state,
                 snapshotEntries.add(entry);
             }
         }
-        updatedSnapshotsInProgress = changedSnapshots ? snapshotsInProgress.withUpdatedEntriesForRepo(repoName, snapshotEntries) : null;
+        updatedSnapshotsInProgress = changedSnapshots
+            ? snapshotsInProgress.createCopyWithUpdatedEntriesForRepo(repoName, snapshotEntries)
+            : null;
         final SnapshotDeletionsInProgress updatedDeletionsInProgress;
         boolean changedDeletions = false;
         final List<SnapshotDeletionsInProgress.Entry> deletionEntries = new ArrayList<>();

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

@@ -196,7 +196,7 @@ public static class Builder {
          *                     If the request is from SLM it will contain a key "policy" with an SLM policy name as the value.
          * @param snapshotId the snapshotId to potentially add to the registered set
          */
-        void maybeAdd(Map<String, Object> userMetadata, SnapshotId snapshotId) {
+        void addIfSnapshotIsSLMInitiated(Map<String, Object> userMetadata, SnapshotId snapshotId) {
             final String policy = getPolicyFromMetadata(userMetadata);
             if (policy != null) {
                 snapshots.add(new PolicySnapshot(policy, snapshotId));

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

@@ -436,19 +436,23 @@ public void onFailure(Exception e) {
         startShardSnapshotTaskRunner.runSyncTasksEagerly(threadPool.executor(ThreadPool.Names.SNAPSHOT));
     }
 
-    private void pauseShardSnapshotsForNodeRemoval(String localNodeId, SnapshotsInProgress.Entry entry) {
-        final var localShardSnapshots = shardSnapshots.getOrDefault(entry.snapshot(), Map.of());
+    /**
+     * Iterates all the shard snapshots for the given snapshot entry. Any shard snapshots assigned to this data node will be: immediately
+     * paused, if not already running; or flagged for pausing, if running.
+     */
+    private void pauseShardSnapshotsForNodeRemoval(String localNodeId, SnapshotsInProgress.Entry masterEntryCopy) {
+        final var localShardSnapshots = shardSnapshots.getOrDefault(masterEntryCopy.snapshot(), Map.of());
 
-        for (final Map.Entry<ShardId, ShardSnapshotStatus> shardEntry : entry.shards().entrySet()) {
+        for (final Map.Entry<ShardId, ShardSnapshotStatus> shardEntry : masterEntryCopy.shards().entrySet()) {
             final ShardId shardId = shardEntry.getKey();
-            final ShardSnapshotStatus masterShardSnapshotStatus = shardEntry.getValue();
+            final ShardSnapshotStatus masterShardSnapshotStatusCopy = shardEntry.getValue();
 
-            if (masterShardSnapshotStatus.state() != ShardState.INIT) {
+            if (masterShardSnapshotStatusCopy.state() != ShardState.INIT) {
                 // shard snapshot not currently scheduled by master
                 continue;
             }
 
-            if (localNodeId.equals(masterShardSnapshotStatus.nodeId()) == false) {
+            if (localNodeId.equals(masterShardSnapshotStatusCopy.nodeId()) == false) {
                 // shard snapshot scheduled on a different node
                 continue;
             }
@@ -457,11 +461,11 @@ private void pauseShardSnapshotsForNodeRemoval(String localNodeId, SnapshotsInPr
             if (localShardSnapshotStatus == null) {
                 // shard snapshot scheduled but not currently running, pause immediately without starting
                 notifyUnsuccessfulSnapshotShard(
-                    entry.snapshot(),
+                    masterEntryCopy.snapshot(),
                     shardId,
                     ShardState.PAUSED_FOR_NODE_REMOVAL,
                     "paused",
-                    masterShardSnapshotStatus.generation(),
+                    masterShardSnapshotStatusCopy.generation(),
                     // Shard snapshot never began, so there is no status object to update
                     (outcomeInfoString) -> {}
                 );