class_docstr: Base class for both local and cluster-based batched kafka processing
class_docstr_tok: ["Base","class","for","both","local","and","cluster-based","batched","kafka","processing"]
code: def poll_kafka(self):
      import confluent_kafka as ck
  
      def commit(_part):
          topic, part_no, _, _, offset = _part[1:]
          _tp = ck.TopicPartition(topic, part_no, offset + 1)
          self.consumer.commit(offsets=[_tp], asynchronous=True)
  
      @gen.coroutine
      def checkpoint_emit(_part):
          ref = RefCounter(cb=lambda: commit(_part), loop=self.loop)
          yield self._emit(_part, metadata=[{"ref": ref}])
  
      if self.npartitions is None:
          kafka_cluster_metadata = self.consumer.list_topics(self.topic)
          if self.engine == "cudf":  # pragma: no cover
              self.npartitions = len(
                  kafka_cluster_metadata[self.topic.encode("utf-8")]
              )
          else:
              self.npartitions = len(
                  kafka_cluster_metadata.topics[self.topic].partitions
              )
      self.positions = [0] * self.npartitions
  
      tps = []
      for partition in range(self.npartitions):
          tps.append(ck.TopicPartition(self.topic, partition))
  
      while True:
          try:
              committed = self.consumer.committed(tps, timeout=1)
          except ck.KafkaException:
              pass
          else:
              for tp in committed:
                  self.positions[tp.partition] = tp.offset
              break
  
      while not self.stopped:
          out = []
  
          if self.refresh_partitions:
              kafka_cluster_metadata = self.consumer.list_topics(
                  self.topic
              )
              if self.engine == "cudf":  # pragma: no cover
                  new_partitions = len(
                      kafka_cluster_metadata[self.topic.encode("utf-8")]
                  )
              else:
                  new_partitions = len(
                      kafka_cluster_metadata.topics[
                          self.topic
                      ].partitions
                  )
              if new_partitions > self.npartitions:
                  self.positions.extend(
                      [-1001] * (new_partitions - self.npartitions)
                  )
                  self.npartitions = new_partitions
  
          for partition in range(self.npartitions):
              tp = ck.TopicPartition(self.topic, partition, 0)
              try:
                  low, high = self.consumer.get_watermark_offsets(
                      tp, timeout=0.1
                  )
              except (RuntimeError, ck.KafkaException):
                  continue
              self.started = True
              if "auto.offset.reset" in self.consumer_params.keys():
                  if (
                      self.consumer_params["auto.offset.reset"]
                      == "latest"
                      and self.positions[partition] == -1001
                  ):
                      self.positions[partition] = high
              current_position = self.positions[partition]
              lowest = max(current_position, low)
              if high > lowest + self.max_batch_size:
                  high = lowest + self.max_batch_size
              if high > lowest:
                  out.append(
                      (
                          self.consumer_params,
                          self.topic,
                          partition,
                          self.keys,
                          lowest,
                          high - 1,
                      )
                  )
                  self.positions[partition] = high
          self.consumer_params["auto.offset.reset"] = "earliest"
  
          for part in out:
              yield self.loop.add_callback(checkpoint_emit, part)
  
          else:
              yield gen.sleep(self.poll_interval)
  
imports: import asyncio
  from glob import glob
  import queue
  import os
  import time
  from tornado import gen
  import weakref
  from .core import Stream, convert_interval, RefCounter, sync
label_desc: Use image node_id 2 for calling the FromKafkaBatched obj's underlying member method code with example usage: obj.poll_kafka() without return types