def _consumer_loop(kafka_client_conn, topic_name=TOPIC_NAME, max_msgs=3, max_runtime_seconds=10) -> List[KMessage]:
    kafka_client_conn.subscribe([topic_name])
    messages = []
    consumed_count = 0
    start_time = datetime.utcnow()
    max_runtime_duration = timedelta(seconds=max_runtime_seconds)
    while True:
        if consumed_count == max_msgs:
            break
        if datetime.utcnow() > start_time + max_runtime_duration:
            print("max polling time limit reached")
            break
        msg = kafka_client_conn.poll(1)
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())
        else:
            print_msg(msg)
            messages.append(msg)
            consumed_count += 1
    return messages

@op(
    required_resource_keys={"kafka"},
    ins={
      "topic_name": In(),
      "max_msgs": In(default_value=100),
      "max_runtime_seconds": In(default_value=45),
    },
)
def consume_topic(context, topic_name, max_msgs, max_runtime_seconds) -> List[KMessage]:
    with context.resources.kafka.connection() as conn:
      return _consumer_loop(conn, topic_name, max_msgs, max_runtime_seconds)

KMessage

confluent_kafka.Message

kafka_client_conn.poll(1)

while True

poll

timeout (float) – Maximum time to block waiting for message