Java 8: Lambda-Streams, filtrar por método com exceção

Estou com um problema ao tentar as expressões Lambda do Java 8. Geralmente funciona bem, mas agora tenho métodos que lançam IOException. É melhor se você observar o seguinte código:

class Bank{
    ....
    public Set<String> getActiveAccountNumbers() throws IOException {
        Stream<Account> s =  accounts.values().stream();
        s = s.filter(a -> a.isActive());
        Stream<String> ss = s.map(a -> a.getNumber());
        return ss.collect(Collectors.toSet());
    }
    ....
}

interface Account{
    ....
    boolean isActive() throws IOException;
    String getNumber() throws IOException;
    ....
}

O problema é que ele não é compilado, porque eu tenho que capturar as possíveis exceções dos métodos isActive e getNumber. Mas, mesmo que eu use explicitamente um bloco try-catch-Block, como abaixo, ele ainda não será compilado porque não capto a exceção. Portanto, ou há um erro no JDK ou não sei como capturar essas exceções.

class Bank{
    ....
    //Doesn't compile either
    public Set<String> getActiveAccountNumbers() throws IOException {
        try{
            Stream<Account> s =  accounts.values().stream();
            s = s.filter(a -> a.isActive());
            Stream<String> ss = s.map(a -> a.getNumber());
            return ss.collect(Collectors.toSet());
        }catch(IOException ex){
        }
    }
    ....
}

Como posso fazê-lo funcionar? Alguém pode me sugerir a solução certa?

Você deve capturar a exceção antes que ela escape do lambda:

s = s.filter(a -> { try { return a.isActive(); } 
                    catch (IOException e) { throw new UncheckedIOException(e); }}});

Considere o fato de que o lambda não é avaliado no local em que você o escreve, mas em algum local completamente não relacionado, dentro de uma classe JDK. Portanto, esse seria o ponto em que essa exceção verificada seria lançada e nesse local não será declarada.

Você pode lidar com isso usando um wrapper de seu lambda que converte exceções verificadas em desmarcadas:

public static <T> T uncheckCall(Callable<T> callable) {
  try { return callable.call(); }
  catch (RuntimeException e) { throw e; }
  catch (Exception e) { throw new RuntimeException(e); }
}

Seu exemplo seria escrito como

return s.filter(a -> uncheckCall(a::isActive))
        .map(Account::getNumber)
        .collect(toSet());

Nos meus projetos, eu lido com esse problema sem embrulhar; em vez disso, uso um método que efetivamente desativa a verificação de exceções do compilador. Escusado será dizer que isso deve ser tratado com cuidado e todos no projeto devem estar cientes de que uma exceção verificada pode aparecer onde não for declarada. Este é o código do encanamento:

public static <T> T uncheckCall(Callable<T> callable) {
  try { return callable.call(); }
  catch (Exception e) { return sneakyThrow(e); }
}
public static void uncheckRun(RunnableExc r) {
  try { r.run(); } catch (Exception e) { sneakyThrow(e); }
}
public interface RunnableExc { void run() throws Exception; }


@SuppressWarnings("unchecked")
private static <T extends Throwable> void sneakyThrow(Throwable t) throws T {
  throw (T) t;
}

e você pode esperar um IOExceptiongolpe na sua cara, mesmo que collectnão o declare. Na maioria dos casos, mas não em todos os casos da vida real, você deseja refazer a exceção, de qualquer maneira, e tratá-la como uma falha genérica. Em todos esses casos, nada se perde em clareza ou correção. Apenas tome cuidado com os outros casos em que você realmente deseja reagir à exceção no local. O desenvolvedor não será avisado pelo compilador de que há um problema IOExceptionpara capturar lá e o compilador de fato reclamará se você tentar capturá-lo, porque nós o enganamos, acreditando que nenhuma exceção pode ser lançada.

Você também pode propagar sua dor estática com lambdas, para que tudo pareça legível:

s.filter(a -> propagate(a::isActive))

propagateaqui recebe java.util.concurrent.Callablecomo parâmetro e converte qualquer exceção capturada durante a chamada em RuntimeException. Existe um método de conversão semelhante Throwables # propagate (Throwable) na Goiaba.

Esse método parece ser essencial para o encadeamento do método lambda, portanto, espero que um dia ele seja adicionado a uma das bibliotecas populares ou esse comportamento de propagação seja por padrão.

