java - "No transaction is in process"与 Spring Kafka 使用 ReplyingKafkaTemplate、@KafkaListener 和 @SendTo

Question

我想使用 Spring Kafka 的 ReplyingKafkaTemplate与 Kafka 事务一起。响应应用程序的代码如下所示:

@SpringBootApplication(scanBasePackages= {"com.example.myapp"})
public class ResponseApplication {
    private static final Logger LOGGER = LoggerFactory.getLogger(ResponseApplication.class);

    public static void main(String[] args) {
        SpringApplication.run(ResponseApplication.class, args);
    }

    @KafkaListener(id="${myapp.kafka.groupId}", topics="kRequests1")
    @SendTo(value= {"kReplies1"})
    @Transactional
    public String listen(String in) {
        return in;
    }

    @Bean
    public NewTopic kRequests1() {
        return new NewTopic("kRequests1", 2, (short) 1);
    }
}

配置类如下所示:

@Configuration
@EnableKafka
@EnableTransactionManagement
public class KafkaConfig {

    @Value( "${myapp.kafka.groupId}" )
    private String groupId;

    @Bean
    public KafkaTransactionManager<String, String> KafkaTransactionManager(
            ProducerFactory<String, String> producerFactory) {
        return new KafkaTransactionManager<String, String>(producerFactory);
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerConfigs());
        producerFactory.setTransactionIdPrefix("myapp");
        return producerFactory;
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("kafka.host", "localhost") + ":9092");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        return props;
    }

    @Bean
    public ContainerProperties containerProperties(KafkaTransactionManager<String, String> kafkaTransactionManager) {
        ContainerProperties containerProperties = new ContainerProperties("kRequests1");
        containerProperties.setTransactionManager(kafkaTransactionManager);
        containerProperties.setGroupId(groupId);
        return containerProperties;
    }

    @Bean("kafkaListenerContainerFactory")
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        KafkaTemplate<?, ?> replyTemplate = new KafkaTemplate<>(producerFactory());
        factory.setReplyTemplate(replyTemplate);
        return factory;
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getProperty("kafka.host", "localhost") + ":9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "iprs");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}

每次应用程序收到消息时，都会导致以下异常:

java.lang.IllegalStateException: No transaction is in process

由于该方法使用 @Transactional 注解、启用了事务管理且 ProducerFactory 有事务 ID 前缀，Spring 容器不应该启动事务吗？

Answer 1

我发现您正在使用回复模板；在任何情况下，该模板上的发送都会超出 @Transactional 的范围。

您不应在此用例中使用@Transactional；将事务管理器注入(inject)监听器容器工厂，以便容器可以在监听器退出时将偏移量发送到事务。