nits

Author: tlrx

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

@@ -21,9 +21,10 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 import java.util.function.Consumer;
 
 import static java.util.Collections.unmodifiableList;
@@ -39,7 +40,6 @@
  * id, the snapshot name, the repository name and the repository id (if it exists) once, along with the time at which the snapshot was added
  * to the pending deletion, in a {@link SnapshotDeletionsPending} entry.
  *
- *
  * When cluster state is updated with such entries the {@link org.elasticsearch.snapshots.SnapshotsService} executes corresponding snapshot
  * delete requests to effectively delete the snapshot from the repository. It is possible that the deletion of a snapshot failed for various
  * reason (ex: conflicting snapshot operation, repository removed etc). In such cases the snapshot pending deletion is kept in the cluster
@@ -51,21 +51,28 @@ public class SnapshotDeletionsPending extends AbstractNamedDiffable<Custom> impl
     public static final SnapshotDeletionsPending EMPTY = new SnapshotDeletionsPending(List.of());
     public static final String TYPE = "snapshot_deletions_pending";
 
+    /**
+     * Version from which a snapshot can be marked as to be deleted after an index is deleted.
+     */
+    public static final Version SNAPSHOT_DELETIONS_PENDING_VERSION = Version.V_8_1_0;
+
     /**
      * Setting for the maximum number of snapshots pending deletion allowed in the cluster state.
      * <p>
      * This setting is here to prevent the cluster to grow too large. In the case that the number of snapshots pending deletion exceeds
      * the value of this setting the oldest entries are removed from the cluster state. Snapshots that are discarded are removed before
      * they can be deleted from their repository and are therefore considered as "leaking" and should be logged as such as warnings.
+     * <p>
+     * This setting is a non-dynamic, node-level only setting that is only used on the elected master node.
      */
     public static final Setting<Integer> MAX_PENDING_DELETIONS_SETTING = Setting.intSetting(
         "cluster.snapshot.snapshot_deletions_pending.size",
-        500,
+        5_000,
         Setting.Property.NodeScope
     );
 
     /**
-     * A list of snapshots to delete, sorted by creation time
+     * A list of snapshots to delete, in the order deletions were requested.
      */
     private final List<Entry> entries;
 
