title: Kafka Consumer is not receiving message in Spring Boot
body: <p>My spring/java consumer is not able to access the message produced by producer. However, when I run the consumer from console/terminal it is able to receive the message produced by spring/java producer.</p>
  
  <p>Consumer Configuration :</p>
  
  <pre><code>@Component
  @ConfigurationProperties(prefix="kafka.consumer")
  public class KafkaConsumerProperties {
  
      private String bootstrap;
      private String group;
      private String topic;
  
      public String getBootstrap() {
          return bootstrap;
      }
  
      public void setBootstrap(String bootstrap) {
          this.bootstrap = bootstrap;
      }
  
      public String getGroup() {
          return group;
      }
  
      public void setGroup(String group) {
          this.group = group;
      }
  
      public String getTopic() {
          return topic;
      }
  
      public void setTopic(String topic) {
          this.topic = topic;
      }
  }
  </code></pre>
  
  <p>Listener Configuration : </p>
  
  <pre><code>@Configuration
  @EnableKafka
  public class KafkaListenerConfig {
  
      @Autowired
      private KafkaConsumerProperties kafkaConsumerProperties;
  
      @Bean
      public Map&lt;String, Object&gt; getConsumerProperties() {
          Map&lt;String, Object&gt; properties = new HashMap&lt;&gt;();
          properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProperties.getBootstrap());
          properties.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperties.getGroup());
          properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
          properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
          properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
          return properties;
      }
  
      @Bean
      public Deserializer stringKeyDeserializer() {
          return new StringDeserializer();
      }
  
      @Bean
      public Deserializer transactionJsonValueDeserializer() {
          return new JsonDeserializer(Transaction.class);
      }
  
      @Bean
      public ConsumerFactory&lt;String, Transaction&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(getConsumerProperties(), stringKeyDeserializer(), transactionJsonValueDeserializer());
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, Transaction&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, Transaction&gt; factory =
                  new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConcurrency(1);
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  
  }
  </code></pre>
  
  <p>Kafka Listener :</p>
  
  <pre><code>@Service
  public class TransactionConsumer {
      private static final Logger LOGGER = LoggerFactory.getLogger(Transaction.class);
  
      @KafkaListener(topics={"transactions"}, containerFactory = "kafkaListenerContainerFactory")
      public void onReceive(Transaction transaction) {
          LOGGER.info("transaction = {}",transaction);
      }
  }
  </code></pre>
  
  <p>Consumer Application :</p>
  
  <pre><code>@SpringBootApplication
  public class ConsumerApplication {
  
      public static void main(String[] args) {
          SpringApplication.run(ConsumerApplication.class, args);
      }
  
  }
  
  </code></pre>
  
  <p><strong>TEST CASE 1 : PASS</strong>
  I started my spring/java producer and run the consumer from console. When I produce message form producer my console consumer is able to access the  message.</p>
  
  <p><strong>TEST CASE 2 : FAILED</strong>
  I started my spring/java consumer and run the producer from console. When I produce message form console producer my spring/java consumer is not able to access the  message.</p>
  
  <p><strong>TEST CASE 3 : FAILED</strong>
  I started my spring/java consumer and run the spring/java producer. When I produce message form spring/java producer my spring/java consumer is not able to access the  message.</p>
  
  <p><strong>Question</strong></p>
  
  <ol>
  <li><p>Is there anything wrong in my consumer code ?</p></li>
  <li><p>Am I missing any configuration for my kafka-listener?</p></li>
  <li><p>Do I need to explicitly run the listener? (I don't think so since I can see on the terminal log connecting to topic, still I am not sure)</p></li>
  </ol>