code: def get_message_batch(
      kafka_params, topic, partition, keys, low, high, timeout=None
  ):
      """Fetch a batch of kafka messages (keys & values) in given topic/partition
  
      This will block until messages are available, or timeout is reached.
      """
      import confluent_kafka as ck
  
      t0 = time.time()
      consumer = ck.Consumer(kafka_params)
      tp = ck.TopicPartition(topic, partition, low)
      consumer.assign([tp])
      out = []
      try:
          while True:
              msg = consumer.poll(0)
              if msg and msg.value() and msg.error() is None:
                  if high >= msg.offset():
                      if keys:
                          out.append(
                              {"key": msg.key(), "value": msg.value()}
                          )
                      else:
                          out.append(msg.value())
                  if high <= msg.offset():
                      break
              else:
                  time.sleep(0.1)
                  if timeout is not None and time.time() - t0 > timeout:
                      break
      finally:
          consumer.close()
      return out
  
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 5 for calling a global function with example usage: get_message_batch(kafka_params, topic, partition, keys, low, high, timeout) and returns: out