Python Confluent-kafka DeserializingConsumer无法读取Kafka主题消息

调试输出中Consumer Assignment []是核心异常标志：代表消费者没有完成和Kafka集群的消费组协调流程，未被分配任何可消费的分区，此时所有偏移量查询返回的-1001是confluent-kafka内置的无效偏移量常量（OFFSET_INVALID），自然poll方法会持续返回None。
导致该问题的具体原因有4个：

• 分区分配是异步流程：调用subscribe()不会立刻完成分区分配，重平衡、组加入的逻辑是在poll()调用过程中后台执行的。原代码在首次执行poll()前就查询分配状态、偏移量，此时必然返回空和无效值；且总轮询时长过短（5次*3秒=15秒），如果集群网络延迟稍高，重平衡还没完成就已经退出轮询循环。
• 配置逻辑存在隐患：自定义的_popSchemaRegistryParamsFromConfig方法可能误删了连接集群必需的参数（如bootstrap.servers、SASL认证配置等），导致消费者根本连不上集群，永远无法完成分区分配。
• 偏移量配置不符合预期：当前设置auto.offset.reset = 'latest'，代表消费组无已提交偏移量时，仅会消费消费者启动完成后新生产的消息；如果使用固定group.id = "python_example"的消费组之前提交过偏移量，会直接从上次提交的位置开始消费，不会读取历史消息。
• 偏移量查询逻辑错误：代码中硬编码查询partition=0的偏移量，如果主题存在多个分区，或消费者最终没有分配到0号分区，查询结果永远是无效值。

按以下步骤调整即可解决问题：

1. 增加分区分配回调，明确感知分配状态
   订阅主题时传入on_assign回调，只有拿到分配的分区后再查询偏移量、统计消费进度，避免无效查询。
2. 调整轮询逻辑，给重平衡留足时间
   去掉固定5次轮询的退出逻辑，改为统计连续空poll次数，既给集群重平衡留足时间，也能在无新消息时正常退出；同时删除首次poll前的所有偏移量、分配状态查询逻辑。
3. 校验消费者配置完整性
   在_set_consumer_config方法返回配置前打印全量配置，确认bootstrap.servers、所有SASL/安全协议相关的连接参数没有被误删，schema registry相关的参数（schema.registry.url、认证信息等）确实被移除，不会传给消费者导致配置报错。
4. 调整偏移量配置匹配业务需求
   • 如果需要读取主题内所有未消费的历史消息，将auto.offset.reset改为'earliest'；如果仅需要消费启动后的新消息，保留'latest'即可。
   • 如果之前测试时python_example消费组已经提交过错误的偏移量，先通过Kafka CLI删除该消费组，再启动消费者，避免旧偏移量优先生效。
   • 不要硬编码查询0号分区的偏移量，等on_assign回调拿到实际分配到的分区列表后，再查询对应分区的position和committed偏移量。

def get_list_of_unconsumed_msgs(self, topic):
    text_file = open('avro schema file path')
    avro_schema = text_file.read()
    schema_registry_client = SchemaRegistryClient(schema_registry_conf)
    avro_deserializer = AvroDeserializer(schema_registry_client, avro_schema)
    consumer = DeserializingConsumer(self._set_consumer_config(conf, avro_deserializer))

    assigned_partitions = []
    def on_assign(consumer, partitions):
        nonlocal assigned_partitions
        assigned_partitions = partitions
        print(f"成功分配分区: {partitions}")
        # 若需要强制从头消费，可打开以下注释重置分区偏移量
        # for p in partitions:
        #     p.offset = OFFSET_BEGINNING
        # consumer.assign(partitions)

    consumer.subscribe([topic], on_assign=on_assign)
    messages = []
    empty_poll_count = 0
    max_empty_poll = 10  # 连续10次poll无消息则退出，总等待时长30秒足够完成重平衡
    while empty_poll_count < max_empty_poll:
        msg = consumer.poll(3.0)
        if msg is None:
            empty_poll_count += 1
            if assigned_partitions:
                print(f"当前分区消费位置: {consumer.position(assigned_partitions)}")
                print(f"已提交偏移量: {consumer.committed(assigned_partitions)}")
            continue
        if msg.error():
            print(f"消费报错: {msg.error()}")
            continue
        messages.append(msg.value())
        empty_poll_count = 0  # 拿到消息后重置空计数，持续拉取直到无新消息
    consumer.close()
    return messages

注：Kafka CLI可以正常消费是因为CLI默认会生成随机消费组ID，且内置了等待重平衡完成的逻辑，不会出现轮询时长不足、提前查询分配状态的问题。

内容的提问来源于stack exchange，提问作者Ashish Ahuja