Kafka Consumer Invalid Payload ErrorHandler如何实现发邮件与错误存库

实现逻辑说明 #

你需要首先区分两类异常场景：

• 无效消息：对应@Valid参数校验失败抛出的MethodArgumentNotValidException，触发邮件告警
• 消费逻辑错误：除校验异常外的其他业务/系统异常，触发错误信息入库
  注意你配置的SeekToCurrentErrorHandler会先执行3次重试（每次间隔5秒），重试全部失败后才会进入自定义恢复回调执行逻辑。

代码改造步骤 #

1. 注入依赖服务 #

在配置类中注入邮件发送服务和数据库操作服务，确保两个服务已经提前实现并注入Spring容器。

2. 改造错误处理器逻辑 #

需要先提取Spring Kafka包装后的根异常，再根据异常类型执行对应逻辑，同时可配置校验异常不重试减少无效开销。

完整改造后代码 #

@Configuration
@EnableKafka
public class KafkaConsumerConfig implements KafkaListenerConfigurer{

    // 注入数据库操作服务
    @Resource
    private DbService dbService;
    // 注入自定义邮件发送服务
    @Resource
    private MailService mailService;

    @Bean
    ErrorHandler errorHandler() {
        SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler((rec, ex) -> {
            // 提取根异常
            Throwable rootEx = getRootCause(ex);
            // 判断是否为参数校验异常（无效消息）
            if (rootEx instanceof MethodArgumentNotValidException) {
                // 组装告警邮件内容，可自定义字段：topic、offset、消息内容、校验错误信息等
                String alertContent = String.format("收到无效Kafka消息，topic:%s，offset:%d，校验错误：%s",
                        rec.topic(), rec.offset(), ((MethodArgumentNotValidException) rootEx).getBindingResult().getAllErrors());
                // 发送邮件告警，建议邮件服务内部实现异步发送避免阻塞消费者线程
                mailService.sendAlertMail(alertContent);
            } else {
                // 其他消费错误，保存到数据库
                dbService.saveErrorMsg(rec, rootEx.getMessage());
            }
        }, new FixedBackOff(5000, 3));

        // 配置参数校验异常不需要重试，直接进入回调
        errorHandler.addNotRetryableExceptions(MethodArgumentNotValidException.class);
        return errorHandler;
    }

    // 工具方法：提取异常根因
    private Throwable getRootCause(Throwable ex) {
        Throwable cause = ex;
        while (cause.getCause() != null && cause != cause.getCause()) {
            cause = cause.getCause();
        }
        return cause;
    }

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        registrar.setValidator(this.validator);
    }

    @KafkaListener(topics = "mytopic", concurrency = "3", groupId = "mytopic-1-groupid")
    public void consumeFromTopic1(@Payload @Valid ValidatedClass val, ConsumerRecordMetadata meta) throws Exception {
        dbservice.callDB(val,"t");
    }
}

注意事项 #

• 邮件发送逻辑建议实现异步处理，避免阻塞消费者线程影响正常消息消费
• 错误信息入库操作建议单独加事务，避免和消费逻辑的事务互相影响
• 可根据业务需要调整是否在校验异常时也同步保存错误信息到数据库

内容的提问来源于stack exchange，提问作者Pale Blue Dot