title: Apache Beam ReadFromKafka vs KafkaConsume
body: <p>I'm working with a simple Apache Beam pipeline consisting of reading from an unbounded Kafka topic and printing the values out. I have two flavors of this. This is done via the Flink Runner.</p>
  <h3>Version 1</h3>
  <pre class="lang-py prettyprint-override"><code>  with beam.Pipeline(options=beam_options) as p:
          (p
           | &quot;Read from Kafka topic&quot; &gt;&gt; ReadFromKafka(
                      consumer_config=consumer_config,
                      topics=[producer_topic])
           | 'log' &gt;&gt; beam.ParDo(LogData())
  </code></pre>
  <p>This one uses <code>from apache_beam.io.kafka import ReadFromKafka</code> (i.e. the default implementation that comes with Apache Beam).</p>
  <h3>Version 2</h3>
  <pre class="lang-py prettyprint-override"><code>   with beam.Pipeline(options=beam_options) as p:
          (p
           | &quot;Read from Kafka topic (KafkaConsumer)&quot; &gt;&gt; KafkaConsume(
                      consumer_config={
                          &quot;topic&quot;: producer_topic,
                          'auto_offset_reset': 'earliest',
                          &quot;group_id&quot;: 'transaction_classification',
                          &quot;bootstrap_servers&quot;: servers,
                      })
  </code></pre>
  <p>This one is using Beam nuggets:</p>
  <pre><code>from beam_nuggets.io.kafkaio import KafkaConsume
  </code></pre>
  <p>I have configured the Kafka producer to produce an element every 1 second.</p>
  <p>What I've observed is that when I consume from <code>ReadFromKafka</code> (version 1), the elements get produced around 4-6 seconds apart, and are batched together.</p>
  <p>On the other hand, if I tried the same thing with <code>KafkaConsume</code> (version 2), then I get elements as they are produced (i.e. every second), which is exactly the behavior I expected.</p>
  <p>I have tried to make the <code>consumer_config</code> to be the same for both, but it doesn't seem to have any effect on version 1.</p>
  <p>Now, I would like to stick to version 1 because that gives me proper metrics in the Flink UI, while version 2 works better, I don't get any metrics in Flink (everything is reported as 0 bytes received / 0 records received).</p>
  <p>Does anyone have any ideas?</p>