How to send Kafka offsets in transaction, which is created by KafkaTemplate?

Question

I want to implement read-process-write pattern - https://www.confluent.io/blog/transactions-apache-kafka/. So, I need to consume records, process them and then to commit consumed offsets.

I use org.apache.kafka.clients.consumer.KafkaConsumer for consuming messages. I mean, it is not spring related consumer.

I use org.springframework.kafka.core.KafkaTemplate for producing messages. I create its bean like this:

@Bean
public Map<String, Object> producerConfigs() {
    final Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "bootstrapServers");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    return props;
}

@Bean
public DefaultKafkaProducerFactory<String, String> defaultKafkaProducerFactory() {
    DefaultKafkaProducerFactory<String, String> kafkaProducerFactory = new DefaultKafkaProducerFactory<>(producerConfigs());
    kafkaProducerFactory.setTransactionIdPrefix("transaction-id-prefix");
    return kafkaProducerFactory;
}

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

I produce result messages like this:

ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(POLL_INTERVAL_IN_MS));

List<List<String>> outputMessages = produceOutput(consumerRecords);

kafkaTemplate.executeInTransaction(kafkaProducer -> {
    for (List<String> resultTasks : outputMessages) {
        for (String resultTask : resultTasks) {
            kafkaProducer.send("topic", "key", resultTask);
        }
    }

    kafkaProducer.sendOffsetsToTransaction(getOffsetsForCommit(consumerRecords), "consumerGroupId");
    return true;
});

Finally, I have this error:

java.lang.IllegalArgumentException: No transaction in process
    at org.springframework.util.Assert.isTrue(Assert.java:118)
    at org.springframework.kafka.core.KafkaTemplate.sendOffsetsToTransaction(KafkaTemplate.java:345)

Exception throws in this method:

@Override
public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) {
    @SuppressWarnings("unchecked")
    KafkaResourceHolder<K, V> resourceHolder = (KafkaResourceHolder<K, V>) TransactionSynchronizationManager
            .getResource(this.producerFactory);
    Assert.isTrue(resourceHolder != null, "No transaction in process"); // here
    if (resourceHolder.getProducer() != null) {
        resourceHolder.getProducer().sendOffsetsToTransaction(offsets, consumerGroupId);
    }
}

So, how to properly commit these offsets?

Answer 1

It's a bug; sendOffsetsToTransaction() doesn't work in executeInTransaction - it assumes a Spring transaction is bound to the thread.

As a work-around, you can either use @Transactional on the method or use a transaction template with a KafkaTransactionManager to start the Spring transaction instead of using executeInTransaction().

TransactionTemplate tt = new TransactionTemplate(tm);

...

        this.tt.execute(s -> {
                    template.send(...);
                    template.sendOffsetsToTransaction(...);
                    return null;
                });

Please open a GitHub Issue and we'll fix this.