How Java streams works when the list it is working is concurrently modified?

Question

If I have a list with 10 elements and I take a stream view of it, filter it by age and collect it in anther list. And in between someone adds 5 elements to the list. What will be the behavior of the streams will it work with only 10 elements or 15 ??.

Answer 1

Read the "Non-Interference" section in the javadoc. In particular:

Unless the stream source is concurrent, modifying a stream's data source during execution of a stream pipeline can cause exceptions, incorrect answers, or nonconformant behavior.

So the answer to your question is that it depends on the type of list, whether it's concurrent or not, and if it's not, how it handles concurrent operations and at what point you make the moddification (before or in the middle of the terminal operation).

Answer 2

You will get a ConcurrentModificationException - if you're lucky. From the exception's documentation, the most relevant part is:

In general, the results of the iteration are undefined under these circumstances. Some Iterator implementations (including those of all the general purpose collection implementations provided by the JRE) may choose to throw this exception if this behavior is detected. Iterators that do this are known as fail-fast iterators, as they fail quickly and cleanly, rather that risking arbitrary, non-deterministic behavior at an undetermined time in the future.

I tried this example code:

List<Integer> myList = new ArrayList<>();
myList.add(1);
myList.add(2);
myList.add(3);
Thread threadA = new Thread(() -> {
    try {
        myList.add(4);
        Thread.sleep(1000);
        myList.add(5);
        Thread.sleep(1000);
        myList.add(6);
        Thread.sleep(1000);
        myList.add(7);
        Thread.sleep(1000);
        myList.add(8);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
});
Thread threadB = new Thread(() -> {
    myList.forEach(i -> {
        try {
            System.out.println(i);
            Thread.sleep(800);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
});

threadA.start();
threadB.start();

Which results in:

1
2
Exception in thread "Thread-1" java.util.ConcurrentModificationException
    at java.util.ArrayList.forEach(ArrayList.java:1252)
    at com.company.Main.lambda$main$2(Main.java:30)
    at java.lang.Thread.run(Thread.java:748)

Switching out the non-thread-safe ArrayList for a synchronized Vector results in:

1
2
3
4

But you can't rely on this behavior. The documentation is quite clear that it is undefined, and that makes it likely to change between different implementations or even different versions. Don't do it. Use thread safe containers such as Vector and ConcurrentHashMap, and use synchronization primitives around them in a way that makes sense for the problem you're trying to solve.

Answer 3

Well, I tried with both ArrayList and Vector and both of them gave me ConcurrentModificationException.

However CopyOnWriteArrayList works fine.

    int valueLength = 10;
    List<Integer> myList = new Vector<>();
    for (int i = 0; i < valueLength; i++) {
        myList.add(i);
    }
    new Thread(() -> {
        try {
            myList.stream().filter(i -> {
                try {
                    if (i % 3 == 0) {
                        Thread.sleep(10);
                        return false;
                    }
                    return true;
                } catch (Exception e) {
                    e.printStackTrace();
                    return false;
                }
            }).collect(Collectors.toList());
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }).start();

    new Thread(() -> {
        try {
            for (int i = valueLength; i > 0; i--) {
                if (i % 3 == 0) {
                    Thread.sleep(5);
                    myList.remove(i);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();