title: How to flush data batch in Kafka Consumer when there are no more records in topic
body: <p>Consider this Kafka consumer that receives data from topic, buffers it into PreparedStatement and when 100K records is batched, it issues INSERT query into DB.</p>
  
  <p>This works well just until the data are still incoming. However when for example 20K records is buffered and there are no more records incoming, it still waits for more 80K records until in <em>flushes</em> the statement. But <strong>I'd like to flush those 20K</strong> if stalled after some time. How can I do that? I don't see any way how to hook onto it.</p>
  
  <p><em>For example in PHP that uses php-rdkafka extension based on librdkafka I get <code>RD_KAFKA_RESP_ERR__PARTITION_EOF</code> when end of partition is reached so it's pretty easy to hook buffer flush when that happens.</em></p>
  
  <p><em>I tried to simplify the code so only significant parts remained</em></p>
  
  <pre class="lang-java prettyprint-override"><code>public class TestConsumer {
  
      private final Connection connection;
      private final CountDownLatch shutdownLatch;
      private final KafkaConsumer&lt;String, Message&gt; consumer;
      private int processedCount = 0;
  
      public TestConsumer(Connection connection) {
          this.connection = connection;
          this.consumer = new KafkaConsumer&lt;&gt;(getConfig(), new StringDeserializer(), new ProtoDeserializer&lt;&gt;(Message.parser()));
          this.shutdownLatch = new CountDownLatch(1);
      }
  
      public void execute() {
          PreparedStatement statement;
          try {
              statement = getPreparedStatement();
          } catch (SQLException e) {
              throw new RuntimeException(e);
          }
  
          Runtime.getRuntime().addShutdownHook(new Thread(() -&gt; {
              commit(statement);
  
              consumer.wakeup();
          }));
  
          consumer.subscribe(Collections.singletonList("source.topic"));
  
          try {
              while (true) {
                  ConsumerRecords&lt;String, Message&gt; records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
  
                  records.forEach(record -&gt; {
                      Message message = record.value();
                      try {
                          fillBatch(statement, message);
                          statement.addBatch();
                      } catch (SQLException e) {
                          throw new RuntimeException(e);
                      }
                  });
  
                  processedCount += records.count();
  
                  if (processedCount &gt; 100000) {
                      commit(statement);
                  }
              }
          } catch (WakeupException e) {
              // ignore, we're closing
          } finally {
              consumer.close();
              shutdownLatch.countDown();
          }
      }
  
      private void commit(PreparedStatement statement) {
          try {
              statement.executeBatch();
              consumer.commitSync();
              processedCount = 0;
          } catch (SQLException e) {
              throw new RuntimeException(e);
          }
      }
  
  
      protected void fillBatch(PreparedStatement statement, Message message) throws SQLException {
          try {
              statement.setTimestamp(1, new Timestamp(message.getTime() * 1000L));
          } catch (UnknownHostException e) {
              throw new RuntimeException(e);
          }
      }
  </code></pre>