public class PropagateExceptionsSample {
    // a simplified version of Throwables#propagate
    public static RuntimeException runtime(Throwable e) {
        if (e instanceof RuntimeException) {
            return (RuntimeException)e;
        }

        return new RuntimeException(e);
    }

    // this is a new one, n/a in public libs
    // Callable just suits as a functional interface in JDK throwing Exception 
    public static <V> V propagate(Callable<V> callable){
        try {
            return callable.call();
        } catch (Exception e) {
            throw runtime(e);
        }
    }

    public static void main(String[] args) {
        class Account{
            String name;    
            Account(String name) { this.name = name;}

            public boolean isActive() throws IOException {
                return name.startsWith("a");
            }
        }


        List<Account> accounts = new ArrayList<>(Arrays.asList(new Account("andrey"), new Account("angela"), new Account("pamela")));

        Stream<Account> s = accounts.stream();

        s
          .filter(a -> propagate(a::isActive))
          .map(a -> a.name)
          .forEach(System.out::println);
    }
}

Essa UtilExceptionclasse auxiliar permite usar quaisquer exceções verificadas nos fluxos Java, como este:

Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
      .map(rethrowFunction(Class::forName))
      .collect(Collectors.toList());

Nota Class::forNamelança ClassNotFoundException, que está marcado . O próprio fluxo também lança ClassNotFoundException, e NÃO alguma exceção desmarcada.

public final class UtilException {

@FunctionalInterface
public interface Consumer_WithExceptions<T, E extends Exception> {
    void accept(T t) throws E;
    }

@FunctionalInterface
public interface BiConsumer_WithExceptions<T, U, E extends Exception> {
    void accept(T t, U u) throws E;
    }

@FunctionalInterface
public interface Function_WithExceptions<T, R, E extends Exception> {
    R apply(T t) throws E;
    }

@FunctionalInterface
public interface Supplier_WithExceptions<T, E extends Exception> {
    T get() throws E;
    }

@FunctionalInterface
public interface Runnable_WithExceptions<E extends Exception> {
    void run() throws E;
    }

/** .forEach(rethrowConsumer(name -> System.out.println(Class.forName(name)))); or .forEach(rethrowConsumer(ClassNameUtil::println)); */
public static <T, E extends Exception> Consumer<T> rethrowConsumer(Consumer_WithExceptions<T, E> consumer) throws E {
    return t -> {
        try { consumer.accept(t); }
        catch (Exception exception) { throwAsUnchecked(exception); }
        };
    }

public static <T, U, E extends Exception> BiConsumer<T, U> rethrowBiConsumer(BiConsumer_WithExceptions<T, U, E> biConsumer) throws E {
    return (t, u) -> {
        try { biConsumer.accept(t, u); }
        catch (Exception exception) { throwAsUnchecked(exception); }
        };
    }

/** .map(rethrowFunction(name -> Class.forName(name))) or .map(rethrowFunction(Class::forName)) */
public static <T, R, E extends Exception> Function<T, R> rethrowFunction(Function_WithExceptions<T, R, E> function) throws E {
    return t -> {
        try { return function.apply(t); }
        catch (Exception exception) { throwAsUnchecked(exception); return null; }
        };
    }

/** rethrowSupplier(() -> new StringJoiner(new String(new byte[]{77, 97, 114, 107}, "UTF-8"))), */
public static <T, E extends Exception> Supplier<T> rethrowSupplier(Supplier_WithExceptions<T, E> function) throws E {
    return () -> {
        try { return function.get(); }
        catch (Exception exception) { throwAsUnchecked(exception); return null; }
        };
    }

/** uncheck(() -> Class.forName("xxx")); */
public static void uncheck(Runnable_WithExceptions t)
    {
    try { t.run(); }
    catch (Exception exception) { throwAsUnchecked(exception); }
    }

/** uncheck(() -> Class.forName("xxx")); */
public static <R, E extends Exception> R uncheck(Supplier_WithExceptions<R, E> supplier)
    {
    try { return supplier.get(); }
    catch (Exception exception) { throwAsUnchecked(exception); return null; }
    }

/** uncheck(Class::forName, "xxx"); */
public static <T, R, E extends Exception> R uncheck(Function_WithExceptions<T, R, E> function, T t) {
    try { return function.apply(t); }
    catch (Exception exception) { throwAsUnchecked(exception); return null; }
    }

@SuppressWarnings ("unchecked")
private static <E extends Throwable> void throwAsUnchecked(Exception exception) throws E { throw (E)exception; }

}

