title: Error handling in spring integration - How to get all the errors thrown in multiple threads and send them to the error-channel
body: <p>I have two threads and each of the method in the thread throws an exception. How do I get all the errors that's thrown in each of the thread? In this code, the error channel just catch one of the errors. Basically my goal is to catch all the errors and send them to the caller (a rest controller). Any help will be appreciated. Thanks.</p>
  
  <p>Integration.java</p>
  
  <pre><code>public IntegrationFlow provisionUserFlow() {
  return IntegrationFlows.from("input.channel")
    .publishSubscribeChannel(Executors.newCachedThreadPool(),
        s -&gt; s
              .subscribe(f -&gt; f.handle(provisionerA, "provision"))
              .subscribe(f -&gt; f.handle(provisionerB, "provision"))
    .get();
  }
  
  @ServiceActivator( inputChannel = "errorChannel", outputChannel = "replyChannel")
  
  public boolean processErrors(Exception message) throws RuntimeException{
  
  System.out.println("Message" + message.getMessage());
  System.out.println ("******************************");
  
  throw new RuntimeException(message.getMessage());
  }
  </code></pre>
  
  <p>MGateway.java</p>
  
  <pre><code>@MessagingGateway(errorChannel = "errorChannel")
  public interface MGateway {
  
  @Gateway(requestChannel = "input.channel", replyChannel = "replyChannel") 
  boolean invokeProvisioner(User user);
  }
  </code></pre>
  
  <p><strong>SOLUTION</strong></p>
  
  <pre><code>@Bean
  public IntegrationFlow provisionUserFlow() {
  return
      IntegrationFlows.from("input.channel")
      .publishSubscribeChannel(Executors.newCachedThreadPool(),
          s -&gt; s.applySequence(true)
              .subscribe(f -&gt; f.enrichHeaders(e -&gt; e.header(MessageHeaders.ERROR_CHANNEL, "errorChannel", true))
                  .handle(provisionerA, "provision")
                  .channel("aggregatorChannel")
              )
              .subscribe(f -&gt; f.enrichHeaders(e -&gt; e.header(MessageHeaders.ERROR_CHANNEL, "errorChannel", true))
                  .handle(provisionerB, "provision")
                  .channel("aggregatorChannel"))
              )
          .get();
  }
  
  @Bean
  public IntegrationFlow aggregateFlow() {
  return IntegrationFlows.from("aggregatorChannel")
      .channel( aggregatorChannel)
      .aggregate( a -&gt; a.processor( collect, "aggregatingMethod"))
      .get();
  }
  
  @Transformer( inputChannel = "errorChannel", outputChannel = "aggregatorChannel")
  public Message&lt;?&gt; errorChannelHandler(ErrorMessage errorMessage) throws RuntimeException {
  
  Message&lt;?&gt; failedMessage =  ((MessagingException) errorMessage.getPayload())
      .getFailedMessage();
  
  Exception exception = (Exception) errorMessage.getPayload();
  
  return  MessageBuilder.withPayload( exception.getMessage())
         .copyHeadersIfAbsent( failedMessage.getHeaders() )
         .build();
  }
  </code></pre>