spring-cloud-stream-kafka在应用程序启动后仅使用最新消息

Question

在我们的项目中，我们使用spring-cloud-stream-binder-kafka 1.1.2版与kafka集成。最近我们遇到了这样一种情况：我们的某项服务在启动后会消耗某个主题的旧消息（已经消耗）。该主题有2个分区和2个消费者组成的消费者群体。我们不确定偏移量是否适用于动物园管理员。在启动过程中，每条消息都会引发错误消息。

[-kafka-listener-2] ERROR o.s.k.listener.LoggingErrorHandler.handle - Error while processing: ConsumerRecord(topic = statemachine_deal_notification, partition = 1, offset = 926, key = null, value = [B@6fab0a32) 

为确保不再发生，我们一直只想阅读主题中的最新消息。我发现设置resetOffsets到true和startOffset到latest将做这项工作。但是这些属性不会影响消费者。后来发现这个功能已经被丢弃了。

有没有其他方法可以确保特定群组中的消费者只消费最新消息？

Answer 1

这是一个bug，随后在这里修复：https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/commit/7355ada4613ad50fe95430f1859d4ea65f004be1。您可以试用SNAPSHOT（1.2.2.BUILD-SNAPSHOT或1.3.0.BUILD-SNAPSHOT）版本来验证此修复。