如何配置KafkaTemplate避免KafkaTransactionManager因TopicAuthorizationException进入错误状态

核心问题在于：默认情况下，Kafka生产者遇到TopicAuthorizationException这类异常时，会将事务上下文标记为失败，进而导致KafkaTransactionManager把全局事务置为错误状态，后续操作无法执行。要实现「忽略该异常、继续处理后续Item、保留数据库事务」的需求，需要从自定义生产者错误处理器入手，替代默认的异常处理逻辑。

步骤1：实现自定义ProducerErrorHandler #

创建一个自定义的错误处理器，针对TopicAuthorizationException做特殊处理——仅记录日志，不向上抛出异常，避免触发事务管理器的错误状态：

import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerErrorHandler;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class IgnoreAuthorizationErrorHandler implements ProducerErrorHandler {

    private static final Logger logger = LoggerFactory.getLogger(IgnoreAuthorizationErrorHandler.class);

    @Override
    public void handleError(ProducerRecord<?, ?> record, Exception exception, @Nullable ProducerFactory<?, ?> producerFactory) {
        if (exception instanceof TopicAuthorizationException) {
            logger.warn("发送消息到主题[{}]时遇到授权异常，忽略该异常继续处理", record.topic(), exception);
            // 不抛出异常，阻止事务上下文被标记为失败
        } else {
            // 其他异常按原有逻辑处理，比如抛出以触发事务回滚
            throw new RuntimeException("Kafka发送失败", exception);
        }
    }

    @Override
    public void handleError(ProducerRecord<?, ?> record, @Nullable SendResult<?, ?> result, Exception exception) {
        handleError(record, exception, null);
    }
}

步骤2：配置KafkaTemplate使用自定义错误处理器 #

在配置类中，给KafkaTemplate绑定这个自定义的错误处理器，替换默认的异常处理逻辑：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

@Configuration
public class KafkaConfig {

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
        KafkaTemplate<String, Object> template = new KafkaTemplate<>(producerFactory);
        // 设置自定义错误处理器
        template.setProducerErrorHandler(new IgnoreAuthorizationErrorHandler());
        // 无需依赖ProducerListener，错误处理器会直接拦截异常
        return template;
    }
}

关键说明 #

• 为什么不用ProducerListener？因为ProducerListener仅做异常监听，无法阻止异常传播到KafkaTransactionManager；而ProducerErrorHandler是在异常触达事务管理器前拦截处理，能直接控制是否触发事务失败。
• 保留数据库事务：由于拦截了TopicAuthorizationException不向上抛出，数据库事务不会因该异常回滚，后续的Item删除操作可正常执行。
• 异常区分处理：仅忽略授权类异常，其他Kafka异常（如网络故障）仍抛出以触发事务回滚，避免数据不一致。

额外注意点 #

如果需要记录发送失败的Item（用于后续重试或排查），可在handleError方法中将失败Item的信息存入日志或单独的重试记录表。

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