Kafka Streams-处理器上下文提交 [英] Kafka Streams - Processor context commit

问题描述

我们是否应该亲自在 Processor 实现中调用 processorContext.commit()?我的意思是在计划的 Punctuator 实现内部或 process 方法内部调用 commit 方法.

should we ever invoke processorContext.commit() in Processor implementation by ourselves? I mean invoking commit method inside scheduled Punctuator implementation or inside process method.

我们应该在哪些用例中这样做，并且我们根本需要吗?这个问题与具有 transform()的Kafka DSL和Processor API都有关.

in which use cases should we do that, and do we need that at all? the question relates to both Kafka DSL with transform() and Processor API.

似乎Kafka Streams自己处理它，并且调用 processorContext.commit()也不保证它会立即完成.

seems Kafka Streams handles it by itself, also invoking processorContext.commit() does not guarantee that it will be done immediately.

推荐答案

从处理器或标点调用 commit()是可以的，这就是提供此API的原因

It is ok to call commit() -- either from the Processor or from a Punctuation -- that's why this API is offered.

虽然Kafka Streams以固定(可配置)的时间间隔提交，但是您可以在使用它时请求中间提交.一个示例用例是，您通常执行廉价的计算，但是有时您执行一些昂贵的操作，并希望在此操作之后尽快提交，而不是等待下一个提交间隔(以减少在执行昂贵的操作之后执行失败的可能性，并且下一个提交间隔).另一个用例是，如果将提交间隔设置为 MAX_VALUE ，则哪些有效地禁用"了常规提交并根据您的业务逻辑决定何时提交.

While Kafka Streams commits on a regular (configurable) interval, you can request intermediate commits when you use it. One example use case would be, that you usually do cheap computation, but sometimes you do something expensive and want to commit asap after this operation instead of waiting for the next commit interval (to reduce the likelihood of a failure after the expensive operation and the next commit interval). Another use case would be, if you set the commit interval to MAX_VALUE what effectively "disables" regular commits and to decide when to commit base on your business logic.

我想，对于大多数用例而言，调用 commit()并不是必需的.

I guess, that calling commit() is not necessary for most use cases thought.

这篇关于Kafka Streams-处理器上下文提交的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！