咨询：基于Kafka实现10分钟内的跨应用Avro消息去重方案

这个问题很典型！你提到的Exactly Once语义确实主要解决生产者重试时的幂等性问题，但对于这种跨分钟、甚至跨应用实例的重复消息场景，我们可以通过基于消息唯一标识的幂等处理结合Kafka的特性来解决，下面是几个可行的方案：

Exactly Once的生产者幂等依赖生产者ID+序列号，只能覆盖单生产者短时间内的重试重复。而你需要的是跨实例、长窗口（10分钟）的去重，核心是要给每条消息分配一个全局唯一且稳定的业务标识（比如事件UUID、订单ID+操作类型），以此作为去重的判断依据。

方案一：生产者端前置去重（优先推荐） #

在消息发送到Kafka之前就做去重校验，避免无效消息进入Topic：

• 本地缓存方案：用Guava Cache或Caffeine这类本地缓存，设置10分钟的过期时间。发送消息前先检查缓存中是否存在该消息的唯一标识，不存在则发送并写入缓存。缺点是多实例部署时，本地缓存不共享，可能会有漏判。
• 分布式缓存方案：用Redis的SETNX命令（或带过期时间的SET命令），把消息唯一标识作为key，设置10分钟过期。只有当SETNX返回成功时，才发送消息到Kafka。这种方案适合多实例部署的场景，能保证全局去重。

方案二：Kafka Streams端状态存储去重（适合流式链路） #

既然你已经在使用Kafka Streams，可以直接利用它的状态存储实现流处理中的去重：

1. 调整消息Key：把消息的业务唯一标识设置为Kafka消息的key（如果原来的key不是的话，通过KStream.map()转换）。
2. 使用带过期的状态存储：
   • 创建一个KeyValueStore类型的状态存储，配置10分钟的过期时间（通过Materialized.as(...).withRetention(Duration.ofMinutes(10))）。
   • 用KStream.transform()或KStream.filter()处理每条消息：检查状态存储中是否存在当前key，不存在则保留消息并写入状态存储，存在则过滤掉重复消息。
3. 注意分区一致性：因为你的Topic有40个分区，要确保相同业务标识的消息落到同一个分区（可以用自定义分区器，根据业务ID哈希到固定分区），否则不同分区的状态存储无法共享去重信息。

方案三：事务+幂等性增强（局限性较大） #

如果你的重复消息是因为生产者重启或跨实例的重复触发，可以尝试结合事务和自定义标识，但实现复杂度较高，不推荐作为首选：

• 开启生产者幂等性（enable.idempotence=true）和事务（设置全局唯一的transactional.id，比如包含业务标识前缀），但事务的幂等性依然依赖生产者ID，要覆盖跨实例的重复，需要额外把业务唯一标识纳入事务的判断逻辑，实现起来比较繁琐。

• 唯一标识的稳定性：必须选择不会变化的字段作为唯一标识，比如全局UUID、业务主键+操作类型，不能用时间戳这类可变字段。
• 过期时间匹配：缓存或状态存储的过期时间要严格匹配你的去重窗口（10分钟），避免占用过多存储资源。
• 分区策略适配：如果用Kafka Streams去重，一定要保证相同业务ID的消息落到同一个分区，否则去重逻辑会失效。

内容的提问来源于stack exchange，提问作者T1234