Краткий ответ

Нарушается требование statelessness, то есть безсостоя́тности поведенческого параметра Stream. Результат предиката начинает зависеть не только от текущего элемента, но и от истории обработки, порядка выполнения и межпоточной синхронизации. В параллельном стриме это может привести к гонкам, недетерминированным результатам и зависимости поведения от степени параллелизма.

Исторический контекст

Stream API проектировался как декларативная модель обработки данных: разработчик описывает преобразования, фильтрацию и сбор результата, а библиотека выбирает способ выполнения — последовательный или параллельный. Чтобы такая модель сохраняла предсказуемость и позволяла безопасно распараллеливать операции, функции обработки должны быть независимыми от внешнего изменяемого состояния.

Это связано не с синтаксисом лямбд, а с контрактом поведения операций Stream. Предикат, функция преобразования или другая лямбда должны вычислять результат на основании входного элемента, не используя скрытое состояние предыдущих вызовов.

Постановка проблемы

Если предикат изменяет общий счётчик, коллекцию или другой объект, порядок вызовов становится частью логики. В последовательном стриме такой код иногда выглядит рабочим, но его результат уже зависит от количества и порядка фактически выполненных проверок.

В параллельном стриме несколько потоков могут одновременно обращаться к одному состоянию. Для обычной переменной это создаёт гонку данных: обновления могут теряться, а чтения — не видеть актуальное значение. Даже потокобезопасный объект не делает саму бизнес-логику корректной: он устраняет повреждение структуры данных, но не гарантирует нужный порядок или соответствие элемента конкретному номеру вызова.

Подробное решение

Поведенческий параметр должен быть stateless: одинаковый элемент при одинаковых внешних условиях должен обрабатываться независимо от того, какие элементы обрабатывались до него. Он также должен быть non-interfering — не изменять источник стрима во время его обхода.

Например, атомарный счётчик предотвращает потерю инкрементов, но не превращает логику в корректную фильтрацию по позиции вызова:

import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

AtomicInteger calls = new AtomicInteger();
var result = IntStream.range(0, 1000)
    .parallel()
    .filter(x -> calls.incrementAndGet() % 2 == 0)
    .boxed()
    .toList();

Здесь технически безопасно обновляется calls, но принадлежность элемента к результату зависит от расписания потоков. При изменении источника, режима выполнения или наличия короткого замыкания число вызовов и состав результата могут измениться.

Если состояние действительно необходимо, его следует вынести из функции обработки и использовать явный этап агрегации: Collector, reduce с корректным контрактом или отдельную синхронизированную структуру, если побочный эффект является целью. При этом нужно отдельно проверить ассоциативность, отсутствие зависимости от порядка и корректность параллельного объединения.

Ситуация из практики

Допустим, нужно выбрать каждый десятый элемент большого набора. Один вариант — хранить общий счётчик внутри предиката. Его плюс — внешняя простота, но результат зависит от порядка вызовов и плохо переносится с последовательного стрима на параллельный.

Второй вариант — сначала сформировать упорядоченную модель данных, затем использовать индексы или отдельную операцию группировки. Он может потребовать больше памяти или дополнительного прохода, зато правило выбора становится явным и воспроизводимым.

Выбранное решение должно определяться семантикой задачи. Если важна позиция элемента в общем порядке, нельзя кодировать её через порядок вызовов предиката; нужно работать с явными индексами или последовательной обработкой. Если порядок не важен, допустима параллельная агрегация, но состояние должно принадлежать корректному коллектору, а не общей переменной лямбды.

Что кандидаты часто упускают

Достаточно ли использовать AtomicInteger, чтобы сделать stateful-предикат корректным?

Нет. AtomicInteger делает отдельные операции над счётчиком атомарными, но не гарантирует, какой элемент получит конкретное значение счётчика. Он также не устраняет зависимость результата от порядка планирования и количества реально выполненных вызовов. Потокобезопасность контейнера и корректность алгоритма — разные свойства.

Можно ли считать последовательный стрим безопасным для изменяемого состояния внутри лямбды?

Не всегда. В последовательном режиме гонки между потоками обычно нет, но нарушается безсостоя́тность функции, а результат зависит от порядка обхода и побочных эффектов. Кроме того, изменение источника во время обхода может нарушить требование невмешательства и вызвать исключение либо некорректное поведение.

Почему побочный эффект в лямбде может выполниться не для всех элементов?

Терминальная операция может использовать короткое замыкание, например при поиске подходящего элемента, а параллельный стрим может продолжать или отменять уже запущенные задачи в зависимости от реализации и состояния вычисления. Поэтому лямбду Stream нельзя использовать как надёжный механизм учёта каждого элемента, если это не гарантируется контрактом конкретной операции. Для обязательной обработки всех элементов нужен явно выбранный этап с подходящей семантикой, а не случайный побочный эффект внутри промежуточной операции.