Spring Kafka消费者事务超时：为何消费时触发事务？

我们的事务型Producer运行正常，未出现任何问题，但Consumer端日志中出现了事务启动及超时异常，想咨询：为何在消费消息的过程中会启动事务，进而引发该异常？

消费端异常日志 #

2022-12-28 18:02:05.986 DEBUG  [qa] 85474 --- [tainer#0-51-C-1] o.s.k.t.KafkaTransactionManager          :  Creating new transaction with name [null]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
2022-12-28 18:02:10.437 DEBUG  [qa] 85474 --- [tainer#0-51-C-1] o.s.k.t.KafkaTransactionManager          :  Created Kafka transaction on producer [CloseSafeProducer [delegate=brave.kafka.clients.TracingProducer@43e18a26]]
2022-12-28 18:02:10.438 DEBUG  [qa] 85474 --- [tainer#0-51-C-1] abc.abc.kafka                 :  NORMAL uid[n/a] cid[uPRwhWZikfcYpGfIqh7TxXFnm6VmZWkf] m[[S] Consuming message] data[record = ConsumerRecord(topic =XXXXXXX)))]
2022-12-28 18:02:10.438 DEBUG  [qa] 85474 --- [tainer#0-51-C-1] abc.abc.kafka                 :  NORMAL uid[n/a] cid[uPRwhWZikfcYpGfIqh7TxXFnm6VmZWkf] m[Processed message]

2022-12-28 18:03:10.439  INFO  [qa] 85474 --- [tainer#0-51-C-1] abc.abc.common.Generic                   :  NORMAL uid[n/a] cid[uPRwhWZikfcYpGfIqh7TxXFnm6VmZWkf] m[Throwable t] data[exception = [Ljava.lang.StackTraceElement;@6d314e1d] ex[Timeout expired after 60000 milliseconds while awaiting AddOffsetsToTxn] sts[org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000 milliseconds while awaiting AddOffsetsToTxn
]
2022-12-28 18:03:10.439 DEBUG  [qa] 85474 --- [tainer#0-51-C-1] abc.abc.kafka                 :  NORMAL uid[n/a] cid[uPRwhWZikfcYpGfIqh7TxXFnm6VmZWkf] m[[E] Consuming message]
2022-12-28 18:03:10.439 DEBUG  [qa] 85474 --- [tainer#0-51-C-1] o.s.k.t.KafkaTransactionManager          :  Initiating transaction commit
2022-12-28 18:04:10.444 ERROR  [qa] 85474 --- [tainer#0-51-C-1] o.s.k.core.DefaultKafkaProducerFactory   : org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000 milliseconds while awaiting EndTxn(true)
 commitTransaction failed: CloseSafeProducer [delegate=brave.kafka.clients.TracingProducer@43e18a26]
2022-12-28 18:04:10.444 DEBUG  [qa] 85474 --- [tainer#0-51-C-1] o.s.k.t.KafkaTransactionManager          : org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000 milliseconds while awaiting EndTxn(true)
 Initiating transaction rollback after commit exception
2022-12-28 18:04:10.445  WARN  [qa] 85474 --- [tainer#0-51-C-1] o.s.k.core.DefaultKafkaProducerFactory   :  Error during some operation; producer removed from cache: CloseSafeProducer [delegate=brave.kafka.clients.TracingProducer@43e18a26]
2022-12-28 18:04:12.436 ERROR  [qa] 85474 --- [tainer#0-51-C-1] o.s.k.l.KafkaMessageListenerContainer    : org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000 milliseconds while awaiting EndTxn(true)
 Transaction rolled back

相关配置 #

spring:
  data:
    mongodb:
      uri: indb-prop
      auto-index-creation: false
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      retries: 2
      client-id: ${spring.application.name}-${info.cluster}-${IP_ADDRESS}PP
      transaction-id-prefix: tx-${spring.kafka.producer.client-id}-
      properties:
        enable.idempotence: true
        spring.json.add.type.headers: false
      bootstrap-servers:  ${kafka_bootstrap_servers_2}
#    listener:
#      missing-topics-fatal: true
#      type: batch
#      concurrency: 15
#      ack-mode: manual_immediate
#      poll-timeout: 1s
    consumer:
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      group-id: ${spring.application.name}-${info.cluster}-111111111112
      client-id: ${spring.application.name}-${info.cluster}-${IP_ADDRESS}
#      client-id: ${spring.kafka.consumer.group-id}-${IP_ADDRESS}
      auto-offset-reset: earliest
      enable-auto-commit: false
      isolation-level: read_committed
#      max-poll-records: 3
      fetch-max-wait: 5s
      properties:
        max.poll.interval.ms: 20000000
        spring.json.trusted.packages: '*'
        spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.value.default.type: net.abc.abc.EventData
      bootstrap-servers: ${kafka_bootstrap_servers}

Consumer代码 #

@KafkaListener(
        topics = "SOME_TOPIC",
        autoStartup = "true")
// @Transactional(transactionManager = "mongoTransactionManager", propagation = Propagation.REQUIRED)
@Override
public void onMessage(ConsumerRecord<String, EventData> data) {

    if (data == null || data.value() == null) {
        return;
    }

    logger.debug(m -> m.event(KAFKA)
            .msg("[S] Consuming message")
            .with("record", data));

    try {

        logger.debug(m -> m.event(KAFKA)
                .msg("Processed message"));
    }
    finally {
        logger.debug(m -> m.event(KAFKA)
                .msg("[E] Consuming message"));

    }
}

版本信息 #

Spring Boot：2.6.6

消费过程启动事务的原因 #

1. 自动创建的Kafka事务管理器：Producer配置了transaction-id-prefix，会触发Spring自动创建KafkaTransactionManager实例。当容器中存在该管理器，且Consumer配置了enable-auto-commit: false时，Spring Kafka会默认用它管理消费偏移量的提交——将偏移量提交纳入Kafka事务，确保原子性。
2. 潜在的全局事务配置：即使代码中@Transactional被注释，若项目启用了@EnableTransactionManagement且未指定默认事务管理器，Spring会自动选用容器中存在的KafkaTransactionManager，从而触发消费时的事务。

超时异常的原因 #

日志中的AddOffsetsToTxn和EndTxn(true)超时，说明将消费偏移量加入事务、提交事务的过程中，客户端与Kafka集群的通信超出了默认60秒的超时时间。可能诱因包括：

• Kafka集群负载过高，处理事务请求缓慢；
• 客户端与集群间网络延迟大；
• 默认事务超时参数（transaction.timeout.ms=60000）过短，无法匹配集群响应速度。

解决方案 #

1. 关闭消费端事务（若不需要）：如果消费逻辑不需要偏移量提交与业务操作绑定成原子事务，直接移除Producer配置中的transaction-id-prefix，Spring不会创建KafkaTransactionManager，消费过程就不会启动事务。
2. 显式指定事务管理器：若需要使用Mongo事务，开启@Transactional并指定transactionManager="mongoTransactionManager"，同时配置默认事务管理器为Mongo的，避免Spring默认选用KafkaTransactionManager。
3. 调整超时参数：增大Producer的transaction.timeout.ms（比如设为120000），给集群足够时间处理事务请求；同时检查Consumer的网络超时参数，确保通信超时时间合理。
4. 排查集群状态：检查Kafka集群运行状态，确认Broker正常、分区副本同步，排除集群故障或网络瓶颈导致的响应缓慢。

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