apache-kafka - 如何使用Kafka Stream手动提交？

Question

有没有办法用Kafka Stream手动提交？

通常使用KafkaConsumer时，我会执行以下操作：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records){
       // process records
    }
   consumer.commitAsync();
}

我在哪里手动调用commit。我没有为 KStream使用类似的API。

Answer 1

提交是由Streams内部自动处理的，因此通常没有理由手动提交。请注意，Streams的处理方式与使用者自动提交的处理方式不同-实际上，内部使用的使用者使用自动提交功能是禁用的，Streams是“手动”管理提交的。原因是，提交只能在处理过程中的某些点发生，以确保不会丢失任何数据（在更新状态和刷新结果方面存在许多内部依赖性）。

对于更频繁的提交，可以通过StreamsConfig参数commit.interval.ms缩短提交间隔。

但是，可以通过低级Processor API间接进行手动提交。您可以使用通过context方法提供的init()对象来调用context#commit()。请注意，这只是尽快提交的“请求流”，而不是直接发出提交。