Spring Boot Kafka：ReplyingKafkaTemplate事务性配置问题求助

当前遇到两个核心问题：

• 启用read_committed隔离级别时，抛出KafkaReplyTimeoutException: Reply timed out
• 不启用read_committed时，抛出IllegalStateException: No transaction is in process

问题根源分析 #

1. read_committed下超时问题：生产者在@Transactional方法中调用sendAndReceive时，请求消息被纳入未提交的事务，使用read_committed隔离级别的消费者无法读取未提交消息，导致无法回复最终超时。
2. 无事务异常问题：消费者方法标注@Transactional但未绑定Kafka事务管理器，回复消息使用的事务型KafkaTemplate没有可用事务上下文，触发异常。

解决方案：完整配置与代码修改 #

1. 完善消费者容器工厂的事务配置 #

给ConcurrentKafkaListenerContainerFactory添加Kafka事务管理器，确保消费者的消息处理、回复发送、offset提交在同一事务中：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Event> kafkaListenerContainerFactory(
        KafkaTemplate<String, Event> kafkaTemplate, ProducerFactory<String, Event> producerFactory) {
    Map<String, Object> consumerFactoryConfigProps = new HashMap<>();
    consumerFactoryConfigProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
    consumerFactoryConfigProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    consumerFactoryConfigProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

    ConsumerFactory<String, Event> consumerFactory = new DefaultKafkaConsumerFactory<>(
            consumerFactoryConfigProps, new StringDeserializer(), new JsonDeserializer<>(Event.class));

    ConcurrentKafkaListenerContainerFactory<String, Event> kafkaListenerContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();
    kafkaListenerContainerFactory.setConsumerFactory(consumerFactory);
    kafkaListenerContainerFactory.setReplyTemplate(kafkaTemplate);
    // 绑定Kafka事务管理器，确保消费者事务与生产者事务联动
    kafkaListenerContainerFactory.setTransactionManager(new KafkaTransactionManager<>(producerFactory));
    // 设置事务超时时间，根据业务场景调整
    kafkaListenerContainerFactory.getContainerProperties().setTransactionTimeout(30_000);

    return kafkaListenerContainerFactory;
}

2. 调整生产者的请求发送逻辑 #

拆分请求发送与回复接收的事务边界，确保请求消息提交后再等待回复：

public void sendSaveFileEvent(SubmissionDto submissionDto) throws ExecutionException, InterruptedException, TimeoutException {
    SaveFileEvent outgoingSaveFileEvent = saveFileEventFactory
            .createEvent(submissionDto.getIncomingFileDto(), false, null);

    // 生成唯一关联ID，用于匹配回复消息
    String correlationId = UUID.randomUUID().toString();

    // 1. 在独立事务中发送请求消息（可同时添加其他事务性发送操作）
    replyingKafkaTemplate.executeInTransaction(template -> {
        ProducerRecord<String, Event> record = new ProducerRecord<>(topicName, outgoingSaveFileEvent.key(), outgoingSaveFileEvent);
        // 设置回复主题与关联ID
        record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, "reply_topic".getBytes()));
        record.headers().add(new RecordHeader(KafkaHeaders.CORRELATION_ID, correlationId.getBytes()));
        template.send(record);
        // 可添加其他事务性消息发送
        // template.send("other_business_topic", relatedEvent);
        return null;
    });

    // 2. 事务提交后，等待对应关联ID的回复
    RequestReplyFuture<String, Event, Event> replyFuture = replyingKafkaTemplate.receive(correlationId, 30_000);
    ConsumerRecord<String, Event> consumerRecord = replyFuture.get();

    log.debug("Received reply message: {}", consumerRecord.value());
}

3. 优化消费者的回复逻辑 #

确保回复消息正确关联请求的correlationId，保证生产者能匹配到对应回复：

@KafkaListener(id = "server", topics = "${application.kafka.saveFile.topicName}")
@SendTo
public Event handleSaveFileEvent(SaveFileEvent incomingSaveFileEvent, @Header(KafkaHeaders.CORRELATION_ID) byte[] correlationIdBytes) {
    try {
        log.debug("Processing request message: {}", incomingSaveFileEvent);

        SaveFileEvent outgoingSaveFileEvent = saveFileEventFactory
                .createEvent(incomingSaveFileEvent.getIncomingFileDto(), true, "Success");

        // 手动绑定关联ID到回复消息
        ProducerRecord<String, Event> replyRecord = new ProducerRecord<>("reply_topic", outgoingSaveFileEvent.key(), outgoingSaveFileEvent);
        replyRecord.headers().add(new RecordHeader(KafkaHeaders.CORRELATION_ID, correlationIdBytes));
        return replyRecord;
    } catch (Exception ex) {
        log.error("Failed to process request", ex);
        // 异常时返回带关联ID的错误回复
        ProducerRecord<String, Event> errorReply = new ProducerRecord<>("reply_topic", incomingSaveFileEvent.key(), incomingSaveFileEvent);
        errorReply.headers().add(new RecordHeader(KafkaHeaders.CORRELATION_ID, correlationIdBytes));
        return errorReply;
    }
}

4. 确保回复容器的配置一致性 #

回复容器需与业务消费者使用相同隔离级别，避免读取未提交消息：

@Bean
public ConcurrentMessageListenerContainer<String, Event> repliesContainer(
        ConcurrentKafkaListenerContainerFactory<String, Event> kafkaListenerContainerFactory) {
    Properties repliesContainerConfigProps = new Properties();
    repliesContainerConfigProps.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    // 显式设置隔离级别，与业务消费者保持一致
    repliesContainerConfigProps.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

    ConcurrentMessageListenerContainer<String, Event> repliesContainer =
            kafkaListenerContainerFactory.createContainer("reply_topic");
    repliesContainer.getContainerProperties().setGroupId("reply-group-" + UUID.randomUUID());
    repliesContainer.getContainerProperties().setKafkaConsumerProperties(repliesContainerConfigProps);

    return repliesContainer;
}

关键说明 #

• 事务边界拆分：生产者请求发送在独立事务中提交，确保read_committed消费者能立即读取；消费者的处理、回复、offset提交在同一事务中，保证原子性。
• 关联ID管理：手动生成并传递correlationId，避免ReplyingKafkaTemplate自动生成时的事务时序问题。
• 隔离级别一致性：所有消费者（业务处理、回复接收）统一使用read_committed，确保只读取已提交的事务消息。

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