Throw InterruptedException when using parallelStream - Java

Question

I have a method with nested for loops as follows:

public MinSpecSetFamily getMinDomSpecSets() throws InterruptedException {
    MinSpecSetFamily result = new MinSpecSetFamily();
    ResourceType minRT = this.getFirstEssentialResourceType();
    if (minRT == null || minRT.noSpecies()) {
        System.out.println("There is something wrong with the "
                + "minimal rticator, such as adjacent to no species. ");
    }
    for (Species spec : minRT.specList) {
        ArrayList<SpecTreeNode> leafList = this.getMinimalConstSpecTreeRootedAt(spec).getLeaves();
        for (SpecTreeNode leaf : leafList) {
            result.addSpecSet(new SpecSet(leaf.getAncestors()));
        }
    }
    return result;
}

This works fine, but the application is performance critical so I modified the method to use parallelStream() as follows:

public MinSpecSetFamily getMinDomSpecSets() throws InterruptedException {
    ResourceType minRT = this.getFirstEssentialResourceType();
    if (minRT == null || minRT.noSpecies()) {
        System.out.println("There is something wrong with the "
                + "minimal rticator, such as adjacent to no species. ");
    }

    MinSpecSetFamily result = minRT.specList.parallelStream()
            .flatMap(spec -> getMinimalConstSpecTreeRootedAt(spec).getLeaves().parallelStream())
            .map(leaf -> new SpecSet(leaf.getAncestors()))
            .collect(MinSpecSetFamily::new, MinSpecSetFamily::addSpecSet, MinSpecSetFamily::addMSSF);
    return result;
}

This worked fine until I wanted to introduce an InterruptedException in the 'getLeaves()' method. Now the parallelStream version will not compile as it says I have an unreported InterruptedException which must be caught or declared to be thrown. I think this is because the parallelStream runs on multiple threads. No combination of try/catch blocks suggested by my IDE resolves the issue.

The second solution posted in Interrupt parallel Stream execution suggests that I may be able to resolve the issue using ForkJoinPool but I have been unable to figure out how to modify my method to use this approach.

An `InterruptedException` is a checked exception, hence the compilation issue. It does not actually matter that you are running this in parallel, you need a way to pass this exception from inside the stream to outside of it. — Didier L, Jul 11 '18 at 15:01

assylias · Answer 1 · 2018-02-15T10:38:59.420

2

If you want to stick to your current design, you just need to catch the exception:

.flatMap(spec -> {
     try {
       return getMinimalConstSpecTreeRootedAt(spec).getLeaves().parallelStream();
     } catch (InterruptedException e) {
       // return something else to indicate interruption
       // maybe an empty stream?
     }
 }).map(...)

Note that a parallel stream of parallel streams is possibly unnecessary and parallelising the top level stream only may be sufficient performance-wise.

edited Feb 15 '18 at 10:38

answered Feb 15 '18 at 10:12

assylias

321,522
82
660
783

1

There is an extra ) after parallelStream in your answer, but thanks, this did the job. Also, I changed the second parallelStream to .stream() as suggested. – Steve W Feb 15 '18 at 10:24
On second thoughts, this does not fix the issue. If I return null in the catch block, the thread continues running for each additional 'spec'. Can I jump out of the parallelStream and return null from the whole method if any spec is null? – Steve W Feb 15 '18 at 11:03
@SteveW you may be able to throw an unchecked exception (but I'm not 100% sure if it will stop the whole stream though and it feels a bit hacky). It sounds like you may be better off using a different design, for example with an ExecutorService: submit the tasks and store the futures, then call `.get()` on the futures and do your thing. If any task gets interrupted you will know it and be able to stop the processing. – assylias Feb 15 '18 at 11:21
OK, that comment is a bit over my head. But if you are suggesting manually creating the required threads, I don't think I can do that. I replaced a nested for loop (which is inherently serial) with parallelStream and got a 50% speed improvement. If I don't use parallelStream I'm probably better off reverting to nested for loops as far as I can see. – Steve W Feb 15 '18 at 11:52

Throw InterruptedException when using parallelStream - Java

1 Answers1