(kafka_params, topic, partition, keys, low, high, timeout=None)
return out
Fetch a batch of kafka messages (keys & values) in given topic/partition

    This will block until messages are available, or timeout is reached.
Fetch a batch of kafka messages (keys & values) in given topic/partition
Fetch a batch of kafka messages ( keys & values ) in given topic / partition
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
def get_message_batch ( kafka_params , topic , partition , keys , low , high , timeout = None ) : 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