Remove onItemCompletion param from ThrottledIterator (#113919)

This param is a no-op for all production callers but one, and it's easy
to adapt the one exception to its removal, so it's effectively useless.
This commit removes it.

Author: DaveCTurner

server/src/internalClusterTest/java/org/elasticsearch/rest/ChunkedZipResponseIT.java

@@ -269,7 +269,6 @@ private static void handleZipRestRequest(
                         )
                     ),
                     between(1, 10),
-                    () -> {},
                     Releasables.wrap(refs.acquire(), chunkedZipResponse)::close
                 );
             }

server/src/internalClusterTest/java/org/elasticsearch/rest/StreamingXContentResponseIT.java

@@ -192,7 +192,6 @@ private static void handleStreamingXContentRestRequest(
                         })
                     ),
                     between(1, 10),
-                    () -> {},
                     () -> {
                         try (streamingXContentResponse; finalRef) {
                             streamingXContentResponse.writeFragment(p -> ChunkedToXContentHelper.endObject(), refs.acquire());

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

@@ -322,7 +322,6 @@ public void onFailure(Exception e) {
                                 })
                             ),
                             getSnapshotInfoExecutor.getMaxRunningTasks(),
-                            () -> {},
                             () -> {}
                         );
                     }));

server/src/main/java/org/elasticsearch/action/admin/indices/shards/TransportIndicesShardStoresAction.java

@@ -191,7 +191,6 @@ void run() {
                 Iterators.flatMap(Iterators.forArray(concreteIndices), this::getIndexIterator),
                 this::doShardRequest,
                 maxConcurrentShardRequests,
-                () -> {},
                 outerListener::close
             );
         }

server/src/main/java/org/elasticsearch/common/util/concurrent/ThrottledIterator.java