Muitos outros exemplos de como usá-lo (após a importação estaticamente UtilException):

@Test
public void test_Consumer_with_checked_exceptions() throws IllegalAccessException {
    Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
          .forEach(rethrowConsumer(className -> System.out.println(Class.forName(className))));

    Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
          .forEach(rethrowConsumer(System.out::println));
    }

@Test
public void test_Function_with_checked_exceptions() throws ClassNotFoundException {
    List<Class> classes1
          = Stream.of("Object", "Integer", "String")
                  .map(rethrowFunction(className -> Class.forName("java.lang." + className)))
                  .collect(Collectors.toList());

    List<Class> classes2
          = Stream.of("java.lang.Object", "java.lang.Integer", "java.lang.String")
                  .map(rethrowFunction(Class::forName))
                  .collect(Collectors.toList());
    }

@Test
public void test_Supplier_with_checked_exceptions() throws ClassNotFoundException {
    Collector.of(
          rethrowSupplier(() -> new StringJoiner(new String(new byte[]{77, 97, 114, 107}, "UTF-8"))),
          StringJoiner::add, StringJoiner::merge, StringJoiner::toString);
    }

@Test    
public void test_uncheck_exception_thrown_by_method() {
    Class clazz1 = uncheck(() -> Class.forName("java.lang.String"));

    Class clazz2 = uncheck(Class::forName, "java.lang.String");
    }

@Test (expected = ClassNotFoundException.class)
public void test_if_correct_exception_is_still_thrown_by_method() {
    Class clazz3 = uncheck(Class::forName, "INVALID");
    }

Mas não o use antes de entender as seguintes vantagens, desvantagens e limitações :

• Se o código de chamada for manipular a exceção verificada, você DEVE adicioná-lo à cláusula throws do método que contém o fluxo. O compilador não forçará você a adicioná-lo mais, por isso é mais fácil esquecê-lo.

• Se o código de chamada já manipular a exceção verificada, o compilador lembrará que você deve adicionar a cláusula throws à declaração do método que contém o fluxo (caso contrário, ele dirá: A exceção nunca é lançada no corpo da instrução try correspondente )

• Em qualquer caso, você não poderá cercar o próprio fluxo para capturar a exceção verificada INSIDE no método que contém o fluxo (se você tentar, o compilador dirá: A exceção nunca é lançada no corpo da instrução try correspondente).

• Se você está chamando um método que literalmente nunca pode lançar a exceção declarada, não deve incluir a cláusula throws. Por exemplo: new String (byteArr, "UTF-8") lança UnsupportedEncodingException, mas UTF-8 é garantido pela especificação Java para sempre estar presente. Aqui, a declaração de lances é um incômodo e qualquer solução para silenciá-la com um mínimo de clichê é bem-vinda.

• Se você odeia exceções verificadas e acha que elas nunca devem ser adicionadas à linguagem Java (um número crescente de pessoas pensa dessa maneira, e eu NÃO SOU uma delas), simplesmente não adicione a exceção verificada ao lança a cláusula do método que contém o fluxo. A exceção verificada se comportará como uma exceção NÃO Verificada.

• Se você estiver implementando uma interface estrita na qual você não tem a opção de adicionar uma declaração de throws e ainda assim lançar uma exceção for inteiramente apropriado, agrupar uma exceção apenas para obter o privilégio de lançá-la resulta em um stacktrace com exceções falsas que não contribuem com informações sobre o que realmente deu errado. Um bom exemplo é Runnable.run (), que não lança nenhuma exceção verificada. Nesse caso, você pode decidir não adicionar a exceção verificada à cláusula throws do método que contém o fluxo.

• De qualquer forma, se você decidir NÃO adicionar (ou esquecer de adicionar) a exceção verificada à cláusula throws do método que contém o fluxo, esteja ciente destas 2 consequências de lançar exceções CHECKED:

1) O código de chamada não poderá identificá-lo pelo nome (se você tentar, o compilador dirá: A exceção nunca é lançada no corpo da instrução try correspondente). Ele irá borbulhar e provavelmente será capturado no loop principal do programa por alguns "catch Exception" ou "catch Throwable", que podem ser o que você deseja.

2) Ele viola o princípio da menor surpresa: não será mais suficiente capturar o RuntimeException para garantir a captura de todas as exceções possíveis. Por esse motivo, acredito que isso não deve ser feito no código da estrutura, mas apenas no código comercial que você controla completamente.

