53
53
import org .elasticsearch .transport .TransportService ;
54
54
55
55
import java .util .ArrayList ;
56
- import java .util .Collection ;
57
56
import java .util .Collections ;
58
57
import java .util .HashMap ;
59
58
import java .util .HashSet ;
60
59
import java .util .Iterator ;
61
60
import java .util .List ;
62
61
import java .util .Map ;
63
- import java .util .Queue ;
64
62
import java .util .Set ;
65
63
import java .util .concurrent .atomic .AtomicInteger ;
66
64
import java .util .function .BiPredicate ;
@@ -182,19 +180,13 @@ private class GetSnapshotsOperation {
182
180
private final GetSnapshotInfoExecutor getSnapshotInfoExecutor ;
183
181
184
182
// results
185
- private final Queue < List <SnapshotInfo >> allSnapshotInfos = ConcurrentCollections . newQueue ( );
183
+ private final List <SnapshotInfo > allSnapshotInfos = Collections . synchronizedList ( new ArrayList <>() );
186
184
187
185
/**
188
186
* Accumulates number of snapshots that match the name/fromSortValue/slmPolicy predicates, to be returned in the response.
189
187
*/
190
188
private final AtomicInteger totalCount = new AtomicInteger ();
191
189
192
- /**
193
- * Accumulates the number of snapshots that match the name/fromSortValue/slmPolicy/after predicates, for sizing the final result
194
- * list.
195
- */
196
- private final AtomicInteger resultsCount = new AtomicInteger ();
197
-
198
190
GetSnapshotsOperation (
199
191
CancellableTask cancellableTask ,
200
192
List <RepositoryMetadata > repositories ,
@@ -438,18 +430,7 @@ private void loadSnapshotInfos(Iterator<AsyncSnapshotInfo> asyncSnapshotInfoIter
438
430
if (cancellableTask .notifyIfCancelled (listener )) {
439
431
return ;
440
432
}
441
- final var repositoryTotalCount = new AtomicInteger ();
442
-
443
- final List <SnapshotInfo > snapshots = new ArrayList <>();
444
- final List <SnapshotInfo > syncSnapshots = Collections .synchronizedList (snapshots );
445
-
446
433
try (var listeners = new RefCountingListener (listener )) {
447
- final var iterationCompleteListener = listeners .acquire (ignored -> {
448
- totalCount .addAndGet (repositoryTotalCount .get ());
449
- // no need to synchronize access to snapshots: all writes happen-before this read
450
- resultsCount .addAndGet (snapshots .size ());
451
- allSnapshotInfos .add (snapshots );
452
- });
453
434
ThrottledIterator .run (
454
435
Iterators .failFast (asyncSnapshotInfoIterator , () -> cancellableTask .isCancelled () || listeners .isFailing ()),
455
436
(ref , asyncSnapshotInfo ) -> {
@@ -458,9 +439,9 @@ private void loadSnapshotInfos(Iterator<AsyncSnapshotInfo> asyncSnapshotInfoIter
458
439
@ Override
459
440
public void onResponse (SnapshotInfo snapshotInfo ) {
460
441
if (matchesPredicates (snapshotInfo )) {
461
- repositoryTotalCount .incrementAndGet ();
442
+ totalCount .incrementAndGet ();
462
443
if (afterPredicate .test (snapshotInfo )) {
463
- syncSnapshots .add (snapshotInfo .maybeWithoutIndices (indices ));
444
+ allSnapshotInfos .add (snapshotInfo .maybeWithoutIndices (indices ));
464
445
}
465
446
}
466
447
refListener .onResponse (null );
@@ -479,7 +460,7 @@ public void onFailure(Exception e) {
479
460
},
480
461
getSnapshotInfoExecutor .getMaxRunningTasks (),
481
462
() -> {},
482
- () -> iterationCompleteListener . onResponse ( null )
463
+ () -> {}
483
464
);
484
465
}
485
466
}
@@ -489,12 +470,11 @@ private GetSnapshotsResponse buildResponse() {
489
470
cancellableTask .ensureNotCancelled ();
490
471
int remaining = 0 ;
491
472
final var resultsStream = allSnapshotInfos .stream ()
492
- .flatMap (Collection ::stream )
493
473
.peek (this ::assertSatisfiesAllPredicates )
494
474
.sorted (sortBy .getSnapshotInfoComparator (order ))
495
475
.skip (offset );
496
476
final List <SnapshotInfo > snapshotInfos ;
497
- if (size == GetSnapshotsRequest .NO_LIMIT || resultsCount . get () <= size ) {
477
+ if (size == GetSnapshotsRequest .NO_LIMIT || allSnapshotInfos . size () <= size ) {
498
478
snapshotInfos = resultsStream .toList ();
499
479
} else {
500
480
snapshotInfos = new ArrayList <>(size );
0 commit comments