confluent_kafka：如何在读取数据之前可靠地寻找（避免错误状态）

Answer 1

从https://github.com/confluentinc/confluent-kafka-python/issues/11#issuecomment-230089107 看来，一种解决方案是为 Consumer.subscribe 指定一个 on_assign 回调，然后调用 Consumer.assign里面on_assign 回调，例如：

def on_assign_callback(
    consumer,
    partitions,
):
    """Modify assigned partitions to read up to MAX_HISTORY old messages"""
    for partition in partitions:
        min_offset, max_offset = consumer.get_watermark_offsets(partition)
        desired_offset = max_offset - MAX_HISTORY
        if desired_offset <= min_offset:
            desired_offset = OFFSET_BEGINNING
        partition.offset = desired_offset
    consumer.assign(partitions)

细微之处：

• 回调必须分配所有主题分区，即使您不想要某些主题的历史数据。
• 使用选项"auto.offset.reset": "earliest" 构造消费者。这样，如果代理在 on_assign 回调运行时丢弃数据，删除指定偏移量的数据，消费者将从头开始读取。

Answer 2

我找到了你的帖子，因为我遇到了类似的挑战，并且有一个适合我的解决方案。这不是基于水印，而是基于提交的偏移量：

consumer.subscribe([topic_name])
messages = []
seeked = False
while True:
    msg = consumer.poll(5)
    tps_comm = consumer.committed(consumer.assignment())
    if len(tps_comm) == 0:
        continue
    else:
        tp = tps_comm[0]
        if tp.offset == OFFSET_INVALID and not seeked:
            tp.offset = OFFSET_BEGINNING
            consumer.seek(tp)
            seeked = True
    if msg is None:
        continue
    elif msg.error():
        raise Exception(msg.error())
    else:
        print(f"got message at offset: {msg.offset()}")
        messages.append(msg)    

我已经从我的实际解决方案中省略了max_messages 和循环超时逻辑，以支持上面共享的更简单的代码示例，该示例缺少任何循环外的break。

我收集到的是，当消费者连接到代理并订阅一个主题时，它不会立即被分配一个主题分区，如果你的poll 调用超时太短，甚至不会很快。在测试中，第一次尝试几秒钟可能就足够了。但是，通过尝试直到主题分区分配作为非空列表返回，然后检查组分区分配的提交偏移量，如果需要，我的消费者可以决定寻找主题分区的开头，否则，正常情况是poll 将开始为组主题分区分配返回任何新的未提交消息。

由于我的消费者在提交消息之前需要对消息执行其他操作，因此我将 "enable.auto.commit": False 作为消费者配置设置。这是接收消息并在处理后提交其偏移量的不相交代码：

tp_offsets = []
for msg in messages:
    tp = TopicPartition(
        topic=msg.topic(),
        partition=msg.partition(),
        offset=msg.offset() + 1,
    )
    tp_offsets.append(tp)
consumer.commit(offsets=tp_offsets)        

注意：如果您订阅多个主题，上面的代码可能需要重新编写。