title: Failed to start bean kafkaListenerContainer: java.lang.IllegalArgumentException
body: <p>I am using <a href="https://github.com/iainporter/spring-file-poller" rel="nofollow noreferrer">this</a> as example for reading file using spring integration, its working fine but when I try to send file to kafka producer it doesn't work. I tried to look up this issue on internet but could not find help.
  Here is my code:</p>
  <p>file: MessageProcessingIntegrationFlow.java:</p>
  <pre><code>@Bean
  public IntegrationFlow writeToFile() {
  return IntegrationFlows.from(ApplicationConfiguration.INBOUND_CHANNEL)
         .transform(m -&gt; new StringBuilder((String)m).toString().toUpperCase())
  //                .handle(fileWritingMessageHandler)
      .handle(loggingHandler())
      .handle(kafkaProducerMessageHandler())
      .get();
  }
  
  
  
   //producing channel
  @Bean(name=&quot;kafkaChannel&quot;)
  public DirectChannel kafkaChannel() {
      return new DirectChannel();
  }
  
  @Bean
  public DirectChannel consumingChannel() {
    return new DirectChannel();
  }
  
  
      @Bean
  @ServiceActivator(inputChannel = &quot;kafkaChannel&quot;)
  public MessageHandler kafkaProducerMessageHandler() {
      KafkaProducerMessageHandler&lt;String, String&gt; handler =
              new KafkaProducerMessageHandler&lt;&gt;(kafkaTemplate());
      handler.setTopicExpression(new LiteralExpression(kafkaTopic));
      handler.setMessageKeyExpression(new LiteralExpression(&quot;kafka-integration&quot;));
      return handler;
  }
  
  @Bean
  public KafkaTemplate&lt;String, String&gt; kafkaTemplate() {
    return new KafkaTemplate&lt;&gt;(producerFactory());
  }
  
  @Bean
  public ProducerFactory&lt;String, String&gt; producerFactory() {
    return new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());
  }
  
  @Bean
  public Map&lt;String, Object&gt; producerConfigs() {
    Map&lt;String, Object&gt; properties = new HashMap&lt;&gt;();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // introduce a delay on the send to allow more messages to accumulate
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
  
    return properties;
  }
  
  //consumer configuration....
  @Bean
  public KafkaMessageDrivenChannelAdapter&lt;String, String&gt; kafkaMessageDrivenChannelAdapter() {
    KafkaMessageDrivenChannelAdapter&lt;String, String&gt; kafkaMessageDrivenChannelAdapter =
        new KafkaMessageDrivenChannelAdapter&lt;&gt;(kafkaListenerContainer());
    kafkaMessageDrivenChannelAdapter.setOutputChannel(consumingChannel());
    return kafkaMessageDrivenChannelAdapter;
  }
  
  @SuppressWarnings(&quot;unchecked&quot;)
  @Bean
  public ConcurrentMessageListenerContainer&lt;String, String&gt; kafkaListenerContainer() {
    ContainerProperties containerProps = new ContainerProperties(kafkaTopic); //set topic name
    return (ConcurrentMessageListenerContainer&lt;String, String&gt;) new ConcurrentMessageListenerContainer&lt;&gt;(
        consumerFactory(), containerProps);
  }
  
  @Bean
  public ConsumerFactory&lt;?, ?&gt; consumerFactory() {
    return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
  }
  
  @Bean
  public Map&lt;String, Object&gt; consumerConfigs() {
    Map&lt;String, Object&gt; properties = new HashMap&lt;&gt;();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;helloworld&quot;);
    // automatically reset the offset to the earliest offset
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
  
    return properties;
  }
  </code></pre>
  <p>Here is stacktrace:</p>
  <pre><code> org.springframework.context.ApplicationContextException: Failed to start bean 'kafkaListenerContainer'; nested exception is java.lang.IllegalArgumentException: A org.springframework.kafka.listener.KafkaDataListener implementation must be provided
  at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:50) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:348) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:151) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:114) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:880) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:693) [spring-boot-1.5.8.RELEASE.jar:1.5.8.RELEASE]
  at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:360) [spring-boot-1.5.8.RELEASE.jar:1.5.8.RELEASE]
  at org.springframework.boot.SpringApplication.run(SpringApplication.java:303) [spring-boot-1.5.8.RELEASE.jar:1.5.8.RELEASE]
  at org.springframework.boot.SpringApplication.run(SpringApplication.java:1118) [spring-boot-1.5.8.RELEASE.jar:1.5.8.RELEASE]
  at org.springframework.boot.SpringApplication.run(SpringApplication.java:1107) [spring-boot-1.5.8.RELEASE.jar:1.5.8.RELEASE]
  at com.porterhead.Application.main(Application.java:25) [classes/:na]
   Caused by: java.lang.IllegalArgumentException: A org.springframework.kafka.listener.KafkaDataListener implementation must be provided
  at org.springframework.util.Assert.isTrue(Assert.java:92) ~[spring-core-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  at org.springframework.kafka.listener.AbstractMessageListenerContainer.start(AbstractMessageListenerContainer.java:199) ~[spring-kafka-1.2.2.RELEASE.jar:na]
  at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:175) ~[spring-context-4.3.12.RELEASE.jar:4.3.12.RELEASE]
  ... 12 common frames omitted
  
  
       
  </code></pre>
  <p>I don't know what I am doing wrong. Please do let me know if you need more details on this one.
  Thanks.</p>