Concluindo: acredito que as limitações aqui não são sérias e a UtilExceptionclasse pode ser usada sem medo. No entanto, cabe a você!

• Referências:
  • http://www.philandstuff.com/2012/04/28/sneakily-throwing-checked-exceptions.html
  • http://www.mail-archive.com/[email protected]/msg05984.html
  • Anotação do Projeto Lombok: @SneakyThrows
  • Opinião de Brian Goetz (contra) aqui: Como posso lançar exceções CHECKED de dentro dos fluxos do Java 8?
  • /software/225931/workaround-for-java-checked-exceptions?newreg=ddf0dd15e8174af8ba52e091cf85688e *

Você pode potencialmente rolar sua própria Streamvariante envolvendo sua lambda para lançar uma exceção não verificada e depois desembrulhando essa exceção não verificada nas operações do terminal:

@FunctionalInterface
public interface ThrowingPredicate<T, X extends Throwable> {
    public boolean test(T t) throws X;
}

@FunctionalInterface
public interface ThrowingFunction<T, R, X extends Throwable> {
    public R apply(T t) throws X;
}

@FunctionalInterface
public interface ThrowingSupplier<R, X extends Throwable> {
    public R get() throws X;
}

public interface ThrowingStream<T, X extends Throwable> {
    public ThrowingStream<T, X> filter(
            ThrowingPredicate<? super T, ? extends X> predicate);

    public <R> ThrowingStream<T, R> map(
            ThrowingFunction<? super T, ? extends R, ? extends X> mapper);

    public <A, R> R collect(Collector<? super T, A, R> collector) throws X;

    // etc
}

class StreamAdapter<T, X extends Throwable> implements ThrowingStream<T, X> {
    private static class AdapterException extends RuntimeException {
        public AdapterException(Throwable cause) {
            super(cause);
        }
    }

    private final Stream<T> delegate;
    private final Class<X> x;

    StreamAdapter(Stream<T> delegate, Class<X> x) {
        this.delegate = delegate;
        this.x = x;
    }

    private <R> R maskException(ThrowingSupplier<R, X> method) {
        try {
            return method.get();
        } catch (Throwable t) {
            if (x.isInstance(t)) {
                throw new AdapterException(t);
            } else {
                throw t;
            }
        }
    }

    @Override
    public ThrowingStream<T, X> filter(ThrowingPredicate<T, X> predicate) {
        return new StreamAdapter<>(
                delegate.filter(t -> maskException(() -> predicate.test(t))), x);
    }

    @Override
    public <R> ThrowingStream<R, X> map(ThrowingFunction<T, R, X> mapper) {
        return new StreamAdapter<>(
                delegate.map(t -> maskException(() -> mapper.apply(t))), x);
    }

    private <R> R unmaskException(Supplier<R> method) throws X {
        try {
            return method.get();
        } catch (AdapterException e) {
            throw x.cast(e.getCause());
        }
    }

    @Override
    public <A, R> R collect(Collector<T, A, R> collector) throws X {
        return unmaskException(() -> delegate.collect(collector));
    }
}

Então você pode usar isso da mesma maneira exata que Stream:

Stream<Account> s = accounts.values().stream();
ThrowingStream<Account, IOException> ts = new StreamAdapter<>(s, IOException.class);
return ts.filter(Account::isActive).map(Account::getNumber).collect(toSet());

Essa solução exigiria um pouco de clichê, então sugiro que você dê uma olhada na biblioteca que já criei, que faz exatamente o que descrevi aqui para toda a Streamclasse (e mais!).

Use o método #propagate (). Exemplo de implementação não Guava do Java 8 Blog de Sam Beran :

public class Throwables {
    public interface ExceptionWrapper<E> {
        E wrap(Exception e);
    }

    public static <T> T propagate(Callable<T> callable) throws RuntimeException {
        return propagate(callable, RuntimeException::new);
    }

    public static <T, E extends Throwable> T propagate(Callable<T> callable, ExceptionWrapper<E> wrapper) throws E {
        try {
            return callable.call();
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw wrapper.wrap(e);
        }
    }
}

Isso não responde diretamente à pergunta (existem muitas outras respostas), mas tenta evitar o problema em primeiro lugar:

Na minha experiência, a necessidade de lidar com exceções em uma Stream(ou outra expressão lambda) geralmente vem do fato de que as exceções são declaradas lançadas a partir de métodos nos quais elas não devem ser lançadas. Isso geralmente vem da mistura da lógica de negócios com entrada e saída. Sua Accountinterface é um exemplo perfeito:

