title: Implement Custom error message handler in Spring Cloud Aws Kinesis Stream
body: <p>I have written spring cloud stream application listening to Kinesis streams</p>
  
  <p><strong>Message Flow across channel</strong></p>
  
  <pre><code>Kinesis Input --&gt; Transformer --&gt; Router (route to error or actaul processor)
  </code></pre>
  
  <p>I am planning to write error handling across these channels.
  Currently I am explicitly catching the error and sending to different channels. Instead of catching and modifying the payload, I would like to implement generi exception handler</p>
  
  <p>Binder</p>
  
  <pre><code>    public interface Binder {
  
        @Input(ITEM_KINESIS_INPUT)
        SubscribableChannel itemMessage();
  
      }
  </code></pre>
  
  <p>Configuration</p>
  
  <pre><code>    @Configuration
      @EnableIntegration
      public class IntegrationConfiguration {
  
        @Bean(name = DEFAULT_ITEM_LISTENER)
        public MessageChannel defaultItemChannel() {
          return new DirectChannel();
        }
  
        @Bean(name = DEFAULT_ERROR)
        public MessageChannel DEFAULT_ERROR() {
          return new DirectChannel();
        }
  
        @Bean
        @Transformer(inputChannel = ITEM_KINESIS_INPUT, outputChannel = ITEM_TRANSFORMER)
        public ItemTransformer itemTransformer() {
          return new ItemTransformer();
        }
  
  
        @ServiceActivator(inputChannel = ITEM_TRANSFORMER)
        @Bean
        public ItemMessageTypeRouter router() {
          return new ItemMessageTypeRouter();
        }
  
      // I tried this as well Not working
        @ServiceActivator(inputChannel = "errorChannel")
        @Bean
        public AbstractMessageHandler process() {
          return new AbstractMessageHandler() {
            @Override
            protected void handleMessageInternal(Message&lt;?&gt; message1) throws Exception {
              System.out.println("MessageHandler SA");
            }
          };
  
        }
  
      }
  </code></pre>
  
  <p>Transformer     </p>
  
  <pre><code>    public class ItemTransformer extends AbstractTransformer {
  
  
        @Override
        public String getComponentType() {
          return "transformer";
        }
  
  
  
        @Override
        protected Object doTransform(Message&lt;?&gt; message) throws Exception {
          Object transformedPayload = null;
          try {
            //transform
          } catch (Exception e) {
            transformedPayload =
                new MessageTransformationException(message, "failed to transform message", e);
          }
          return MessageBuilder.withPayload(transformedPayload).copyHeaders(message.getHeaders()).build();
        }
      }
  </code></pre>
  
  <p><strong>And in Router routing to different channel for error</strong></p>
  
  <pre><code>public class ItemMessageTypeRouter extends AbstractMappingMessageRouter {
  
  
    @Override
    protected List&lt;Object&gt; getChannelKeys(Message&lt;?&gt; message) {
      return (message.getPayload() instanceof Throwable) ? Arrays.asList(DEFAULT_ERROR)
          : Arrays.asList(DEFAULT_ITEM_LISTENER)
    } 
  }
  </code></pre>
  
  <p>Error Handling</p>
  
  <pre><code>@EnableBinding(Binder.class)
  public class ErrorListener implements MessageListener&lt;Message&gt; {
  
    @Autowired
    private MessagesBackupRepository backupRepository;
  
    @Override
    @StreamListener(DEFAULT_ERROR)
    public void receiveMessage(Message message) {
      log.error("Error has been reported " + message);
    }
  
  }
  </code></pre>
  
  <p>Can we implement generic exception handler all these channels.</p>
  
  <p>Currently I only handle error in Transformer, but if there is any exception occurs in Router or later channels, then this also needs to handled manually</p>
  
  <p>Any insight for better error handling</p>