Коллектор для разделения потока на куски заданного размера

У меня есть проблема, которую я пытаюсь решить с помощью чего-то, что я совершенно уверен, что не должен делать, но не вижу альтернативы. Мне дают список строк, и я должен разбить его на куски заданного размера. Затем результат должен быть передан какому-либо методу для дальнейшей обработки. Поскольку список может быть огромным, обработка должна выполняться асинхронно.

Мой подход заключается в создании пользовательского коллектора, который берет поток строк и преобразует его в Поток>:

final Stream<List<Long>> chunks = list
                        .stream()
                        .parallel()
                        .collect(MyCollector.toChunks(CHUNK_SIZE)) 
                        .flatMap(p -> doStuff(p))
                        .collect(MyCollector.toChunks(CHUNK_SIZE))
                        .map(...)
                        ...

Код для коллектора:

public final class MyCollector<T, A extends List<List<T>>, R extends Stream<List<T>>> implements Collector<T, A, R> {
private final AtomicInteger index = new AtomicInteger(0);
private final AtomicInteger current = new AtomicInteger(-1);
private final int chunkSize;

private MyCollector(final int chunkSize){
    this.chunkSize = chunkSize;
}

@Override
public Supplier<A> supplier() {
    return () -> (A)new ArrayList<List<T>>();
}

@Override
public BiConsumer<A, T> accumulator() {
    return (A candidate, T acc) -> {
        if (index.getAndIncrement() % chunkSize == 0){
            candidate.add(new ArrayList<>(chunkSize));
            current.incrementAndGet();
        }
        candidate.get(current.get()).add(acc);
    };
}

@Override
public BinaryOperator<A> combiner() {
    return (a1, a2) -> {
        a1.addAll(a2);
        return a1;
    };
}
@Override
public Function<A, R> finisher() {
    return (a) -> (R)a.stream();
}

@Override
public Set<Characteristics> characteristics() {
    return Collections.unmodifiableSet(EnumSet.of(Characteristics.CONCURRENT, Characteristics.UNORDERED));
}

public static <T> MyCollector<T, List<List<T>>, Stream<List<T>>> toChunks(final int chunkSize){
    return new MyCollector<>(chunkSize);
}

}

Это, кажется, работает в большинстве случаев, но иногда я получаю NPE.. Я уверен, что в аккумуляторе нет потокобезопасности, так как при добавлении новых списков в основной список могут вмешиваться два потока. Я не возражаю против того, чтобы кусок имел несколько слишком много или слишком мало элементов.

Я попробовал это вместо текущей функции поставщика:

 return () -> (A)new ArrayList<List<T>>(){{add(new ArrayList<T>());}};

Чтобы убедиться, что список присутствует всегда. Это не работает вообще и приводит к пустым спискам.

Вопросы:

• я почти уверен, что пользовательский сплитератор был бы хорошим решением. Однако это не будет работать для синхронных сценариев. Кроме того, я уверен, что сплитератор называется?
  Я знаю, что у меня вообще не должно быть состояния, но не знаю, как его изменить.

Вопросы:

Является ли этот подход полностью неправильным или каким-то образом исправимым?
• Если я использую Spliterator-могу ли я быть уверен, что он называется, или это решает базовая реализация?
• я почти уверен, что приведения к (A) и (R) в поставщике и финишере не нужны, но IntelliJ жалуется. Может быть, я чего-то не понимаю?

EDIT:

• я добавил еще несколько в клиентский код в качестве предложений с IntStream.диапазон не будет работать, когда прикован.
• я понимаю, что мог бы сделать это по-другому, как предложено в комментарии, но это также немного о стиле и о том, возможно ли это.
• У меня есть параллельная характеристика, потому что я предполагаю, что в противном случае Stream API вернется к синхронной обработке. Решение не является потокобезопасным, как было указано ранее.

Лучший, D