@@ -38,18 +38,10 @@ public class ThrottledIterator<T> implements Releasable {
      *                     background task previously called {@link RefCounted#decRef()} on its ref count. This operation should not throw
      *                     any exceptions.
      * @param maxConcurrency The maximum number of ongoing operations at any time.
-     * @param onItemCompletion Executed when each item is completed, which can be used for instance to report on progress. Must not throw
-     *                         exceptions.
-     * @param onCompletion     Executed when all items are completed.
+     * @param onCompletion   Executed when all items are completed.
      */
-    public static <T> void run(
-        Iterator<T> iterator,
-        BiConsumer<Releasable, T> itemConsumer,
-        int maxConcurrency,
-        Runnable onItemCompletion,
-        Runnable onCompletion
-    ) {
-        try (var throttledIterator = new ThrottledIterator<>(iterator, itemConsumer, maxConcurrency, onItemCompletion, onCompletion)) {
+    public static <T> void run(Iterator<T> iterator, BiConsumer<Releasable, T> itemConsumer, int maxConcurrency, Runnable onCompletion) {
+        try (var throttledIterator = new ThrottledIterator<>(iterator, itemConsumer, maxConcurrency, onCompletion)) {
             throttledIterator.run();
         }
     }
@@ -58,22 +50,14 @@ public static <T> void run(
     private final Iterator<T> iterator;
     private final BiConsumer<Releasable, T> itemConsumer;
     private final Semaphore permits;
-    private final Runnable onItemCompletion;
-
-    private ThrottledIterator(
-        Iterator<T> iterator,
-        BiConsumer<Releasable, T> itemConsumer,
-        int maxConcurrency,
-        Runnable onItemCompletion,
-        Runnable onCompletion
-    ) {
+
+    private ThrottledIterator(Iterator<T> iterator, BiConsumer<Releasable, T> itemConsumer, int maxConcurrency, Runnable onCompletion) {
         this.iterator = Objects.requireNonNull(iterator);
         this.itemConsumer = Objects.requireNonNull(itemConsumer);
         if (maxConcurrency <= 0) {
             throw new IllegalArgumentException("maxConcurrency must be positive");
         }
         this.permits = new Semaphore(maxConcurrency);
-        this.onItemCompletion = Objects.requireNonNull(onItemCompletion);
         this.refs = AbstractRefCounted.of(onCompletion);
     }
 
@@ -114,26 +98,19 @@ private class ItemRefCounted extends AbstractRefCounted implements Releasable {
 
         @Override
         protected void closeInternal() {
+            permits.release();
             try {
-                onItemCompletion.run();
-            } catch (Exception e) {
-                logger.error("exception in onItemCompletion", e);
-                assert false : e;
-            } finally {
-                permits.release();
-                try {
-                    // Someone must now pick up the next item. Here we might be called from the run() invocation which started processing
-                    // the just-completed item (via close() -> decRef()) if that item's processing didn't fork or all its forked tasks
-                    // finished first. If so, there's no need to call run() here, we can just return and the next iteration of the run()
-                    // loop will continue the processing; moreover calling run() in this situation could lead to a stack overflow. However
-                    // if we're not within that run() invocation then ...
-                    if (isRecursive() == false) {
-                        // ... we're not within any other run() invocation either, so it's safe (and necessary) to call run() here.
-                        run();
-                    }
-                } finally {
-                    refs.decRef();
+                // Someone must now pick up the next item. Here we might be called from the run() invocation which started processing the
+                // just-completed item (via close() -> decRef()) if that item's processing didn't fork or all its forked tasks finished
+                // first. If so, there's no need to call run() here, we can just return and the next iteration of the run() loop will
+                // continue the processing; moreover calling run() in this situation could lead to a stack overflow. However if we're not
+                // within that run() invocation then ...
+                if (isRecursive() == false) {
+                    // ... we're not within any other run() invocation either, so it's safe (and necessary) to call run() here.
+                    run();
                 }
+            } finally {
+                refs.decRef();
             }
         }
 

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

@@ -1212,7 +1212,6 @@ private void writeUpdatedShardMetadataAndComputeDeletes(ActionListener<Void> lis
                     l -> new IndexSnapshotsDeletion(indexId).run(l)
                 ),
                 threadPool.info(ThreadPool.Names.SNAPSHOT).getMax(),
-                () -> {},
                 listeners::close
             );
         }

server/src/test/java/org/elasticsearch/common/util/concurrent/ThrottledIteratorTests.java

@@ -63,7 +63,10 @@ public void testConcurrency() throws InterruptedException {
             final var blockPermits = new Semaphore(between(0, Math.min(maxRelaxedThreads, maxConcurrency) - 1));
 
             ThrottledIterator.run(IntStream.range(0, items).boxed().iterator(), (releasable, item) -> {
-                try (var refs = new RefCountingRunnable(releasable::close)) {
+                try (var refs = new RefCountingRunnable(() -> {
+                    completedItems.incrementAndGet();
+                    releasable.close();
+                })) {
                     assertTrue(itemPermits.tryAcquire());
                     if (forkSupplier.getAsBoolean()) {
                         var ref = refs.acquire();
@@ -108,7 +111,7 @@ public void onFailure(Exception e) {
                         itemPermits.release();
                     }
                 }
-            }, maxConcurrency, completedItems::incrementAndGet, completionLatch::countDown);
+            }, maxConcurrency, completionLatch::countDown);
 
             assertTrue(completionLatch.await(30, TimeUnit.SECONDS));
             assertEquals(items, completedItems.get());

x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/analyze/RepositoryAnalyzeAction.java

@@ -538,13 +538,7 @@ smallBlob && rarely(random),
                 queue.add(ref -> runBlobAnalysis(ref, blobAnalyzeRequest, node));
             }
 
-            ThrottledIterator.run(
-                getQueueIterator(),
-                (ref, task) -> task.accept(ref),
-                request.getConcurrency(),
-                () -> {},
-                requestRefs::close
-            );
+            ThrottledIterator.run(getQueueIterator(), (ref, task) -> task.accept(ref), request.getConcurrency(), requestRefs::close);
         }
 
         private boolean rarely(Random random) {

x-pack/plugin/snapshot-repo-test-kit/src/main/java/org/elasticsearch/repositories/blobstore/testkit/integrity/RepositoryIntegrityVerifier.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.common.util.concurrent.ThrottledTaskRunner;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
+import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshots;
@@ -824,7 +825,12 @@ private static <T> void runThrottled(
         AtomicLong progressCounter,
         Releasable onCompletion
     ) {
-        ThrottledIterator.run(iterator, itemConsumer, maxConcurrency, progressCounter::incrementAndGet, onCompletion::close);
+        ThrottledIterator.run(
+            iterator,
+            (ref, item) -> itemConsumer.accept(Releasables.wrap(progressCounter::incrementAndGet, ref), item),
+            maxConcurrency,
+            onCompletion::close
+        );
     }
 
     private RepositoryVerifyIntegrityResponseChunk.Builder anomaly(String anomaly) {