KafkaTransactionManager does not rollback produced record when logs say transaction was rolled back

[ostecke]

Seen between spring-kafka versions 2.8.11 and 2.9.6, did not try previous versions

Describe the bug

I'm trying yet failing to create the following situation: a Spring Kafka consumer reading from topic1, producing to topic2, saving data in a SQL database, all in a transaction, where everything is rolled back if an exception occurs after producing and/or saving.

There is a lot of information on docs.spring.io, StackOverflow & this repository, yet I'm not able to make this happen. Main driver for this bugreport is the code example in https://docs.spring.io/spring-kafka/docs/2.9.5/reference/html/#ex-jdbc-sync.

Current behaviour

Using the consumer based transactional code example from the above mentioned URL, appended with an exception (see line 57 of the code mentioned in section 'To Reproduce') I notice the following behaviour:

Using the KafkaListener to start a transaction with the KafkaTransactionManager, when a (runtime) exception is thrown after publishing a record to topic2, the kafka record is still committed. The KafkaTransactionManager and the Producer both log that the transaction will be rolled back, yet the record can be seen committed on topic2 after the application is stopped.

The JDBC transaction is successfully rolled back and cannot be found in the database.

Small snippet of the logs, please reproduce locally for a full stacktrace:

2023-03-14 20:49:37.924 TRACE 11124 --- [   group1-0-C-1] o.s.t.i.TransactionInterceptor           : Completing transaction for [org.example.Application$Listener.listen1] after exception: java.lang.RuntimeException: rollback?
2023-03-14 20:49:37.927 DEBUG 11124 --- [   group1-0-C-1] o.s.t.support.TransactionTemplate        : Initiating transaction rollback on application exception

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void org.example.Application$Listener.listen1(java.lang.String) throws java.lang.Exception' threw exception; nested exception is java.lang.RuntimeException: rollback?; nested exception is java.lang.RuntimeException: rollback?
---snip---

2023-03-14 20:49:37.927 DEBUG 11124 --- [   group1-0-C-1] o.s.k.t.KafkaTransactionManager          : Initiating transaction rollback
2023-03-14 20:49:37.927  INFO 11124 --- [   group1-0-C-1] o.a.k.clients.producer.KafkaProducer     : [Producer clientId=producer-tx-group1.topic1.0, transactionalId=tx-group1.topic1.0] Aborting incomplete transaction
2023-03-14 20:49:37.933 ERROR 11124 --- [   group1-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Transaction rolled back

Expected behavior

I expect both the Kafka & JDBC transactions to be rolled back, the database being empty and topic2 being empty.

To Reproduce

Checkout repository: https://github.com/ostecke/kafka-transactions
Run docker-compose up -d in the root directory to start the necessary background services
Run the application
See the following URL to find unexpected messages on topic2: http://localhost:9090/ui/clusters/local/all-topics/topic2/messages

[garyrussell]

I don't know what you have running on 9090; there is no web code in the sample; perhaps it is using a consumer that doesn't have read_comitted set?

I don't see any messages emitted by your listen2 listener; so it looks like it's working as expected to me.

[ostecke]

When running the docker-compose it will start KafkaUI accessible on port 9090. In KafkaUI I'm able to read the messages on topic2 that I believe should have been rolled back.

[garyrussell]

Oh; you are using the kafka ui - got it.

Kafka logs are immutable - the records are still written to the log; a marker is placed afterwards to indicate whether the tx committed or rolled back.

So that tool is reading the uncommitted records.

[garyrussell]

That's what the isolation.level: read_committed is for; and that is why listen2 doesn't get the records.

[garyrussell]

Notice the offsets:

In general they increment by 2 each time (with the commit/rollback frame accounting for the "missing" offset).

In some cases, it jumps by 3, presumably because, sometimes, the commit rolls back before the record was actually written to the log.

In this case, there are only 8 uncommitted records (and 2 cases where the offset jumps by 3). The default error handler tries 10 times.

[ostecke]

Arghhh... Unbelievable that I never thought of looking at uncommitted records through KafkaUI. Also a bit of a bummer that it's not indicated anywhere.

I'm very happy with your great help in this matter, thank you very much

[garyrussell]

No problem; I find it a bit strange that read_uncommitted is the default behavior; perhaps it has performance implications for consuming non-transactional records; but that would be a question for the Apache Kafka folks.

[garyrussell]

Thanks for providing the repro; it makes supporting the project much easier. I wish more people would take the time.

[dineshp86]

@garyrussell as per your comments Kafka logs are immutable - the records are still written to the log, marker is placed afterwards to indicate whether the tx committed or rolled back.

In next transaction we start consuming messages will Kafka consumer start consuming same message which is already pushed to Kafka topic by producer in previous transactions. After successful flow of new transactions will duplicate messages wil be pushed again which was already pushed in previous transactions.

[garyrussell]

Don't ask questions on issues, use the Discussions tab or, preferably, on Stack Overflow for generic Kafka questions; this project is specific to Spring use of the kafka-clients.

It is not clear what you are asking; "pushed again" - that is purely a function of the sender and has nothing to do with Kafka itself.

Consumers will always start consuming from their last committed offset.