How to reduce a stream of Futures in Java?

Question

Assuming that I have given a Stream of Futures, which I want to reduce by invoking the Stream#reduce method. But I don't want to reduce the Futures itself, but the result of the Future (Future#get). The problem is, that the get method may throw an ExecutionException and does not provide a result in this case.

This is the reason why

Stream<Future<Integer>> stream = ...;
BinaryOperator<Integer> sum = (i1, i2) -> i1 + i2;  
stream.map(future -> future.get())
      .reduce(sum); // does not work, get needs to handle exceptions!

So, I have to catch the exceptions:

stream.map(future -> {
    Integer i = null;
    try {
        i = future.get();
    } catch (InterruptedException e) {
    } catch (ExecutionException e) {}
    return i;
}).reduce(sum); 

But in this approach, I may get troubles because null values may appear.

So, to get rid of these, I would have to filter those out, where a ExecutionException appeared:

stream.filter(future -> {
    Integer i = null;
    try {
        i = future.get();
    } catch (InterruptedException e) {
    } catch (ExecutionException e) {
    }
    return i != null;
})
.map(future -> {
    Integer i = null;
    try {
        i = future.get();
    } catch (InterruptedException e) {
    } catch (ExecutionException e) {
    }
    return i;
}).reduce(sum);

I think, this code would work.. But I don't want to believe, that this is the only and the smartest way to reduce Futures.

Any ideas or suggestions?

Answer 1

You could extract the value from the future first, and then filter out null:

Integer result = stream
    .map(future -> {
        try {
          return future.get();
        } catch (InterruptedException | ExecutionException e) {
        }
        return null; })
    .filter(Objects::nonNull)
    .reduce(sum)
    .orElse(0);

Answer 2

One of the ways to simplify it could be:

void reduceImpl(Stream<Future<Integer>> stream) {
    Optional<Integer> integerOptional = stream
            .map(this::transform)
            .filter(Objects::nonNull)
            .reduce(Integer::sum);
}

private Integer transform(Future<Integer> future) {
    try {
        return future.get();
    } catch (InterruptedException | ExecutionException e) {
        return null; // should ideally be handled properly
    }
}