Example of non-interference in Java 8

Question

According to this question, we can modify the source and it's not called interference:

you can modify the stream elements themselves and it should not be called as "interference".

According to this question, the code

List<String> list = new ArrayList<>();
  list.add("test");
  list.forEach(x -> list.add(x));

will throw ConcurrentModificationException.

But my code,

Employee[] arrayOfEmps = {
                new Employee(1, "Jeff Bezos"),
                new Employee(2, "Bill Gates"),
                new Employee(3, "hendry cavilg"),
                new Employee(4, "mark cuban"),
                new Employee(5, "zoe"),
                new Employee(6, "billl clinton"),
                new Employee(7, "ariana") ,
                new Employee(8, "cathre"),
                new Employee(9, "hostile"),
                new Employee(10, "verner"),
            };
        Employee el=new Employee(1, "Jeff Bezos");
        List<Employee> li=Arrays.asList(arrayOfEmps);
        li.stream().map(s->{s.setName("newname");return s;}).forEach(System.out::print);

doesn't throw ConcurrentModificationException, even though it in fact changes the source.

And this code,

Employee[] arrayOfEmps = {
                new Employee(1, "Jeff Bezos"),
                new Employee(2, "Bill Gates"),
                new Employee(3, "hendry cavilg"),
                new Employee(4, "mark cuban"),
                new Employee(5, "zoe"),
                new Employee(6, "billl clinton"),
                new Employee(7, "ariana") ,
                new Employee(8, "cathre"),
                new Employee(9, "hostile"),
                new Employee(10, "verner"),
            };
        Employee el=new Employee(1, "Jeff Bezos");
        List<Employee> li=Arrays.asList(arrayOfEmps);
        li.stream().map(s->{s.setName("newname");li.add(s);return s;}).limit(10).forEach(System.out::print);

throws

Exception in thread "main" java.lang.UnsupportedOperationException
    at java.util.AbstractList.add(Unknown Source)
    at java.util.AbstractList.add(Unknown Source)
    at java8.Streams.lambda$0(Streams.java:33)
    at java.util.stream.ReferencePipeline$3$1.accept(Unknown Source)
    at java.util.Spliterators$ArraySpliterator.forEachRemaining(Unknown Source)
    at java.util.stream.AbstractPipeline.copyInto(Unknown Source)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source)
    at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(Unknown Source)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(Unknown Source)
    at java.util.stream.AbstractPipeline.evaluate(Unknown Source)
    at java.util.stream.ReferencePipeline.forEach(Unknown Source)

So, I don't exactly understand what type of modifications are allowed to the source and what are not. It would be very helpful to see an example which interferes and have a stateful and side-effect producing stream, with proper indication that which is which.

Answer 1

When you do this:

li.stream().map(s->{s.setName("newname");return s;})

you didn't alter the list itself but an element within this list; so it doesn't trigger a ConcurrentModificationException as you expected.

In the last code snippet, you are using Array.asList.
You have to know that Array.asList returns a mere read-only wrapper over an array (a specific inner ArrayList class) explaining why add is not supported.

Indeed this inner class does not override AbstractList#add method by design; causing UnsupportedOperationException; and still not ConcurrentModificationException as you expected again.

Here's an example similar to your last snippet that does throw a ConcurrentModificationException:

public static void main(String[] args) {
    Employee[] arrayOfEmps = {
      new Employee(1, "Jeff Bezos")
    };
    Employee el = new Employee(11, "Bill Gates");
    List<Employee> li = new ArrayList<>(Arrays.asList(arrayOfEmps)); // to avoid read-only restriction
    li.stream().peek(s -> li.add(el)).forEach(System.out::print);
} 

Note that I'm wrapping the Arrays.List return with a "true" ArrayList, allowing writing because this one implements the add method ;)

Answer 2

Your first example changes the existing elements of the Stream, but doesn't add or remove elements from the source. Therefore it's not an interference.

Your second example attempts to do interference, by adding an element to the source during the Stream pipeline. However, you get UnsupportedOperationException instead of ConcurrentModificationException, since you try to add elements to a fixed sized List (which is returned by Arrays.asList).

Change your second example to:

List<Employee> li=new ArrayList<>(Arrays.asList(arrayOfEmps));
li.stream().map(s->{s.setName("newname");li.add(s);return s;}).limit(10).forEach(System.out::print);

and you should get ConcurrentModificationException.

Answer 3

This is called a structural on non-structural change of the source of the Stream. For example ArrayList doc says:

merely setting the value of an element is not a structural modification...

So in your example this means that changing Employee per-se, does not changes the List itself (does not remove or add an element). But changing the List itself, will fail with a ConcurrentModificationException:

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

list.stream().forEach(x -> list.remove(x));

But there are sources where interference is more than OK, called weakly consistent traversal, like ConcurrentHashMap:

 ConcurrentHashMap<Integer, String> chm = new ConcurrentHashMap<>();
 chm.put(1, "one");
 chm.put(2, "two");
 chm.put(3, "three");

 chm.entrySet().stream()
         .forEach(x -> chm.remove(x.getKey()));

This will not fail with ConcurrentModificationException.

Answer 4

Your code is modifying the stream element, not the list itself:

s->{s.setName("newname") changes a field name in the element of the stream, this doesn't interfere with the stream or its source.

The java.lang.UnsupportedOperationException is also unrelated, it's caused by the fact that you attempt to call add on a fixed-size list (which is the result of Arrays.asList). Any time one calls add, remove, etc. on the result of Arrays.asList, the unsupported operation exception is raised, as the collect is fixed in size.

Answer 5

Your first example does not modify the initial list.

A stream is just a view on the list, therefore the initial list is not at all affected by your stream code.

Whereas the second example makes explicit use of list.add().

That is all there is to this.