当禁用autocommit并接收确认头时，您只需要调用 acknowledge() 一旦你的信息被处理。这假设即使您在不同的线程中处理消息，也会保留对 Acknowledgment 示例，或作为原始的一部分 Message -或者，如果您正在进行转换，那么您正在复制头。但是这个电话需要你的密码。
第二，性能问题——它是由 KafkaNativeOffsetManager 实现使得对代理的调用变得阻塞，相对更昂贵（相对于简单地向压缩的主题发送消息，如 KafkaTopicOffsetManager 做。一般来说，在每条消息之后进行更新是非常昂贵的，在springxd中，我们通过使用https://github.com/spring-projects/spring-xd/blob/master/extensions/spring-xd-extension-kafka/src/main/java/org/springframework/integration/x/kafka/windowingoffsetmanager.java，从而减少有效写入的次数。我想我们可以为spring集成做一些类似的事情。
也就是说：相比之下，10万个更新在9.8秒内完成 KafkaNativeOffsetManager 在0.382秒内 KafkaTopicOffsetManager ，如所示https://github.com/mbogoevici/spring-integration-kafka/blob/perftest/src/test/java/org/springframework/integration/kafka/performance/offsetmanagerperformancetests.java （结果收集在我自己的机器上）。结果可能会有偏差，但仍然显示出很大的差异。在你的工具箱里追踪可以确认结果。

不知道这是怎么回事 KafkaNativeOffsetManager ，将是伟大的，如果你分享一些关于这个问题的调查，一些瓶颈的地方，在我们的代码在jira。
对于延迟补偿提交，我可以建议 autoCommitOffset = false 上 KafkaMessageDrivenChannelAdapter . 把它送到 channel 消息将被 KafkaHeaders.ACKNOWLEDGMENT 正面标题 DefaultAcknowledgment . 它确实满足了您的要求：

/**
 * Invoked when the message for which the acknowledgment has been created has been processed.
 * Calling this method implies that all the previous messages in the partition have been processed already.
 */
void acknowledge();