如何使用 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 (将#修改为@）