CODEWITHSUNDEEP
java-stream
du-it
du-it

Reputation: 3009

Why is the coder after using parallelStream not executed?

I intend to execute some time consuming code using using parallelStream. This seems to work well but I have the problem that the subsequent code is not executed:

@PreDestroy
public void tearDown() {
    final int mapSize = eventStreamProcessorMap.size();
    LOG.info("There are {} subscriptions to be stopped!", mapSize);

    final long start = System.currentTimeMillis();
    LocalTime time = LocalTime.now();

    final AtomicInteger count = new AtomicInteger();
    eventStreamProcessorMap.entrySet().parallelStream().forEach(entry -> {
        final Subscription sub = entry.getKey();
        final StreamProcessor processor = entry.getValue();

        LOG.info("Attempting to stop subscription {} of {} with id {} at {}", count.incrementAndGet(), mapSize, sub.id(), LocalTime.now().format(formatter));

            LOG.info("Stopping processor...");
            processor.stop();
            LOG.info("Processor stopped.");

            LOG.info("Removing subscription...");
            eventStreamProcessorMap.remove(sub);
            LOG.info("Subscription {} removed.", sub.id());

        LOG.info("Finished stopping processor {} with subscription {} in ParallelStream at {}: ", processor, sub, LocalTime.now().format(formatter));
        LOG.info(String.format("Duration: %02d:%02d:%02d:%03d (hh:mm:ss:SSS)",
                TimeUnit.MILLISECONDS.toHours(System.currentTimeMillis() - start),
                TimeUnit.MILLISECONDS.toMinutes(System.currentTimeMillis() - start)%60,
                TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - star0)%60,
                TimeUnit.MILLISECONDS.toMillis(System.currentTimeMillis() - start)%1000));
        LOG.info("--------------------------------------------------------------------------");
    });

    LOG.info("Helloooooooooooooo?????");
    LOG.info(String.format("Overall shutdown duration: %02d:%02d:%02d:%03d (hh:mm:ss:SSS)",
            TimeUnit.MILLISECONDS.toHours(System.currentTimeMillis() - start),
            TimeUnit.MILLISECONDS.toMinutes(System.currentTimeMillis() - start)%60,
            TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - start)%60,
            TimeUnit.MILLISECONDS.toMillis(System.currentTimeMillis() - start)%1000));
    LOG.info("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx");
}

The code after the parallelStream processing is not executed: 

LOG.info("Helloooooooooooooo?????");

does never appear in the log. Why not?

Upvotes: 0

Views: 49

Answers (1)

Eugene
Eugene

Reputation: 120858

This is caused by eventStreamProcessorMap.remove(sub); (which you have removed from the code now with the edit that you made). You are streaming over a Map entrySet (eventStreamProcessorMap) and removing elements from it - this is not allowed, that is why you get that ConcurrentModificationException.

If you really want to remove while iterating, use an Iterator or map.entrySet().removeIf(x -> {...})

Upvotes: 1

Related Questions