title: Kafka Consumer empty records
body: <p>There are a lot of questions about this topic, however, this is NOT a duplicate question!</p>
  <p>The problem I'm facing is that I tried to set up a SpringBoot project with Java 14 and Kafka 2.5.0 and my <strong>Consumer returns an empty list of records</strong>.
  Most answers here indicate some forgotten properties, to <a href="https://stackoverflow.com/questions/54988037/kafks-consumer-poll-returns-no-data">poll frequently</a> or to set the <a href="https://stackoverflow.com/questions/53706400/kafka-consumer-returns-no-records">offset mode to earliest</a>.</p>
  <p>I can't see any <em>logical</em> difference from <a href="https://docs.confluent.io/3.0.0/clients/consumer.html" rel="nofollow noreferrer">docs.confluent.io</a>, eventhough my config-settings seem unconventional (see my setting of the <a href="https://stackoverflow.com/questions/45756543/kafka-login-module-not-specified-in-jaas-config">jaas.conf</a> in the snippet below).</p>
  <pre><code>@EnableKafka
  @Configuration
  public class KafkaConfig {
  
      @Bean
      public KafkaConsumer&lt;Long, MyClass&gt; consumerConfigs() {
          Properties config = new Properties();
  
          config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, &quot;SASL_SSL&quot;);
          config.put(SaslConfigs.SASL_MECHANISM, &quot;PLAIN&quot;);
  
          config.put(ConsumerConfig.CLIENT_ID_CONFIG, &quot;serviceName&quot;);
          config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, &quot;confluent-cloud&quot;);
          config.put(ConsumerConfig.GROUP_ID_CONFIG, &quot;serviceName&quot;);
          config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
          config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MyClass.class);
          config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
          config.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 300_000);
  
          System.setProperty(&quot;java.security.auth.login.config&quot;, &quot;.\\src\\main\\resources\\jaas.conf&quot;);
  
          return new KafkaConsumer&lt;&gt;(config);
      }
  }
  </code></pre>
  <p>This works however. I receive no exceptions (Kafka, or otherwise), and the connection is made.</p>
  <pre><code>// jaas.conf-file
  KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  serviceName=&quot;serviceName&quot;
  username=&quot;username&quot;
  password=&quot;password&quot;;
  };
  </code></pre>
  <p>Here is where I'm actually polling:</p>
  <pre><code>try {
              KafkaConsumer&lt;Long, MyClass&gt; consumer = kafkaConfig.consumerConfigs();
              consumer.subscribe(Collections.singletonList(inputTopic));
  
              int count = 0;
              Long start = System.currentTimeMillis();
              Long end = System.currentTimeMillis();
  
              while (end - start &lt; 900_000) { 
                  // boolean would be set to true in production
                  ConsumerRecords&lt;Long, MyClass&gt; records = consumer.poll(Duration.ofMillis(1000));
                  records.forEach(record -&gt; {
                      MyOtherClass result = myOwnKafkaService.getSomethingFromRecord(record.key(), record.value());
                      System.out.println(result);
                  });
                 
                  consumer.commitSync();
  
                  System.out.println(&quot;visualize number of loops made: &quot; + ++count);
                  end = System.currentTimeMillis();
              }
          } catch (KafkaException e) {
              e.printStackTrace();
          } catch (Exception e) {
              System.out.println(e.getMessage());
          }
  </code></pre>
  <p>I added the prints and other <em>clutter</em> in order to try and find the issue. I run my program in debug-mode and placed the breakpoint on this line:</p>
  <pre><code>MyOtherClass result = myOwnKafkaService.getSomethingFromRecord(record.key(), record.value());
  </code></pre>
  <p>As a result, I see a printed line with the count every second, as one would expect. But since my Consumer returns no records, it never enters the <code>forEach</code> and thus never triggers my breakpoint.</p>
  <p>I can definitely see my topic in the cloud, with two partitions. Messages are generated in a steady stream, so I know I should be able to pick up something.</p>
  <p>I know that it would take some time to connect to the cluster, but with current time being set at a quarter hour, I should receive at least something, right? As an alternative I tried switching the <code>consumer.subscribe()</code> to the <code>consumer.assign()</code> method were I specified my TopicPartition, setting the consumer to <code>consumer.seekToBeginning()</code>. It ran fine, but also returned nothing.</p>
  <p>One other thing that's not found in the most common examples, is that I use my own classes. So instead of <code>KafkaConsumer&lt;String, String&gt;</code>, I implemented custom (de)serializers according to <a href="https://dzone.com/articles/kafka-sending-object-as-a-message" rel="nofollow noreferrer">this tutorial</a>.</p>
  <p>Could it be my configuration settings? Something wrong with the poll timeout? The (de)serialization, or something else entirely? I genuinely can't pinpoint any reason as to why I'm getting zero records. Any feedback would be greatly appreciated!</p>