class AsyncCompositeBI<T> extends AbstractCompositeBI<T> {

        private final Executor executor;
        private final IntSupplier availableThreads;

        AsyncCompositeBI(Executor executor, IntSupplier availableThreads, BatchIterator<T>[] iterators) {
            super(iterators);
            this.executor = executor;
            this.availableThreads = availableThreads;
        }

        @Override
        public boolean moveNext() {
            while (idx < iterators.length) {
                BatchIterator<T> iterator = iterators[idx];
                if (iterator.moveNext()) {
                    return true;
                }
                idx++;
            }
            idx = 0;
            return false;
        }

        private int numIteratorsActive() {
            int activeIts = 0;
            for (var it : iterators) {
                if (!it.allLoaded()) {
                    activeIts++;
                }
            }
            return activeIts;
        }

        @Override
        public CompletionStage<?> loadNextBatch() throws Exception {<FILL_FUNCTION_BODY>}
    }

            if (allLoaded()) {
                throw new IllegalStateException("BatchIterator already loaded");
            }
            int activeIts = numIteratorsActive();
            int numThreads = Math.max(1, availableThreads.getAsInt());
            final int usedThreads = Math.min(numThreads, activeIts);
            ArrayList<CompletableFuture<CompletableFuture<?>>> nestedFutures = new ArrayList<>(usedThreads);
            for (int t = 0; t < usedThreads; t++) {
                final int thread = t;
                nestedFutures.add(supplyAsync(() -> {
                    ArrayList<CompletableFuture<?>> futures = new ArrayList<>();
                    for (int i = 0; i < iterators.length; i++) {
                        var it = iterators[i];
                        if (it.allLoaded() || i % usedThreads != thread) {
                            continue;
                        }
                        try {
                            futures.add(it.loadNextBatch().toCompletableFuture());
                        } catch (Exception e) {
                            throw Exceptions.toRuntimeException(e);
                        }
                    }
                    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
                }, executor));
            }
            return CompletableFutures.allAsList(nestedFutures)
                .thenCompose(innerFutures -> CompletableFuture.allOf(innerFutures.toArray(new CompletableFuture[0])));
        
<no_super_class>