Spring @Retryable未按预期生效：Kafka监听事件处理异常排查

我正在开发一个Spring Boot应用，启动类配置如下：

@SpringBootApplication
@EnableRetry
public class SpringBootApp {
  public static void main(String[] args) {
    SpringApplication.run(SpringBootApp.class, args);
  }
}

配置了对接包含5种不同Schema的Kafka主题，事件拆分逻辑：

private void fillEvents(ConsumerRecords<Key, SpecificRecord> events) {
    events.forEach(event -> {
        SpecificRecord value = event.value();
        if (value instanceof A a) {
            aEvents.add(a);
        } else if (value instanceof B b){
            bEvents.add(b);
        }
        // 其他类型判断
   });
}

Kafka主监听器逻辑：

@KafkaListener(topics = "topicName", groupId = "myApp", containerFactory = "listenerFactory")
public void receive(ConsumerRecords<Key, SpecificRecord> events) {
    Splitter splitter = new Splitter(events); // 执行上述fillEvents逻辑
    aService.handleEvents(splitter.getAEvents());
    bService.handleEvents(splitter.getBEvents());
    // 其他类型事件处理
}

使用MongoDB做持久化，为避免并发访问失败，服务层处理如下：

public void handleEvents(List<A> events) {
    events.forEach(event -> processEvent(event));
}

@Retryable(value = {OptimisticLockingFailureException.class, DuplicateKeyException.class, MongoCommandException.class},
        maxAttempts = 100,
        backoff = @Backoff(random = true, delay = 200, maxDelay = 5000, multiplier = 2))
public void processEvent(A event) {
  refresh(); // 重试失败时刷新依赖
  processBusinessRules(event); // 执行业务规则处理
  aRepository.save(event);
}

当前遇到的问题：Kafka监听器拉取约30条包含A、B类型的消息时，处理A事件触发OptimisticLockingFailureException，但B事件未被处理，线程在首次失败后停止，未触发processEvent方法的重试，仅靠Kafka监听器重新拉取，不符合预期。希望触发processEvent的重试，且不丢弃后续事件。

1. 内部方法调用导致@Retryable未生效 #

Spring的@Retryable基于AOP代理实现，同一个Bean内的方法互相调用不会触发代理逻辑。你的handleEvents直接调用同Bean的processEvent，导致重试注解完全失效，异常直接向上抛出，最终让整个Kafka监听器方法失败，批次被回滚，B事件也无法执行。

解决方法：

• 将processEvent拆分到单独的Bean（如AEventProcessor），通过依赖注入调用该Bean的方法，触发代理。
• 或在当前Bean中用AopContext获取代理对象调用方法：

public void handleEvents(List<A> events) {
    events.forEach(event -> ((YourServiceClass) AopContext.currentProxy()).processEvent(event));
}

注意：使用AopContext需要在启动类添加@EnableAspectJAutoProxy(exposeProxy = true)。

2. 批量监听器的全局异常回滚问题 #

当前使用批量监听器接收ConsumerRecords，一旦方法内抛出未捕获异常，整个批次消息都会被标记为处理失败，触发Kafka重新拉取，导致未出错的B事件也被重复处理，后续逻辑直接中断。

优化方案：

• 捕获单个事件异常：在handleEvents中对每个processEvent调用加try-catch，确保单个事件失败不影响其他事件：

public void handleEvents(List<A> events) {
    events.forEach(event -> {
        try {
            // 调用代理后的processEvent方法
            processEvent(event);
        } catch (Exception e) {
            log.error("处理A事件失败: {}", event.getId(), e);
        }
    });
}

• 改用单条消息监听器：将监听器改为接收单条ConsumerRecord，单条失败不影响其他消息，配合重试逻辑更灵活：

@KafkaListener(topics = "topicName", groupId = "myApp", containerFactory = "listenerFactory")
public void receive(ConsumerRecord<Key, SpecificRecord> event) {
    SpecificRecord value = event.value();
    if (value instanceof A a) {
        aService.processEvent(a);
    } else if (value instanceof B b) {
        bService.processEvent(b);
    }
    // 其他类型处理
}

3. 补充重试耗尽后的降级逻辑 #

建议添加@Recover方法，处理重试100次后仍失败的情况，避免异常无限传播：

@Recover
public void recoverProcessEvent(Exception e, A event) {
    log.error("A事件重试耗尽仍失败: {}", event.getId(), e);
    // 自定义降级逻辑：如写入死信队列、触发告警等
}

4. 批量监听器的自定义错误处理（可选） #

如果坚持用批量监听器，可配置BatchErrorHandler实现部分失败不影响整个批次：

@Bean
public BatchErrorHandler batchErrorHandler() {
    return new BatchLoggingErrorHandler() {
        @Override
        public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
            data.forEach(record -> {
                try {
                    // 单独处理每条失败记录
                    receiveSingle(record);
                } catch (Exception e) {
                    log.error("单条记录处理失败", e);
                }
            });
        }
    };
}

然后在Kafka监听器工厂中配置该错误处理器。

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