interface Account {
    boolean isActive() throws IOException;
    String getNumber() throws IOException;
}

Em vez de jogar um IOExceptionem cada getter, considere este design:

interface AccountReader {
    Account readAccount(…) throws IOException;
}

interface Account {
    boolean isActive();
    String getNumber();
}

O método AccountReader.readAccount(…)pode ler uma conta de um banco de dados ou arquivo ou qualquer outra coisa e lançar uma exceção se não for bem-sucedida. Ele constrói um Accountobjeto que já contém todos os valores, pronto para ser usado. Como os valores já foram carregados readAccount(…), os getters não lançariam uma exceção. Assim, você pode usá-los livremente em lambdas sem a necessidade de quebrar, mascarar ou ocultar as exceções.

Claro que nem sempre é possível fazê-lo da maneira que descrevi, mas muitas vezes é e leva a um código mais limpo (IMHO):

• Melhor separação de preocupações e seguir o princípio de responsabilidade única
• Menos clichê: você não precisa desordenar seu código throws IOExceptionpara não usar, mas para satisfazer o compilador
• Tratamento de erros: você lida com os erros onde eles ocorrem - ao ler de um arquivo ou banco de dados - em vez de em algum lugar no meio da lógica de negócios, apenas porque deseja obter um valor de campos
• Você pode ser Account imutável e lucrar com as vantagens (por exemplo, segurança da linha)
• Você não precisa de "truques sujos" ou soluções alternativas para usar Accountem lambdas (por exemplo, em a Stream)

Pode ser resolvido abaixo do código simples com Stream e Try no AbacusUtil :

Stream.of(accounts).filter(a -> Try.call(a::isActive)).map(a -> Try.call(a::getNumber)).toSet();

Divulgação: Eu sou o desenvolvedor do AbacusUtil.

Estendendo a solução @marcg, você normalmente pode lançar e capturar uma exceção verificada no Streams; isto é, o compilador solicitará que você pegue / lance novamente como você estava fora dos fluxos !!

@FunctionalInterface
public interface Predicate_WithExceptions<T, E extends Exception> {
    boolean test(T t) throws E;
}

/**
 * .filter(rethrowPredicate(t -> t.isActive()))
 */
public static <T, E extends Exception> Predicate<T> rethrowPredicate(Predicate_WithExceptions<T, E> predicate) throws E {
    return t -> {
        try {
            return predicate.test(t);
        } catch (Exception exception) {
            return throwActualException(exception);
        }
    };
}

@SuppressWarnings("unchecked")
private static <T, E extends Exception> T throwActualException(Exception exception) throws E {
    throw (E) exception;
}

Em seguida, seu exemplo seria escrito da seguinte maneira (adicionando testes para mostrá-lo mais claramente):

@Test
public void testPredicate() throws MyTestException {
    List<String> nonEmptyStrings = Stream.of("ciao", "")
            .filter(rethrowPredicate(s -> notEmpty(s)))
            .collect(toList());
    assertEquals(1, nonEmptyStrings.size());
    assertEquals("ciao", nonEmptyStrings.get(0));
}

private class MyTestException extends Exception { }

private boolean notEmpty(String value) throws MyTestException {
    if(value==null) {
        throw new MyTestException();
    }
    return !value.isEmpty();
}

@Test
public void testPredicateRaisingException() throws MyTestException {
    try {
        Stream.of("ciao", null)
                .filter(rethrowPredicate(s -> notEmpty(s)))
                .collect(toList());
        fail();
    } catch (MyTestException e) {
        //OK
    }
}

Para adicionar corretamente o código de manipulação IOException (ao RuntimeException), seu método será semelhante a este:

Stream<Account> s =  accounts.values().stream();

s = s.filter(a -> { try { return a.isActive(); } 
  catch (IOException e) { throw new RuntimeException(e); }});

Stream<String> ss = s.map(a -> { try { return a.getNumber() }
  catch (IOException e) { throw new RuntimeException(e); }});

return ss.collect(Collectors.toSet());

O problema agora é que o IOExceptionarquivo terá que ser capturado como um RuntimeExceptione convertido novamente em um IOException- e isso adicionará ainda mais código ao método acima.

Por que usar Streamquando isso pode ser feito da seguinte maneira - e o método é executado IOExceptionpara que também não seja necessário código extra para isso:

Set<String> set = new HashSet<>();
for(Account a: accounts.values()){
  if(a.isActive()){
     set.add(a.getNumber());
  } 
}
return set;

