Reduce copying in GetSnapshotsOperation#snapshots() (elastic#105765)

No need to create a set (the values are distinct anyway), make a
separate synchronized list, populate them both, and finally copy them
both again into another concatenated list. We can just make the final
list up front.

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/TransportGetSnapshotsAction.java

@@ -400,7 +400,7 @@ private void snapshots(String repositoryName, Collection<SnapshotId> snapshotIds
             if (cancellableTask.notifyIfCancelled(listener)) {
                 return;
             }
-            final Set<SnapshotInfo> snapshotSet = new HashSet<>();
+            final List<SnapshotInfo> snapshots = new ArrayList<>(snapshotIds.size());
             final Set<SnapshotId> snapshotIdsToIterate = new HashSet<>(snapshotIds);
             // first, look at the snapshots in progress
             final List<SnapshotsInProgress.Entry> entries = SnapshotsService.currentSnapshots(
@@ -412,46 +412,46 @@ private void snapshots(String repositoryName, Collection<SnapshotId> snapshotIds
                 if (snapshotIdsToIterate.remove(entry.snapshot().getSnapshotId())) {
                     final SnapshotInfo snapshotInfo = SnapshotInfo.inProgress(entry);
                     if (predicates.test(snapshotInfo)) {
-                        snapshotSet.add(snapshotInfo.maybeWithoutIndices(indices));
+                        snapshots.add(snapshotInfo.maybeWithoutIndices(indices));
                     }
                 }
             }
             // then, look in the repository if there's any matching snapshots left
-            final List<SnapshotInfo> snapshotInfos;
-            if (snapshotIdsToIterate.isEmpty()) {
-                snapshotInfos = Collections.emptyList();
-            } else {
-                snapshotInfos = Collections.synchronizedList(new ArrayList<>());
-            }
-            final ActionListener<Void> allDoneListener = listener.safeMap(v -> {
-                final ArrayList<SnapshotInfo> snapshotList = new ArrayList<>(snapshotInfos);
-                snapshotList.addAll(snapshotSet);
-                return sortSnapshotsWithNoOffsetOrLimit(snapshotList);
-            });
-            if (snapshotIdsToIterate.isEmpty()) {
-                allDoneListener.onResponse(null);
-                return;
-            }
-            final Repository repository;
-            try {
-                repository = repositoriesService.repository(repositoryName);
-            } catch (RepositoryMissingException e) {
-                listener.onFailure(e);
-                return;
-            }
-            repository.getSnapshotInfo(
-                new GetSnapshotInfoContext(
-                    snapshotIdsToIterate,
-                    ignoreUnavailable == false,
-                    cancellableTask::isCancelled,
-                    (context, snapshotInfo) -> {
-                        if (predicates.test(snapshotInfo)) {
-                            snapshotInfos.add(snapshotInfo.maybeWithoutIndices(indices));
-                        }
-                    },
-                    allDoneListener
+            try (
+                var listeners = new RefCountingListener(
+                    // no need to synchronize access to snapshots: Repository#getSnapshotInfo fails fast but we're on the success path here
+                    listener.safeMap(v -> sortSnapshotsWithNoOffsetOrLimit(snapshots))
                 )
-            );
+            ) {
+                if (snapshotIdsToIterate.isEmpty()) {
+                    return;
+                }
+
+                final Repository repository;
+                try {
+                    repository = repositoriesService.repository(repositoryName);
+                } catch (RepositoryMissingException e) {
+                    listeners.acquire().onFailure(e);
+                    return;
+                }
+
+                // only need to synchronize accesses related to reading SnapshotInfo from the repo
+                final List<SnapshotInfo> syncSnapshots = Collections.synchronizedList(snapshots);
+
+                repository.getSnapshotInfo(
+                    new GetSnapshotInfoContext(
+                        snapshotIdsToIterate,
+                        ignoreUnavailable == false,
+                        cancellableTask::isCancelled,
+                        (context, snapshotInfo) -> {
+                            if (predicates.test(snapshotInfo)) {
+                                syncSnapshots.add(snapshotInfo.maybeWithoutIndices(indices));
+                            }
+                        },
+                        listeners.acquire()
+                    )
+                );
+            }
         }
 
         private boolean isCurrentSnapshotsOnly() {