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

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

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

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

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

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

Стрим обычно читает элементы из исходной коллекции во время выполнения терминальной операции. Если функция filter, map, peek или терминальная операция одновременно добавляет, удаляет или изменяет элементы этой коллекции, структура источника меняется прямо во время обхода.

Для обычных коллекций это может привести к ConcurrentModificationException. Однако отсутствие исключения не означает корректность: часть элементов может быть пропущена, обработана повторно или попасть в результат в неожиданном виде.

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

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

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

import java.util.*;

List<Integer> source = new ArrayList<>();
source.add(1);
source.add(2);
source.add(3);

source.stream().forEach(source::remove);

В этом примере forEach читает source, а remove одновременно изменяет её структуру. Для ArrayList обычно возникает ConcurrentModificationException, но полагаться только на исключение нельзя: общий контракт уже нарушен.

Коллекции из пакета java.util.concurrent могут предоставлять слабосогласованный обход и не обязаны выбрасывать такое исключение. Это лишь меняет гарантии видимости изменений, но не превращает произвольные побочные эффекты в безопасный или детерминированный алгоритм.

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

Сервис обрабатывает список заказов и должен удалить просроченные записи после проверки каждой записи.

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

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

Третий вариант — применить специализированную операцию изменения коллекции, например removeIf, если проверка не требует сложного pipeline. Это проще и эффективнее для прямого удаления, но менее удобно, когда перед проверкой нужны несколько преобразований или внешние данные.

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

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

Можно ли изменять источник между созданием стрима и запуском терминальной операции?

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

Является ли изменение полей самих элементов таким же нарушением, как добавление или удаление элементов?

Нет, это разные случаи. Добавление или удаление меняет структуру источника и непосредственно нарушает его неинтерферирующий обход. Изменение поля объекта не обязательно меняет структуру, но может нарушить логику pipeline: последующие операции увидят уже изменённое состояние, а в параллельном стриме возможны гонки и несогласированные результаты. Поэтому изменяемые объекты в поведенческих параметрах также следует использовать осторожно.

Почему отсутствие ConcurrentModificationException не доказывает корректность обработки?

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