title: Spring-Kafka never hits Kafka listener
body: <p>I'm building a simple consumer and decided to migrate over to Springboot and the springkafka depenedency to make the process simple.</p>
  <p>The environment/config values are interpolated from a mix of environment + properties file in Spring. All the values look fine. I have data on the topic and I can consume it from something like kafkacat or console consumer.</p>
  <h3>Problem</h3>
  <p>method with <code>@kafkalistener</code> never gets invoked. I debug the code and it just says &quot;build successful&quot;. It will call the setup, the configuration and the factories, but the consumer listener just doesn't get invoked..</p>
  <h3>Attempt To Resolve</h3>
  <p><strong>Consumer</strong></p>
  <pre><code>@Service
  public class CyclicConsumerService {
      @KafkaListener(topics = &quot;my-topic&quot;, groupId = &quot;test-group-id&quot;)
      public void avroConsumer(GenericRecord message) {
          System.out.println(&quot;&quot;);
      }
  }
  </code></pre>
  <p><strong>Configuration Setup</strong></p>
  <pre><code>
  ...
  
  @EnableKafka
  @Configuration
  public class KafkaConsumerConfig {
  
  ...
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String,
              GenericRecord&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, GenericRecord&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  
      @Bean
      public ConsumerFactory&lt;String, GenericRecord&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(retrieveConsumerConfigs());
      }
  </code></pre>
  <h3>retrieveConsumerConfiguration method</h3>
  <pre><code>    public Map&lt;String, Object&gt; retrieveConsumerConfigs() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, retrieveArgument(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
          props.put(ConsumerConfig.GROUP_ID_CONFIG, retrieveArgument(ConsumerConfig.GROUP_ID_CONFIG));
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, AUTO_OFFSET_RESET_CONFIG);
          props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, ISOLATION_LEVEL_CONFIG);
          props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, SPECIFIC_AVRO_READER_CONFIG);
          props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, retrieveArgument(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG));
          props.put(AbstractKafkaAvroSerDeConfig.AUTO_REGISTER_SCHEMAS, AUTO_REGISTER_SCHEMAS);
          addSaslProperties(props);
          props.put(AbstractKafkaAvroSerDeConfig.VALUE_SUBJECT_NAME_STRATEGY, TopicNameStrategy.class.getName());
          return props;
      }
  </code></pre>
  <pre><code>    private String retrieveArgument(String argName) {
          return Objects.requireNonNull(env.getProperty(argName),
                  &quot;Unable to resolve non-existent Kafka environment variable&quot;);
      }
  </code></pre>
  <h1>Logs</h1>
  <pre><code>: [Consumer clientId=consumer-1, groupId=test-group-id] Initializing the Kafka consumer
  : Successfully logged in.
  : Created SSL context with keystore null, truststore null
  : Added sensor with name fetch-throttle-time
  Added sensor with name connections-closed:
  Added sensor with name connections-created:
  Added sensor with name successful-authentication:
  Added sensor with name successful-reauthentication:
  Added sensor with name successful-authentication-no-reauth:
  Added sensor with name failed-authentication:
  Added sensor with name failed-reauthentication:
  Added sensor with name reauthentication-latency:
  Added sensor with name bytes-sent-received:
  Added sensor with name bytes-sent:
  Added sensor with name bytes-received:
  Added sensor with name select-time:
  Added sensor with name io-time:
  Added sensor with name heartbeat-latency
  Added sensor with name join-latency
  Added sensor with name sync-latency
  Added sensor with name commit-latency
  Added sensor with name bytes-fetched
  Added sensor with name records-fetched
  Added sensor with name fetch-latency
  Added sensor with name records-lag
  Added sensor with name records-lead
  Kafka version: 2.3.1
  Kafka commitId: 18a913733fb71c01
  Kafka startTimeMs: 1606869543614
  [Consumer clientId=consumer-1, groupId=test-group-id] Kafka consumer initialized
  [Consumer clientId=consumer-1, groupId=test-group-id] Subscribed to topic(s): my-topic
  Initializing ExecutorService
  [Consumer clientId=consumer-1, groupId=test-group-id] Sending FindCoordinator request to broker xxxx:9093 (id: -1 rack: null)
  [Consumer clientId=consumer-1, groupId=test-group-id] Received user wakeup
  [Consumer clientId=consumer-1, groupId=test-group-id] Initiating connection to node xxxx:9093 (id: -1 rack: null) using address xxxx
  Set SASL client state to SEND_APIVERSIONS_REQUEST
  Creating SaslClient: client=null;service=kafka;serviceHostname=xxxx;mechs=[SCRAM-SHA-512]
  Setting SASL/SCRAM_SHA_512 client state to SEND_CLIENT_FIRST_MESSAGE
  [Consumer clientId=consumer-1, groupId=test-group-id] Raising WakeupException in response to user wakeup
  [Consumer clientId=consumer-1, groupId=test-group-id] Unsubscribed all topics or patterns and assigned partitions
  Shutting down ExecutorService
  Removed sensor with name connections-closed:
  Removed sensor with name connections-created:
  Removed sensor with name successful-authentication:
  Removed sensor with name successful-reauthentication:
  Removed sensor with name successful-authentication-no-reauth:
  Removed sensor with name failed-authentication:
  Removed sensor with name failed-reauthentication:
  Removed sensor with name reauthentication-latency:
  Removed sensor with name bytes-sent-received:
  Removed sensor with name bytes-sent:
  Removed sensor with name bytes-received:
  Removed sensor with name select-time:
  Removed sensor with name io-time:
  [Consumer clientId=consumer-1, groupId=test-group-id] Kafka consumer has been closed
  test-group-id: Consumer stopped
  </code></pre>