confluent-kafka Python 库 consumer.poll(timeout) 无法按预期工作

Question

当我设置 msg = consumer.poll(timeout=10.0) 时，消费者等待 10 秒并按预期返回 None，但是当我将其更改为 msg = consumer.poll(timeout=3600.0) 时，此消费者只需立即返回 None，而不是按预期等待 3600 秒。我在这里错过了什么吗？如果需要，这里是完整的代码。

running = True
conf = {'bootstrap.servers': bootstrap_servers,
        'group.id': 'foo',
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,
        'on_commit': commit_completed}
consumer = Consumer(conf)


def msg_process(msg):
    print(f"key: {msg.key().decode('utf-8')}, value: {msg.value().decode('utf-8')}")


def basic_consume_loop(consumer, topics):
    try:
        consumer.subscribe(topics)

        msg_count = 0
        while running:
            msg = consumer.poll(timeout=3600.0)
            if msg is None:
                print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: no new message")
                continue

            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # End of partition event
                    print('%% %s [%d] reached end at offset %d\n' %
                          (msg.topic(), msg.partition(), msg.offset()))
                elif msg.error():
                    raise KafkaException(msg.error())
            else:
                # consumer.commit(async=False)
                msg_process(msg)
                msg_count += 1
                if msg_count % MIN_COMMIT_COUNT == 0:
                    consumer.commit(async=True)
    finally:
        # Close down consumer to commit final offsets.
        consumer.close()


def shutdown():
    running = False


basic_consume_loop(consumer, [topic_user])

Answer 1

可能是因为fetch.max.wait.ms 设置低于poll() 中传递的设置。