实现Kafka精确一次语义时触发非法状态转换异常

核心问题：事务嵌套冲突+生产者配置错误 #

你遇到的Invalid transition attempted from state COMMITTING_TRANSACTION to state ABORTING_TRANSACTION错误，根源是事务逻辑重复嵌套，同时生产者配置存在多处冲突：

1. 事务逻辑重复嵌套 #

你在使用kafkaTemplate.executeInTransaction()管理事务的同时，又在pushKafkaNotification方法里手动调用producer.beginTransaction()、commitTransaction()和abortTransaction()。这会导致同一个生产者实例被两个独立的事务逻辑控制：

• executeInTransaction已启动事务，生产者处于事务状态
• 手动调用事务方法会干扰状态，当executeInTransaction尝试提交时，代码可能因异常触发abortTransaction，直接导致状态转换冲突

2. 生产者配置的致命问题 #

• TransactionalId重复定义：同时在getProducerConfigMap里设置随机UUID的transactional.id，又给DefaultKafkaProducerFactory设置事务ID前缀。Spring Kafka会基于前缀自动生成事务ID（前缀+线程ID），手动指定随机值会彻底打乱事务管理逻辑。
• 重复创建生产者实例：同时定义了基于ProducerFactory的KafkaTemplate和独立的KafkaProducer Bean，两者配置冲突且事务管理完全隔离，进一步加剧状态混乱。

修复步骤 #

步骤1：修正生产者配置 #

删除手动指定的transactional.id，让Spring Kafka统一生成事务ID，同时移除独立的KafkaProducer Bean，统一使用KafkaTemplate：

@Configuration
@EnableKafka
public class KafkaConfig {

  @Value("${spring.kafka.bootstrap-servers}")
  private String bootstrapAddress;

  private ProducerFactory<String, String> getProducerFactory() {
    DefaultKafkaProducerFactory<String, String> factory =
        new DefaultKafkaProducerFactory<>(getProducerConfigMap());
    // 仅保留前缀，由Spring自动生成transactional.id
    factory.setTransactionIdPrefix(KafkaEventConstants.TRANSACTION_ID_PREFIX);
    return factory;
  }

  @Bean
  public KafkaTemplate<String, String> getKafkaTemplate() {
    return new KafkaTemplate<>(getProducerFactory());
  }

  private Map<String, Object> getProducerConfigMap() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    // 移除手动设置的transactional.id
    return config;
  }

  // 删除独立的KafkaProducer Bean，统一通过KafkaTemplate操作
}

步骤2：简化事务逻辑，移除嵌套 #

删除pushKafkaNotification里的手动事务代码，直接使用KafkaTemplate提供的事务上下文：

// 调用方代码保留，由executeInTransaction统一管理事务
kafkaTemplate.executeInTransaction(
    kafkaOperations -> {
      kafkaPublisher.pushKafkaNotification(
          kafkaOperations, topic, kafkaNotification.getUserId(), new JSONObject(kafkaNotification).toString());
      return true;
    });

// 修改推送方法，使用事务上下文内的KafkaOperations
public void pushKafkaNotification(
    KafkaOperations<String, String> kafkaOperations,
    String topic, 
    String partitionKey, 
    String serializedKafkaNotification) {
  try {
    ProducerRecord<String, String> producerRecord =
        new ProducerRecord<>(topic, partitionKey, serializedKafkaNotification);
    kafkaOperations.send(
        producerRecord,
        new Callback() {
          @Override
          public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null) {
              log.error(
                  "Callback : Failed to push event to kafka for partition key, notification {} {}",
                  partitionKey,
                  serializedKafkaNotification,
                  exception);
              // 抛出异常触发事务自动回滚
              throw new KafkaException("Failed to send message", exception);
            } else {
              log.info(
                  "Kafka Success Callback : Event pushed successfully to Kafka for partition key, notification {}, {}",
                  partitionKey,
                  serializedKafkaNotification);
            }
          }
        });
  } catch (Exception e) {
    metricLogger.errorMetricLogging(SERVICE_NAME, ErrorMetrics.DLQ_PUBLISH_ERROR.getCode());
    log.error("Exception while pushing notification to DLQ = {}", serializedKafkaNotification, e);
    // 抛出异常让事务回滚
    throw e;
  }
}

步骤3：关键注意事项 #

• 事务边界统一：所有消息发送必须在executeInTransaction的回调内完成，由Spring Kafka统一处理提交和回滚。
• 禁止手动操作生产者：不要直接创建KafkaProducer实例，全部通过KafkaTemplate或ProducerFactory获取事务上下文内的生产者。
• 异常触发回滚：在发送逻辑或回调中抛出异常时，executeInTransaction会自动触发事务回滚，无需手动调用abortTransaction。

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