title: kafka listener container in spring boot with annotations does not consume message
body: <p>Not sure if I am doing some thing wrong but I could not manage to make it work. Below is my code:</p>
  
  <pre><code>    public class EventsApp {
  
      private static final Logger log = LoggerFactory.getLogger(EventsApp.class);
  
      @Value("${kafka.topic:test}")
      private String topic;
  
      @Value("${kafka.messageKey:si.key}")
      private String messageKey;
  
      @Value("${kafka.broker.address:localhost:9092}")
      private String brokerAddress;
  
      @Value("${kafka.zookeeper.connect:localhost:2181}")
      private String zookeeperConnect;
  
  
      /**
       * Main method, used to run the application.
       *
       * @param args the command line arguments
       * @throws UnknownHostException if the local host name could not be resolved into an address
       */
      public static void main(String[] args) throws UnknownHostException, Exception {
  
          ConfigurableApplicationContext context
              = new SpringApplicationBuilder(EventsApp.class)
              .web(false)
              .run(args);
  
          MessageChannel toKafka = context.getBean("toKafka", MessageChannel.class);
          for (int i = 0; i &lt; 10; i++) {
              System.out.println("sending.."+toKafka.toString());
              toKafka.send(new GenericMessage&lt;&gt;("foo" + i));
          }
  
          Thread.sleep(115000);
  
          context.close();
  
  
          System.exit(0);
      }
  
      @KafkaListener(id = "baz", topics = "test",
          containerFactory = "kafkaListenerContainerFactory")
      public void listen(String data, Acknowledgment ack) {
          System.out.println("----- "+data);
          ack.acknowledge();
      }
  
      @Bean
      KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;Integer, String&gt;&gt;
      kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;Integer, String&gt; factory =
              new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          factory.setConcurrency(1);
          factory.getContainerProperties().setPollTimeout(3000);
          return factory;
      }
  
      @ServiceActivator(inputChannel = "toKafka")
      @Bean
      public MessageHandler handler() throws Exception {
          KafkaProducerMessageHandler&lt;String, String&gt; handler =
              new KafkaProducerMessageHandler&lt;&gt;(kafkaTemplate());
          handler.setTopicExpression(new LiteralExpression(this.topic));
          handler.setMessageKeyExpression(new LiteralExpression(this.messageKey));
          return handler;
      }
  
      @Bean
      public KafkaTemplate&lt;String, String&gt; kafkaTemplate() {
          return new KafkaTemplate&lt;&gt;(producerFactory());
      }
  
      @Bean
      public ProducerFactory&lt;String, String&gt; producerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
          props.put(ProducerConfig.RETRIES_CONFIG, 0);
          props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
          props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
          props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
          props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          return new DefaultKafkaProducerFactory&lt;&gt;(props);
      }
  
  
  
      @Bean
      public ConsumerFactory&lt;Integer, String&gt; consumerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddress);
          //props.put("zookeeper.connect", this.zookeeperConnect);
          props.put(ConsumerConfig.GROUP_ID_CONFIG, "siTestGroup");
          props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
          props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
          props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          return new DefaultKafkaConsumerFactory&lt;&gt;(props);
      }
  
  
      @Bean
      public TopicCreator topicCreator() {
          return new TopicCreator(this.topic, this.zookeeperConnect);
      }
  
      public static class TopicCreator implements SmartLifecycle {
  
          private final String topic;
  
          private final String zkConnect;
  
          private volatile boolean running;
  
          public TopicCreator(String topic, String zkConnect) {
              this.topic = topic;
              this.zkConnect = zkConnect;
          }
  
          @Override
          public void start() {
              ZkUtils zkUtils = new ZkUtils(new ZkClient(this.zkConnect, 6000, 6000,
                  ZKStringSerializer$.MODULE$), null, false);
              try {
                  AdminUtils.createTopic(zkUtils, topic, 1, 1, new Properties());
              }
              catch (TopicExistsException e) {
                  // no-op
              }
              this.running = true;
          }
  
          @Override
          public void stop() {
          }
  
          @Override
          public boolean isRunning() {
              return this.running;
          }
  
          @Override
          public int getPhase() {
              return Integer.MIN_VALUE;
          }
  
          @Override
          public boolean isAutoStartup() {
              return true;
          }
  
          @Override
          public void stop(Runnable callback) {
              callback.run();
          }
  
      }
     }
  </code></pre>
  
  <p>While I am able to produce message. I am using spring boot version 1.4.0.RELEASE and spring-integration-kafka version 2.0.1.RELEASE.</p>