title: How to get messages from Kafka Consumer one by one in java?
body: <p>I'm using Apache Kafka API and trying to get only one message at a time. I'm only writing to one topic. I can send and receive messages by having a pop UI screen with a textbox. I input a string in the textbox and click &quot;send.&quot; I can send as many messages as I want. Let's say I send 3 messages and my 3 messages were &quot;hi,&quot; &quot;lol,&quot; &quot;bye.&quot; There is also a &quot;receive&quot; button. Right now, using the traditional code found in <a href="https://www.tutorialspoint.com/apache_kafka/apache_kafka_consumer_group_example.htm#:%7E:text=Kafka%20assigns%20the%20partitions%20of,were%20stored%20in%20the%20log." rel="nofollow noreferrer">TutorialsPoint</a>, I get all 3 messages (hi, lol, bye) at once printed on the console when I click on the receive button. However, I want only want one message to be printed at a time when I click &quot;receive&quot; on the UI. For example, the first time I hit the receive button, it would print &quot;hi,&quot; the second time would be &quot;lol,&quot; and the third time would be &quot;bye.&quot; I am new to Kafka and am confused on how to do this. I tried removing both the loops from the code so it just has</p>
  <pre><code>ConsumerRecords&lt;String, String&gt; records = consumer.poll(100);
  System.out.printf(records.iterator().next().value());
  </code></pre>
  <p>If I just have those 2 lines of code, the first time I hit the receive button, it would print &quot;hi&quot; but the second time I press it,  get the message &quot;attempt to heartbeat failed since group is rebalancing kafka.&quot; I'm getting errors when I set the max.poll.records = 1 too as I want all my messages <em>eventually</em> but just one of them needs to be logged to the console when the receive button is pressed. The next time, the next message in the topic not logged would be logged.</p>
  <p>Hope that makes sense!
  Appreciate any help!
  Thanks in advance!</p>
  <p>EDIT:
  New code after including queues and also so we can alternate between sending and receiving messages &amp; update the queue whenever there is a new message:</p>
  <pre><code>        if (payloadQueue.isEmpty()){
              ConsumerRecords&lt;String, String&gt; records = consumer.poll(Duration.ofMillis(500));
              if (records.isEmpty()) {
                  log.info(&quot;No More Records to Add&quot;);
                  consumer.close();
              }
              else {
                  records.iterator().forEachRemaining(record -&gt; {
                      log.info(&quot;RECORD: &quot; + record);
                      payloadQueue.offer(record);
                  });
                  payload = payloadQueue.poll().value();
                  log.info(&quot;Received event from KAFKA on subject {} with payload \&quot;{}\&quot;&quot;, subject, payload);
              }
          }
          else {
              payload = payloadQueue.poll().value();
              log.info(&quot;Received event from KAFKA on subject {} with payload \&quot;{}\&quot;&quot;, subject, payload);
          }
  </code></pre>