title: Spring Integration: Reading from Kafka Queue
body: <p>In a Spring Boot application, I want to use Spring Integration to read from a Kafka queue. The following is configured:</p>
  
  <pre><code>@Bean
  public KafkaMessageDrivenChannelAdapter&lt;String, String&gt;
  adapter(KafkaMessageListenerContainer&lt;String, String&gt; container) {
      KafkaMessageDrivenChannelAdapter&lt;String, String&gt; kafkaMessageDrivenChannelAdapter =
              new KafkaMessageDrivenChannelAdapter&lt;&gt;(container);
      kafkaMessageDrivenChannelAdapter.setOutputChannel(receiver());
      return kafkaMessageDrivenChannelAdapter;
  }
  
  @Bean
  public KafkaMessageListenerContainer&lt;String, String&gt; container() throws Exception {
      ContainerProperties properties = new ContainerProperties(this.topic);
      return new KafkaMessageListenerContainer&lt;&gt;(consumerFactory(), properties);
  }
  
  @Bean
  public ConsumerFactory&lt;String, String&gt; consumerFactory() {
      Map&lt;String, Object&gt; props = ... // set properties
      return new DefaultKafkaConsumerFactory&lt;&gt;(props);
  }
  
  @Bean
  public DirectChannel receiver() {
      return new DirectChannel();
  }
  
  @Autowired
  private Resolver resolver;
  
  @Bean
  public EventDrivenConsumer getEventDrivenConsumer() {
      return new EventDrivenConsumer(receiver(), resolver);
  }
  </code></pre>
  
  <p>The <code>Resolver</code> bean implements <code>MessageHandler</code>.</p>
  
  <p>Messages are received on the queue, but are not processed by the resolver bean.</p>
  
  <p>The Spring Boot application is annotated as follows:</p>
  
  <pre><code>@SpringBootApplication(exclude = KafkaAutoConfiguration.class)
  </code></pre>
  
  <p>So there should be no auto configuration of Kafka beans.</p>
  
  <p>The following is the error:</p>
  
  <pre><code>java.lang.NullPointerException: null
      at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:188) ~[spring-kafka-1.1.7.RELEASE.jar:na]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:72) ~[spring-kafka-1.1.7.RELEASE.jar:na]
      at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:47) ~[spring-kafka-1.1.7.RELEASE.jar:na]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:792) [spring-kafka-1.1.7.RELEASE.jar:na]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:736) [spring-kafka-1.1.7.RELEASE.jar:na]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.access$2100(KafkaMessageListenerContainer.java:246) [spring-kafka-1.1.7.RELEASE.jar:na]
      at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer$ListenerInvoker.run(KafkaMessageListenerContainer.java:1025) [spring-kafka-1.1.7.RELEASE.jar:na]
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_20]
      at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_20]
      at java.lang.Thread.run(Thread.java:745) [na:1.8.0_20]
  </code></pre>
  
  <p>Debugging, it is apparent that in <code>RecordMessagingMessageListenerAdapter</code> (top of the stacktrace), <code>this.methodHandler</code> is null.</p>
  
  <p>What is the correct way in Spring Integration to connect the channel to the bean which should process the messages?</p>