root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: Remove flakiness caused when producing aborted transaction in ShareConsumerTest (#19824) 

This PR attempts to removed the flakiness in
`testAlterReadCommittedToReadUncommittedIsolationLevelWithReleaseAck`
and `testAlterReadCommittedToReadUncommittedIsolationLevelWithRejectAck`.
This flakiness could potentially be caused because we were not ensuring
that the aborted transaction record produce happened. In this PR, I have
added a blocking call to make sure the produce future completes before
we abort the transaction.

Reviewers: Andrew Schofield <aschofield@confluent.io>
This commit is contained in:
Abhinav Dixit 2025-05-27 17:22:07 +05:30 committed by GitHub
parent 0013d78298
commit d9233d2f16
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 2 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java
View File

@ -2623,8 +2623,9 @@ public class ShareConsumerTest {
try { try {
transactionalProducer.beginTransaction(); transactionalProducer.beginTransaction();
ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, message.getBytes(), message.getBytes()); ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, message.getBytes(), message.getBytes());
transactionalProducer.send(record); Future<RecordMetadata> future = transactionalProducer.send(record);
transactionalProducer.flush(); transactionalProducer.flush();
future.get(); // Ensure producer send is complete before aborting
transactionalProducer.abortTransaction(); transactionalProducer.abortTransaction();
} catch (Exception e) { } catch (Exception e) {
transactionalProducer.abortTransaction(); transactionalProducer.abortTransaction();