5454import java .util .Map ;
5555import java .util .Set ;
5656import java .util .concurrent .BlockingQueue ;
57- import java .util .concurrent .Executor ;
5857import java .util .concurrent .LinkedBlockingQueue ;
5958import java .util .stream .Collectors ;
6059
@@ -279,7 +278,6 @@ private void snapshots(SnapshotsInProgress snapshotsInProgress,
279278 // put snapshot info downloads into a task queue instead of pushing them all into the queue to not completely monopolize the
280279 // snapshot meta pool for a single request
281280 final int workers = Math .min (threadPool .info (ThreadPool .Names .SNAPSHOT_META ).getMax (), snapshotIdsToIterate .size ());
282- final Executor executor = threadPool .executor (ThreadPool .Names .SNAPSHOT_META );
283281 final BlockingQueue <SnapshotId > queue = new LinkedBlockingQueue <>(snapshotIdsToIterate );
284282 final ActionListener <Void > workerDoneListener = new GroupedActionListener <>(allDoneListener , workers ).delegateResponse ((l , e ) -> {
285283 queue .clear (); // Stop fetching the remaining snapshots once we've failed fetching one since the response is an error response
@@ -300,15 +298,14 @@ private void snapshots(SnapshotsInProgress snapshotsInProgress,
300298 queue ,
301299 snapshotInfos ,
302300 task ,
303- executor ,
304301 workerDoneListener
305302 );
306303 }
307304 }
308305
309306 /**
310307 * Tries to poll a {@link SnapshotId} to load {@link SnapshotInfo} for from the given {@code queue}. If it finds one in the queue,
311- * loads the snapshot info from the repository on the given {@code executor} and adds it to the given {@code snapshotInfos} collect,
308+ * loads the snapshot info from the repository and adds it to the given {@code snapshotInfos} collect,
312309 * then invokes itself again to try and poll another task from the queue.
313310 * If the queue is empty resolves {@code} listener.
314311 */
@@ -317,14 +314,13 @@ private void getOneSnapshotInfo(boolean ignoreUnavailable,
317314 BlockingQueue <SnapshotId > queue ,
318315 Collection <SnapshotInfo > snapshotInfos ,
319316 CancellableTask task ,
320- Executor executor ,
321317 ActionListener <Void > listener ) {
322318 final SnapshotId snapshotId = queue .poll ();
323319 if (snapshotId == null ) {
324320 listener .onResponse (null );
325321 return ;
326322 }
327- executor .execute (() -> {
323+ threadPool . executor ( ThreadPool . Names . SNAPSHOT_META ) .execute (() -> {
328324 if (task .isCancelled ()) {
329325 listener .onFailure (new TaskCancelledException ("task cancelled" ));
330326 return ;
@@ -342,7 +338,7 @@ private void getOneSnapshotInfo(boolean ignoreUnavailable,
342338 );
343339 }
344340 }
345- getOneSnapshotInfo (ignoreUnavailable , repository , queue , snapshotInfos , task , executor , listener );
341+ getOneSnapshotInfo (ignoreUnavailable , repository , queue , snapshotInfos , task , listener );
346342 });
347343 }
348344
0 commit comments