@@ -111,7 +118,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
 
     @Override
     public Version getMinimalSupportedVersion() {
-        return Version.CURRENT.minimumCompatibilityVersion();
+        return SNAPSHOT_DELETIONS_PENDING_VERSION;
     }
 
     public SnapshotDeletionsPending withRemovedSnapshots(List<SnapshotId> snapshotIds) {
@@ -120,8 +127,9 @@ public SnapshotDeletionsPending withRemovedSnapshots(List<SnapshotId> snapshotId
         }
         boolean changed = false;
         final List<Entry> updatedEntries = new ArrayList<>();
+        final Set<SnapshotId> removedSnapshotIds = new HashSet<>(snapshotIds);
         for (Entry entry : entries) {
-            if (snapshotIds.contains(entry.snapshotId)) {
+            if (removedSnapshotIds.contains(entry.snapshotId)) {
                 changed = true;
                 continue;
             }
@@ -138,46 +146,35 @@ public SnapshotDeletionsPending withRemovedSnapshots(List<SnapshotId> snapshotId
 
     @Override
     public String toString() {
-        final StringBuilder builder = new StringBuilder("SnapshotDeletionsPending[");
-        boolean prepend = true;
-        final Iterator<Entry> iterator = entries.stream().iterator();
-        while (iterator.hasNext()) {
-            if (prepend == false) {
-                builder.append(',');
-            }
-            builder.append(iterator.next());
-            prepend = false;
-        }
-        builder.append(']');
-        return builder.toString();
+        return "SnapshotDeletionsPending[" + entries + ']';
     }
 
     public static class Entry implements Writeable, ToXContentObject {
 
         private final String repositoryName;
         private final String repositoryUuid;
         private final SnapshotId snapshotId;
-        private final long creationTime;
+        private final long indexDeletionTime;
 
-        public Entry(String repositoryName, String repositoryUuid, SnapshotId snapshotId, long creationTime) {
+        public Entry(String repositoryName, String repositoryUuid, SnapshotId snapshotId, long indexDeletionTime) {
             this.repositoryName = Objects.requireNonNull(repositoryName);
             this.repositoryUuid = Objects.requireNonNull(repositoryUuid);
             this.snapshotId = Objects.requireNonNull(snapshotId);
-            this.creationTime = creationTime;
+            this.indexDeletionTime = indexDeletionTime;
         }
 
         private Entry(StreamInput in) throws IOException {
             this.repositoryName = in.readString();
             this.repositoryUuid = in.readString();
-            this.creationTime = in.readVLong();
+            this.indexDeletionTime = in.readVLong();
             this.snapshotId = new SnapshotId(in);
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeString(repositoryName);
             out.writeString(repositoryUuid);
-            out.writeVLong(creationTime);
+            out.writeVLong(indexDeletionTime);
             snapshotId.writeTo(out);
         }
 
@@ -193,24 +190,24 @@ public SnapshotId getSnapshotId() {
             return snapshotId;
         }
 
-        public long getCreationTime() {
-            return creationTime;
+        public long getIndexDeletionTime() {
+            return indexDeletionTime;
         }
 
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Entry entry = (Entry) o;
-            return creationTime == entry.creationTime
+            return indexDeletionTime == entry.indexDeletionTime
                 && Objects.equals(repositoryName, entry.repositoryName)
                 && Objects.equals(repositoryUuid, entry.repositoryUuid)
                 && Objects.equals(snapshotId, entry.snapshotId);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(repositoryName, repositoryUuid, snapshotId, creationTime);
+            return Objects.hash(repositoryName, repositoryUuid, snapshotId, indexDeletionTime);
         }
 
         @Override
@@ -219,7 +216,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             {
                 builder.field("repository_name", repositoryName);
                 builder.field("repository_uuid", repositoryUuid);
-                builder.timeField("creation_time_millis", "creation_time", creationTime);
+                builder.timeField("creation_time_millis", "creation_time", indexDeletionTime);
                 builder.field("snapshot", snapshotId);
             }
             builder.endObject();
@@ -228,7 +225,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
 
         @Override
         public String toString() {
-            return '[' + repositoryName + '/' + repositoryUuid + ',' + snapshotId + ',' + creationTime + ']';
+            return '[' + repositoryName + '/' + repositoryUuid + ',' + snapshotId + ',' + indexDeletionTime + ']';
         }
     }
 

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexService.java

@@ -237,7 +237,7 @@ private SnapshotDeletionsPending updateSnapshotDeletionsPending(
                                     + "cluster state before snapshot [{}] deleted on [{}] in repository [{}/{}] could be deleted",
                                 maxPendingDeletions,
                                 evicted.getSnapshotId(),
-                                Instant.ofEpochMilli(evicted.getCreationTime()).atZone(ZoneOffset.UTC),
+                                Instant.ofEpochMilli(evicted.getIndexDeletionTime()).atZone(ZoneOffset.UTC),
                                 evicted.getRepositoryName(),
                                 evicted.getRepositoryUuid()
                             )

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

@@ -118,6 +118,7 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 import static java.util.Collections.unmodifiableList;
 import static org.elasticsearch.cluster.SnapshotsInProgress.completed;
@@ -575,32 +576,26 @@ private static void ensureSnapshotNameAvailableInRepo(RepositoryData repositoryD
         }
     }
 
-    private static Set<SnapshotId> listOfCloneSources(final ClusterState state) {
+    private static Set<SnapshotId> cloneSources(final ClusterState state) {
         return state.custom(SnapshotsInProgress.TYPE, SnapshotsInProgress.EMPTY)
             .asStream()
             .filter(SnapshotsInProgress.Entry::isClone)
             .map(SnapshotsInProgress.Entry::source)
-            .collect(Collectors.toSet());
+            .collect(Collectors.toUnmodifiableSet());
     }
 
-    private static Set<SnapshotId> listOfRestoreSources(final ClusterState state) {
-        final Set<SnapshotId> snapshotIds = new HashSet<>();
-        for (RestoreInProgress.Entry restore : state.custom(RestoreInProgress.TYPE, RestoreInProgress.EMPTY)) {
-            snapshotIds.add(restore.snapshot().getSnapshotId());
-        }
-        return Set.copyOf(snapshotIds);
+    private static Set<SnapshotId> restoreSources(final ClusterState state) {
+        return StreamSupport.stream(state.custom(RestoreInProgress.TYPE, RestoreInProgress.EMPTY).spliterator(), false)
+            .map(restore -> restore.snapshot().getSnapshotId())
+            .collect(Collectors.toUnmodifiableSet());
     }
 
-    private static Set<SnapshotId> listOfDeletionsSources(final ClusterState state) {
-        final SnapshotDeletionsInProgress deletionsInProgress = state.custom(SnapshotDeletionsInProgress.TYPE);
-        if (deletionsInProgress == null) {
-            return Set.of();
-        }
-        final Set<SnapshotId> snapshotIds = new HashSet<>();
-        for (SnapshotDeletionsInProgress.Entry deletion : deletionsInProgress.getEntries()) {
-            snapshotIds.addAll(deletion.getSnapshots());
-        }
-        return Set.copyOf(snapshotIds);
+    private static Set<SnapshotId> deletionsSources(final ClusterState state) {
+        return state.custom(SnapshotDeletionsInProgress.TYPE, SnapshotDeletionsInProgress.EMPTY)
+            .getEntries()
+            .stream()
+            .flatMap(deletion -> deletion.getSnapshots().stream())
+            .collect(Collectors.toUnmodifiableSet());
     }
 
     /**
@@ -1324,9 +1319,9 @@ private void triggerSnapshotsPendingDeletions(final ClusterState state) {
             || state.custom(RepositoryCleanupInProgress.TYPE, RepositoryCleanupInProgress.EMPTY).hasCleanupInProgress()) {
             return;
         }
-        final Set<SnapshotId> currentDeletions = listOfDeletionsSources(state);
-        final Set<SnapshotId> currentRestores = listOfRestoreSources(state);
-        final Set<SnapshotId> currentClones = listOfCloneSources(state);
+        final Set<SnapshotId> currentDeletions = deletionsSources(state);
+        final Set<SnapshotId> currentRestores = restoreSources(state);
+        final Set<SnapshotId> currentClones = cloneSources(state);
 
         // the list of snapshot ids to trigger deletion for, per repository
         final Map<RepositoryMetadata, Set<SnapshotId>> snapshotsToDelete = new HashMap<>();
@@ -1378,7 +1373,7 @@ private void triggerSnapshotsPendingDeletions(final ClusterState state) {
                     snapshot.getRepositoryName(),
                     snapshot.getRepositoryUuid(),
                     snapshotId,
-                    Instant.ofEpochMilli(snapshot.getCreationTime()).atZone(ZoneOffset.UTC)
+                    Instant.ofEpochMilli(snapshot.getIndexDeletionTime()).atZone(ZoneOffset.UTC)
                 );
                 continue;
             }
@@ -1390,7 +1385,7 @@ private void triggerSnapshotsPendingDeletions(final ClusterState state) {
                     repository.name(),
                     repository.uuid(),
                     snapshotId,
-                    Instant.ofEpochMilli(snapshot.getCreationTime()).atZone(ZoneOffset.UTC)
+                    Instant.ofEpochMilli(snapshot.getIndexDeletionTime()).atZone(ZoneOffset.UTC)
                 );
                 continue;
             }
@@ -2441,7 +2436,7 @@ public ClusterState execute(ClusterState currentState) {
                     return currentState;
                 }
 
-                final Set<SnapshotId> activeCloneSources = listOfCloneSources(currentState);
+                final Set<SnapshotId> activeCloneSources = cloneSources(currentState);
                 for (SnapshotId snapshotId : snapshotIds) {
                     if (activeCloneSources.contains(snapshotId)) {
                         throw new ConcurrentSnapshotExecutionException(