class BatchPagingIterator<Key> implements BatchIterator<Row> {

    private final PagingIterator<Key, Row> pagingIterator;
    private final Function<Key, KillableCompletionStage<? extends Iterable<? extends KeyIterable<Key, Row>>>> fetchMore;
    private final BooleanSupplier isUpstreamExhausted;
    private final Consumer<? super Throwable> closeCallback;

    private Throwable killed;
    private KillableCompletionStage<? extends Iterable<? extends KeyIterable<Key, Row>>> currentlyLoading;
    private Iterator<Row> it;
    private boolean closed = false;
    private Row current;

    public BatchPagingIterator(PagingIterator<Key, Row> pagingIterator,
                               Function<Key, KillableCompletionStage<? extends Iterable<? extends KeyIterable<Key, Row>>>> fetchMore,
                               BooleanSupplier isUpstreamExhausted,
                               Consumer<? super Throwable> closeCallback) {
        this.pagingIterator = pagingIterator;
        this.it = pagingIterator;
        this.fetchMore = fetchMore;
        this.isUpstreamExhausted = isUpstreamExhausted;
        this.closeCallback = closeCallback;
    }

    @Override
    public Row currentElement() {
        return current;
    }

    @Override
    public void moveToStart() {
        raiseIfClosedOrKilled();
        this.it = pagingIterator.repeat().iterator();
        current = null;
    }

    @Override
    public boolean moveNext() {
        raiseIfClosedOrKilled();

        if (it.hasNext()) {
            current = it.next();
            return true;
        }
        current = null;
        return false;
    }

    @Override
    public void close() {<FILL_FUNCTION_BODY>}

    @Override
    public CompletionStage<?> loadNextBatch() throws Exception {
        if (closed) {
            throw new IllegalStateException("BatchIterator already closed");
        }
        if (allLoaded()) {
            throw new IllegalStateException("All data already loaded");
        }
        Throwable err;
        KillableCompletionStage<? extends Iterable<? extends KeyIterable<Key, Row>>> future;
        synchronized (this) {
            err = this.killed;
            if (err == null) {
                currentlyLoading = future = fetchMore.apply(pagingIterator.exhaustedIterable());
            } else {
                future = KillableCompletionStage.failed(err);
            }
        }
        if (err == null) {
            return future.whenComplete(this::onNextPage);
        }
        return future;
    }

    private void onNextPage(Iterable<? extends KeyIterable<Key, Row>> rows, Throwable ex) {
        if (ex == null) {
            pagingIterator.merge(rows);
            if (isUpstreamExhausted.getAsBoolean()) {
                pagingIterator.finish();
            }
        } else {
            killed = ex;
            throw Exceptions.toRuntimeException(ex);
        }
    }

    @Override
    public boolean allLoaded() {
        return isUpstreamExhausted.getAsBoolean();
    }

    @Override
    public boolean hasLazyResultSet() {
        return true;
    }

    private void raiseIfClosedOrKilled() {
        Throwable err;
        synchronized (this) {
            err = killed;
        }
        if (err != null) {
            Exceptions.rethrowUnchecked(err);
        }
        if (closed) {
            throw new IllegalStateException("Iterator is closed");
        }
    }

    @Override
    public void kill(@NotNull Throwable throwable) {
        KillableCompletionStage<? extends Iterable<? extends KeyIterable<Key, Row>>> loading;
        synchronized (this) {
            killed = throwable;
            loading = this.currentlyLoading;
        }
        close();
        if (loading != null) {
            loading.kill(throwable);
        }
    }
}

        if (!closed) {
            closed = true;
            pagingIterator.finish(); // release resource, specially possible ram accounted bytes
            closeCallback.accept(killed);
        }
    
<no_super_class>