对于kafka streams，如果我们将commit.internal.ms config设置为long.max\u值，这将有效地避免kafka streams自动提交，而只在调用较低级别的处理器api context.commit（）后提交，我们可以控制是否提交？如果是，那么在出现问题或消息包含格式错误的情况下，我们如何从主题中删除这些消息？
资源：https://docs.confluent.io/current/streams/index.html
对于kafka streams，如果我们将commit.internal.ms config设置为long.max\u值，这将有效地避免kafka streams自动提交，而只在调用较低级别的处理器api context.commit（）后提交，我们可以控制是否提交？如果是，那么在出现问题或消息包含格式错误的情况下，我们如何从主题中删除这些消息？
资源：https://docs.confluent.io/current/streams/index.html
1条答案
按热度按时间c3frrgcw1#
无法从主题中删除邮件。。。Kafka主题是只附加不可变的日志。
另外请注意，如果您提交了一条消息，它也不会从主题中删除。kafka不是一个消息队列——提交偏移量只意味着跟踪日志中的当前进度（即偏移量），但它不会删除任何消息(你可以把它想象成一个“光标”。）
因此，如果消息格式不正确，您不想处理它，您可以跳过它。实际上，没有理由为这种情况设置一个大的提交间隔并进行手动提交。