Search code examples
javaspringapache-kafkaspring-transactionsspring-kafka

"No transaction is in process" with Spring Kafka using ReplyingKafkaTemplate, @KafkaListener and @SendTo


I want to use Spring Kafka's ReplyingKafkaTemplate together with Kafka transactions. The code of the response application looks like this:

@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);
    }
}

The configuration class looks like this:

@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);
    }
}

Everytime the application receives a message, it results in the following exception:

java.lang.IllegalStateException: No transaction is in process

Shouldn't the Spring container start a transaction as the method is annotated with @Transactional, transaction management is enabled and ProducerFactory has a transaction ID prefix?


Solution

  • I see you are using a reply template; the send on that template would, in any case, be outside the scope of the @Transactional.

    You should not use @Transactional for this use case; inject the transaction manager into the listener container factory so the container can send the offset to the transaction when the listener exits.