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 start(self):
      import confluent_kafka as ck
  
      if self.engine == "cudf":  # pragma: no cover
          from custreamz import kafka
  
      if self.stopped:
          if self.engine == "cudf":  # pragma: no cover
              self.consumer = kafka.Consumer(self.consumer_params)
          else:
              self.consumer = ck.Consumer(self.consumer_params)
          weakref.finalize(
              self,
              lambda consumer=self.consumer: _close_consumer(consumer),
          )
          self.stopped = False
          tp = ck.TopicPartition(self.topic, 0, 0)
  
          # blocks for consumer thread to come up and invoke poll to establish
          # connection with broker to fetch oauth token for kafka
          self.consumer.poll(timeout=1)
          self.consumer.get_watermark_offsets(tp)
          self.loop.add_callback(self.poll_kafka)
  
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 3 for calling the FromKafkaBatched obj's underlying member method code with example usage: obj.start() without return types