Removes completed snapshot from cluster state on master change (#24605)

Previously, if a master node updated the cluster state to reflect that a
snapshot is completed, but subsequently failed before processing a
cluster state to remove the snapshot from the cluster state, then the
newly elected master would not know that it needed to clean up the
leftover cluster state.

This commit ensures that the newly elected master sees if there is a
snapshot in the cluster state that is in the completed state but has not
yet been removed from the cluster state.

Closes #24452

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

@@ -634,6 +634,7 @@ public void applyClusterState(ClusterChangedEvent event) {
                 if (event.routingTableChanged()) {
                     processStartedShards(event);
                 }
+                removeFinishedSnapshotFromClusterState(event);
                 finalizeSnapshotDeletionFromPreviousMaster(event);
             }
         } catch (Exception e) {
@@ -663,6 +664,26 @@ private void finalizeSnapshotDeletionFromPreviousMaster(ClusterChangedEvent even
         }
     }
 
+    /**
+     * Removes a finished snapshot from the cluster state.  This can happen if the previous
+     * master node processed a cluster state update that marked the snapshot as finished,
+     * but the previous master node died before removing the snapshot in progress from the
+     * cluster state.  It is then the responsibility of the new master node to end the
+     * snapshot and remove it from the cluster state.
+     */
+    private void removeFinishedSnapshotFromClusterState(ClusterChangedEvent event) {
+        if (event.localNodeMaster() && !event.previousState().nodes().isLocalNodeElectedMaster()) {
+            SnapshotsInProgress snapshotsInProgress = event.state().custom(SnapshotsInProgress.TYPE);
+            if (snapshotsInProgress != null && !snapshotsInProgress.entries().isEmpty()) {
+                for (SnapshotsInProgress.Entry entry : snapshotsInProgress.entries()) {
+                    if (entry.state().completed()) {
+                        endSnapshot(entry);
+                    }
+                }
+            }
+        }
+    }
+
     /**
      * Cleans up shard snapshots that were running on removed nodes
      *