Mantendo essa questão em mente, desenvolvi uma pequena biblioteca para lidar com exceções e lambdas verificadas. Adaptadores personalizados permitem integrar com tipos funcionais existentes:

stream().map(unchecked(URI::new)) //with a static import

https://github.com/TouK/ThrowingFunction/

Seu exemplo pode ser escrito como:

import utils.stream.Unthrow;

class Bank{
   ....
   public Set<String> getActiveAccountNumbers() {
       return accounts.values().stream()
           .filter(a -> Unthrow.wrap(() -> a.isActive()))
           .map(a -> Unthrow.wrap(() -> a.getNumber()))
           .collect(Collectors.toSet());
   }
   ....
}

A classe Unthrow pode ser obtida aqui https://github.com/SeregaLBN/StreamUnthrower

Se você não se importa em usar bibliotecas de terceiros, a cyclops-react lib da AOL , disclosure :: eu sou um colaborador, possui uma classe ExceptionSoftener que pode ajudar aqui.

 s.filter(softenPredicate(a->a.isActive()));

As interfaces funcionais em Java não declaram nenhuma exceção marcada ou desmarcada. Precisamos alterar a assinatura dos métodos de:

boolean isActive() throws IOException; 
String getNumber() throwsIOException;

Para:

boolean isActive();
String getNumber();

Ou manipule-o com o bloco try-catch:

public Set<String> getActiveAccountNumbers() {
  Stream<Account> s =  accounts.values().stream();
  s = s.filter(a -> 
    try{
      a.isActive();
    }catch(IOException e){
      throw new RuntimeException(e);
    }
  );
  Stream<String> ss = s.map(a -> 
    try{
      a.getNumber();
    }catch(IOException e){
      throw new RuntimeException(e);
    }
  );
  return ss.collect(Collectors.toSet());
}

Outra opção é escrever um wrapper personalizado ou usar uma biblioteca como ThrowingFunction. Com a biblioteca, precisamos adicionar apenas a dependência ao nosso pom.xml:

<dependency>
    <groupId>pl.touk</groupId>
    <artifactId>throwing-function</artifactId>
    <version>1.3</version>
</dependency>

E use as classes específicas como ThrowingFunction, ThrowingConsumer, ThrowingPredicate, ThrowingRunnable, ThrowingSupplier.

No final, o código fica assim:

public Set<String> getActiveAccountNumbers() {
  return accounts.values().stream()
    .filter(ThrowingPredicate.unchecked(Account::isActive))
    .map(ThrowingFunction.unchecked(Account::getNumber))
    .collect(Collectors.toSet());
}

Não vejo nenhuma maneira de lidar com a exceção verificada no fluxo (Java -8), a única maneira que apliquei é pegar a exceção verificada no fluxo e lançá-las novamente como exceção não verificada.

        Arrays.stream(VERSIONS)
        .map(version -> TemplateStore.class
                .getClassLoader().getResourceAsStream(String.format(TEMPLATE_FILE_MASK, version)))
        .map(inputStream -> {
            try {
                return ((EdiTemplates) JAXBContext.newInstance(EdiTemplates.class).createUnmarshaller()
                        .unmarshal(inputStream)).getMessageTemplate();
            } catch (JAXBException e) {
                throw new IllegalArgumentException(ERROR, e);
            }})
        .flatMap(Collection::stream)
        .collect(Collectors.toList());

Se você deseja manipular a exceção no fluxo e continuar processando as adicionais, há um excelente artigo no DZone de Brian Vermeer, usando o conceito de Either. Ele mostra uma excelente maneira de lidar com essa situação. A única coisa que falta é o código de exemplo. Esta é uma amostra da minha exploração usando os conceitos desse artigo.

@Test
public void whenValuePrinted_thenPrintValue() {

    List<Integer> intStream = Arrays.asList(0, 1, 2, 3, 4, 5, 6);
    intStream.stream().map(Either.liftWithValue(item -> doSomething(item)))
             .map(item -> item.isLeft() ? item.getLeft() : item.getRight())
             .flatMap(o -> {
                 System.out.println(o);
                 return o.isPresent() ? Stream.of(o.get()) : Stream.empty();
             })
             .forEach(System.out::println);
}

private Object doSomething(Integer item) throws Exception {

    if (item == 0) {
        throw new Exception("Zero ain't a number!");
    } else if (item == 4) {
        return Optional.empty();
    }

    return item;
}