title: Spring Cloud Stream Message Handling
body: <p>While playing around with Spring Cloud Stream, Kafka, Debezium and Postgresql I am facing the following issue :
  To highlight :
  A table person(id,name) is there in postgresql , debezium fetched the change events and throws to Kafka topic(person) , verified messages are there in the topic , the form of the message is as </p>
  
  <pre><code>{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"}],"optional":true,"name":"dbserver1.datafeeds.person.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"name"}],"optional":true,"name":"dbserver1.datafeeds.person.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":true,"field":"version"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":true,"field":"ts_usec"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"boolean","optional":true,"default":false,"field":"snapshot"},{"type":"boolean","optional":true,"field":"last_snapshot_record"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"dbserver1.datafeeds.person.Envelope"},"payload":{"before":null,"after":{"id":123,"name":"Muhammad Sufyian"},"source":{"version":"0.9.0.Alpha1","name":"dbserver1","ts_usec":946684800000000000,"txId":350418,"lsn":9647073220,"snapshot":false,"last_snapshot_record":null},"op":"c","ts_ms":1534254709618}}
  </code></pre>
  
  <p>on the consumer side , I am using Spring Cloud Stream , 
  Message class looks like this </p>
  
  <pre><code>public class Person {
  
      private Integer id;
      private String name;
  
      public Person() {
      }
  
      public Person(Integer id, String name) {
          this.id = id;
          this.name = name;
      }
  
      public Integer getId() {
          return id;
      }
  
      public void setId(Integer id) {
          this.id = id;
      }
  
      public String getName() {
          return name;
      }
  
      public void setName(String name) {
          this.name = name;
      }
  
      @Override
      public String toString() {
          return "Person{" +
                  "id=" + id +
                  ", name='" + name + '\'' +
                  '}';
      }
  }
  
  Channel class looks like this :
  </code></pre>
  
  <p>@EnableBinding(Processor.class)
  public class Channel {</p>
  
  <pre><code>@StreamListener(Processor.INPUT)
  public void notify(@Payload Person person){
          System.out.println("Print");
          System.out.println("Id:"+person.getId());
          System.out.println("Name:"+person.getName());
  }
  </code></pre>
  
  <p>}</p>
  
  <p>and main class look like this </p>
  
  <pre><code>@SpringBootApplication
  @EnableAutoConfiguration(exclude={DataSourceAutoConfiguration.class})
  public class PartnerHotelChangedProcessorApplication {
  
      public static void main(String[] args) {
          SpringApplication.run(PartnerHotelChangedProcessorApplication.class, args);
      }
  
  
  }
  </code></pre>
  
  <p>application.properties looks like this :</p>
  
  <pre><code>spring.cloud.stream.kafka.binder.brokers=&lt;list-of-kafka-brokers&gt;
  spring.cloud.stream.bindings.input.destination=dbserver1.datafeeds.person
  spring.cloud.stream.bindings.input.content-type=application/json
  </code></pre>
  
  <p>The payload fetched via @StreamListner is null , I doubt there are problems with Message conversion here .</p>
  
  <p>Any direction is appreciated</p>
  
  <p>Stack trace :</p>
  
  <pre><code>2018-08-16 15:23:36.102  INFO 23525 --- [container-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] Discovered group coordinator kafka-02.hotel02.pro06.eu.idealo.com:9092 (id: 2147481645 rack: null)
  2018-08-16 15:23:36.104  INFO 23525 --- [container-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] Revoking previously assigned partitions []
  2018-08-16 15:23:36.105  INFO 23525 --- [container-0-C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$1  : partitions revoked: []
  2018-08-16 15:23:36.105  INFO 23525 --- [container-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] (Re-)joining group
  2018-08-16 15:23:39.168  INFO 23525 --- [container-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] Successfully joined group with generation 1
  2018-08-16 15:23:39.170  INFO 23525 --- [container-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] Setting newly assigned partitions [dbserver1.datafeeds.person-0]
  2018-08-16 15:23:39.232  INFO 23525 --- [container-0-C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$1  : partitions assigned: [dbserver1.datafeeds.person-0]
  2018-08-16 15:23:59.003  INFO 23525 --- [a9-3b8495d0ab90] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] Group coordinator kafka-02.hotel02.pro06.eu.idealo.com:9092 (id: 2147481645 rack: null) is unavailable or invalid, will attempt rediscovery
  [B@25c28bf5
  Print
  2018-08-16 15:24:00.512  INFO 23525 --- [container-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] Discovered group coordinator kafka-02.hotel02.pro06.eu.idealo.com:9092 (id: 2147481645 rack: null)
  2018-08-16 15:24:00.538 ERROR 23525 --- [container-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=anonymous.44607cf2-3162-459a-8ca9-3b8495d0ab90] Offset commit failed on partition dbserver1.datafeeds.person-0 at offset 58: The coordinator is not aware of this member.
  2018-08-16 15:24:00.555 ERROR 23525 --- [container-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: null
  </code></pre>