java - 如何在 Kafka 中为消费者发送 OffsetCommitRequest？

Question

我正在尝试在 Kafka 消费者 API 中使用 OffsetCommitRequest 版本 0.9，它包含在以下包中:org.apache.kafka.common.requests.OffsetCommitRequest

如何发送这个请求？使用它的理想方法是什么？我想在 Kafka 本身中提交偏移量。我没有找到任何与 0.9 版本相关的文档。其中大部分可用于 0.8.x

此外，此请求的构造函数采用生成 ID、成员 ID 和保留时间。这些字段是什么？

Answer 1

如果你想手动提交偏移量，也许你应该设置消费者属性

enable.auto.commit=false

并使用 kafka 消费者的 commitSync() 或 commitAsync() 方法。例如，您可以在处理完所有 ConsumerRecords 后调用 commitSync()。或者，即使在每次收到 ConsumerRecord 之后，您也可以只提交您想要的 TopicPartition。像这样:

Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
offsetMap.put(new TopicPartition(someTopic, somePartition), new OffsetAndMetadata(someOffset));
kafkaConsumer.commitSync(offsetMap);