Spring-kafka使用RecordFilterStrategy丢弃消息提交offset触发重平衡报错

问题核心解决思路 #

• 第一优先修改不合理的pollTimeout配置
  你代码中设置的factory.getContainerProperties().setPollTimeout(9999999)相当于单次poll调用最长等待近3小时，这是触发重平衡的最核心原因：pollTimeout是Kafka消费者单次拉取消息的最大等待时间，设置过大时消费者会长时间阻塞在拉取消息的逻辑里，直接导致两次poll调用的间隔超过max.poll.interval.ms阈值，触发消费组重平衡，提交offset时自然就会报错。
  建议调整为合理值，比如默认的3000毫秒即可：

factory.getContainerProperties().setPollTimeout(3000);

• 调整offset提交方式降低阻塞
  你当前使用的AckMode.MANUAL_IMMEDIATE是同步提交每条消息的offset，过滤的消息较多时频繁的同步提交会阻塞消费线程，拉长两次poll的间隔。可以修改为异步提交：

// 开启异步提交，避免提交offset阻塞消费线程
factory.getContainerProperties().setSyncCommits(false);
// 可选：配置异步提交的回调，方便排查提交失败的问题
factory.getContainerProperties().setCommitCallback((offsets, exception) -> {
    if (exception != null) {
        // 自定义日志记录提交异常
        log.error("offset提交失败:{}", offsets, exception);
    }
});

如果不需要每条消息都实时提交，也可以把AckMode调整为MANUAL，等整个批次的消息处理/过滤完成后统一提交，大幅减少提交次数。

• 优化过滤逻辑避免阻塞
  检查RecordFilterStrategy的实现逻辑：你当前直接强转consumerRecord.value()为Shipment类型，如果消息格式不匹配会抛出类型转换异常，异常未捕获的情况下会触发消费重试，导致消费线程卡住。建议补充异常捕获逻辑：

factory.setRecordFilterStrategy(new RecordFilterStrategy<Object, Object>() {
    @Override
    public boolean filter(ConsumerRecord<Object, Object> consumerRecord) {
        try {
            Shipment shipment = (Shipment) consumerRecord.value();
            return shipment.getType().contains("YAW");
        } catch (Exception e) {
            log.error("消息过滤失败，消息key:{}", consumerRecord.key(), e);
            // 异常消息直接丢弃，或者根据业务需求返回false进入正常消费逻辑处理
            return true;
        }
    }
});

• 确认参数生效及消费链路无阻塞

1. 启动时打印消费者配置，确认你修改的max.poll.interval.ms、max.poll.records参数确实被加载，没有被其他配置覆盖
2. 排查正常业务消息的消费逻辑是否存在慢调用，比如慢SQL、第三方接口超时等，这些都会拉长两次poll的间隔，触发重平衡

• 极端场景优化
  如果需要过滤的消息占比很高，可以优先在生产端或者通过Kafka Streams做前置过滤，把符合要求的消息发送到独立的业务Topic，消费者不需要处理丢弃的消息，从根源上避免这个问题。

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