卡夫卡春中的延迟ACK

Question

我使用Spring和Spring作为批处理服务，从Kafka收集数据，直到满足特定条件，然后转储数据。

当数据离开我的服务时，我想确认提交，但它可能会在内存中停留5-10分钟。

考虑到Spring中的确认实现保留了原始记录，在我转储数据之前保留它们似乎是不合理的，考虑到这会对内存利用率造成什么影响。

仅考虑到分区/偏移量信息，还有其他方法可以确认/提交Spring的偏移量吗？

Answer 1

您可以将AckMode.TIME或AckMode.COUNT与非常大的ackTime或ackCount一起使用，这样容器就不会执行此操作。

然后，将Consumer<?, ?>传递到侦听器方法并执行偏移量提交。

但是，请注意，使用者并不是线程安全的，因此必须对侦听器线程执行提交。

此外，请记住，记录不是单独的，只是偏移量。你不能破坏“无序”。

此外，您可能需要将max.poll.interval.ms增加到其缺省值(5分钟)以外，以避免分区的重新平衡。