Kafka生产者推送同Key事件至分区顺序错乱如何解决

根因说明 #

你遇到的同Key消息消费乱序问题和linger.ms配置无直接关联。Kafka本身承诺同分区内消息按生产写入顺序存储，出现UPDATE早于INSERT被消费的核心原因是生产者重试机制导致的消息重排：
当第一条INSERT消息因网络抖动、Broker Leader切换等临时故障发送失败触发重试时，后发送的UPDATE消息已经成功写入目标分区，就会出现分区偏移量顺序和生产发送顺序不一致的情况，和你贴出的日志表现完全吻合。
linger.ms仅用于控制生产者批量发送的等待时长，只能降低短时间连续发消息触发乱序的概率，无法从根因上解决重试导致的顺序错乱，属于治标不治本的方案。

推荐修复方案（按优先级排序） #

• 方案1：开启幂等生产者（最优雅，无业务侵入，优先选）
  该方案从Kafka协议层保证同分区消息顺序，不需要修改业务逻辑，仅需调整3项生产者配置即可：
  1. 强制设置acks=all（即-1），这是幂等机制生效的必要前置条件，替换你当前配置中可选传入的acks逻辑
  2. 设置enable.idempotence=true，开启生产者幂等能力（Kafka 3.0+版本该参数默认值为true，低版本需手动配置）
  3. 设置max.in.flight.requests.per.connection=5：Kafka 1.0+版本配合幂等机制，该值设为5时既可以保证消息顺序，又不会损失吞吐量；如果你的Kafka版本低于1.0，需要将该值设为1才能保序，会小幅降低生产吞吐量。
     修改后的生产者配置代码示例：

  @Bean
  public Map<String, Object> producerProperties() {
    Map<String, Object> map = new HashMap<>();
  
    map.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    // 强制设置acks为all，满足幂等生效条件
    map.put(ProducerConfig.ACKS_CONFIG, "all");
    // 开启幂等
    map.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    // 配置同连接最大未完成请求数，1.0+版本设为5兼顾顺序和吞吐
    map.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
  
    map.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    map.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
  
    return Collections.unmodifiableMap(map);
  }
  

  实现原理：幂等生产者启动时会从Broker获取唯一的PID（生产者实例标识），给发往同分区的每条消息附加单调递增的序列号，Broker端会对消息序列号做校验，遇到乱序到达的重试消息会暂存等待前置消息到达后再写入，从底层保证同分区消息的写入顺序和生产发送顺序一致，同时还能避免重试导致的消息重复问题。
• 方案2：消费端业务兜底（可选，配合幂等机制做双层防护）
  分布式场景下没有100%的绝对可靠，可在消费端增加轻量兜底逻辑，避免极端异常下的业务错误：
  • 给业务实体附加版本号/更新时间戳，消费UPDATE消息时如果发现当前已存储的实体版本高于消息携带的版本，直接丢弃过期消息
  • 如果消费UPDATE消息时发现对应实体不存在，不要立刻无限重试，可将消息投递到延迟队列等待INSERT消息处理完成后再消费，或按业务规则做兜底插入处理

不推荐使用linger.ms解决问题的原因 #

调整linger.ms仅能让生产者攒够批量或等待固定时长后再统一发送消息，只能降低短时间连续发送多条消息时触发乱序的概率，只要存在单条消息发送失败重试的场景，依然会出现消息重排，无法彻底解决问题。

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