title: Spring Cloud Stream Kafka Error channel
body: <p>I am trying to set up a binding to forward Kafka messages from the Spring Integration errorChannel to a custom channel (for centralised error processing).</p>
  
  <p>The error messages are being sent to the configured channel, but they are arriving as a <code>GenericMessage</code> with a byte[] payload, which consists of exception details and the failed message.</p>
  
  <p>My config:</p>
  
  <pre><code>spring:
    cloud:
      stream:
        kafka:
          bindings:
            accountOut.producer:
              sync: true
          binder:
            autoCreateTopics: false
            headers:
                 - spanId
                 - spanTraceId
                 - spanSampled
                 - spanParentSpanId
                 - spanName
                 - spanFlags
                 - eventType
                 - Authorization
        bindings:
          error:
            destination: test-error
          accountOut:
            producer.partitionKeyExpression: payload.key
            content-type: application/json
            destination: account
    kafka:
      producer.keySerializer: org.apache.kafka.common.serialization.StringSerializer
      consumer.valueDeserializer: org.apache.kafka.common.serialization.StringDeserializer
  </code></pre>
  
  
  
  <p>I'm listening with a <code>@StreamListener(target = "kieran-error")</code>
  and the consumer configured with <code>@Input("kieran-error") SubscribableChannel</code> </p>
  
  <p>Reading <a href="https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#kafka-error-channels" rel="nofollow noreferrer">the docs</a>, I was expecting the message to arrive as and <code>ErrorMessage</code>. Is there a way I can acheive this? Or to configure the payload to arrive as an Object?</p>
  
  <p>Versions I'm using:</p>
  
  <ul>
  <li>Spring Boot 1.5.8</li>
  <li>Spring Cloud Edgware </li>
  <li>Kafka 11 client</li>
  <li>Spring Integration core 4.3.12</li>
  </ul>
  
  <p>--- Question Update ---</p>
  
  <p>I realise now I could configure Spring Integration to forward to a Kafka Topic by listening to the <code>errorChannel</code>, e.g.</p>
  
  
  
  <pre><code>@Bean
  @ServiceActivator(inputChannel = "errorChannel")
  public MessageHandler handler() throws Exception {
      KafkaProducerMessageHandler&lt;String, String&gt; handler =
              new KafkaProducerMessageHandler&lt;&gt;(kafkaTemplate());
      handler.setTopicExpression(new LiteralExpression("someTopic"));
      handler.setMessageKeyExpression(new LiteralExpression("someKey"));
      return handler;
  }
  </code></pre>
  
  
  
  <p>But is it possible to configure this flow in the properties yaml rather than code? That's where all the other Kafka config is so configuring the kafka template in code is not ideal. </p>
  
  <p>Another option would be to listen for the ErrorMessage explicitly and send to a kafka output channel in the code:</p>
  
  
  
  <pre><code>@ServiceActivator(inputChannel = "errorChannel")
  public void handle(ErrorMessage em) {
      outputChannel.kieranError().send(...)
  }
  </code></pre>