是否有可能将偏移量提交到具有多个分区的 Kafka 主题，因此可以将偏移量 1 提交到分区 1，将偏移量 2 提交到 p2 等等？
编辑：
是的，有可能：
consumer = KafkaConsumer()
topicpartitions = [TopicPartition('topicname', partitionId) for partitionId in consumer.partitions_for_topic('topicname')]
consumer.assign(topicpartitions)
for tp in topicpartitions:
consumer.commit({tp: OffsetAndMetadata(1000, None)})
for msg in consumer:
#do whatever
【问题讨论】：
标签： python apache-kafka commit
Kafka 偏移量始终是每个分区的。我的意思是，如果您的主题有 2 个分区，则 p0 中的消息将从偏移量 0 开始，并为每条新消息增加我的 1。 p1 相似度中的消息从偏移量 0 开始并增加 1。
因此，如果您发布了两条消息（没有密钥），一条将进入分区 0，偏移量为 0，另一条进入分区 1，偏移量为 0。
现在，如果另一个应用程序正在使用该主题并提交其偏移量，那么它将向 __consumer_offsets 主题生成消息，其中包括其 group.id、主题、分区号和偏移量。例如，类似 {"myconsumerid","mytopic",P0,1} 和 {"myconsumerid","mytopic",P1,1}。
如果应用停止并且一个或两个其他使用者从相同的 group.id 开始，他们将从分配给他们的分区的最后提交的偏移量继续。
如果您想将消费者偏移量重新定位到任何其他位置，您可以使用 0.11 Kafka 工具更改组的已提交偏移量
bin/kafka-consumer-groups.sh --reset-offsets
如果你给它正确的标志，这个工具将允许你独立设置每个分区的偏移量。
如果您愿意，可以从 Python 程序中调用此工具。消费者组中的所有现有消费者应首先关闭，否则他们可能会覆盖偏移量。
如果你想编写这个工具的 Python 版本而不是运行现有的 CLI 命令，你需要找到一个支持 seek() 的 Python 客户端，或者你可以将偏移量更改为你想要的，然后提交当消费应用程序重新启动时，它们处于该位置。另一种方法是放弃动态分区分配并手动分配（）要更改的分区并将偏移量提交到分配的列表。您不能在同一个应用程序中同时使用动态管理的分区订阅和手动分配的分区。
您还需要确保在这些分区上使用相同使用者组的所有其他使用者都已关闭，否则一旦其他使用者自动提交或手动提交偏移量，已提交的偏移量将被其他使用者覆盖你刚刚设置好了。
【讨论】：