如何解决如何使用 PartitionAssignmentHandler 刷新 Kafka 偏移量
我们正在使用类似于以下示例的内容来处理我们的 Kafka 事件:
Consumer
.sourceWithOffsetContext(consumerSettings,Subscriptions.topics(topic))
.mapAsync(10) { record =>
business(record.key,record.value)
}
.via(Committer.flowWithOffsetContext(committerDefaults))
.toMat(Sink.seq)(DrainingControl.apply)
.run()
我们的消费者进程经常自动缩放,因此我们希望利用 PartitionAssignmentHandler
来避免重复。在 this 示例的帮助下,我了解它是如何工作的,但我不确定处理分区撤销事件的最干净方法是什么。如何在不等待批量大小完成的情况下通知我的提交者提交当前的一组偏移量?
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。