ConcurrentModificationException in unboudned stream

[cervebar]

Not sure is this is bug and I am not able to reproduce it in simple way. But facing this issue when using : unbounded input, LocalExecutor, flow with ReduceWindow.

 protected void execute() throws Exception {
      final ListDataSource<Pair<String, List<WindowRankValue>>> input = ListDataSource.unbounded(getInput());
      final Flow flow = Flow.create();
      CanoConf canoConf = CanoConf.builder().batchSize(2).build();
      Settings settings = new Settings();
      ListDataSink<Pair<String, String>> sink = ListDataSink.get();
      BUG.build(flow.createInput(input), settings).persist(sink);
      Executor executor = new LocalExecutor();
      executor.submit(flow).get();
      sink.getOutputs().forEach(System.out::println);
    }

Flow builder

public static Dataset<Pair<String, String>> build(
                                                    Dataset<Pair<String, List<WindowRankValue>>> input,
                                                    SCRankSettings settings ) {
    int batchSize = settings.getCanoConf().getBatchSize();
    UrlNormProvider urlNormProvider = settings.getUrlNormProvider();

    Dataset<List<NormalizedKeysWithValues>> canonized = ReduceWindow
        .named("reduce window")
        .of(input)
        .valueBy(e -> e)
        .reduceBy(values -> {
          return withFollowChain(values, urlNormProvider);
        })
        .windowBy(Count.of(batchSize))
        .output();

    Dataset<EntitySCranks> expanded = FlatMap
        .named("expand webservers/domains")
        .of(canonized)
        .using((List<NormalizedKeysWithValues> elems, Collector<EntitySCranks> collector) -> {
          elems.forEach(elem -> {
            List<EntitySCranks> val = expand(elem);
              val.forEach(e -> collector.collect(e));
          });
        }).output();

    Dataset<Pair<String, String>> out = ReduceByKey
        .named("reducebykey")
        .of(expanded)
        .keyBy(e -> e.getWindowRankKey().toString())
        .valueBy(e -> e.getValueWithUniqueDocs().toString())
        .combineBy(ranks -> {
          return "";
        })
        .output();
    return out;
  }

Exception trace:

2018-02-10 17:12:31 INFO  ReduceStateByKeyReducer:314 - Reducer reduce window::ReduceByKey#part-0 processing stats: at watermark 1518279151584, maxElementStamp 1518279151560
2018-02-10 17:12:31 INFO  ReduceStateByKeyReducer:314 - Reducer reducebykey#part-0 processing stats: at watermark 1518279151666, maxElementStamp 1518279151666
java.util.ConcurrentModificationException
	at java.util.HashMap$HashIterator.nextNode(HashMap.java:1442)
	at java.util.HashMap$EntryIterator.next(HashMap.java:1476)
	at java.util.HashMap$EntryIterator.next(HashMap.java:1474)
	at cz.seznam.euphoria.executor.local.ReduceStateByKeyReducer$ProcessingState.onUpstreamWindowTrigger(ReduceStateByKeyReducer.java:574)
	at cz.seznam.euphoria.executor.local.ReduceStateByKeyReducer.processWindowTrigger(ReduceStateByKeyReducer.java:842)
	at cz.seznam.euphoria.executor.local.ReduceStateByKeyReducer.run(ReduceStateByKeyReducer.java:723)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

[je-ik]

Is this still relevant?