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

MINOR: simplify producer TX abort error handling (#18486) 

Reviewers: Justine Olshan <jolshan@confluent.io>, Jason Gustafson <jason@responsive.dev>
This commit is contained in:
Matthias J. Sax 2025-01-10 17:54:40 -08:00 committed by GitHub
parent 32dbbe6a1f
commit f54cfff1dc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 4 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
View File

@ -462,17 +462,10 @@ public class Sender implements Runnable {
return true; return true;
} }
if (transactionManager.hasAbortableError() || transactionManager.isAborting()) { if (transactionManager.hasAbortableError()) {
if (accumulator.hasIncomplete()) { accumulator.abortUndrainedBatches(transactionManager.lastError());
// Attempt to get the last error that caused this abort. } else if (transactionManager.isAborting()) {
RuntimeException exception = transactionManager.lastError(); accumulator.abortUndrainedBatches(new TransactionAbortedException());
// If there was no error, but we are still aborting,
// then this is most likely a case where there was no fatal error.
if (exception == null) {
exception = new TransactionAbortedException();
}
accumulator.abortUndrainedBatches(exception);
}
} }
TransactionManager.TxnRequestHandler nextRequestHandler = transactionManager.nextRequest(accumulator.hasIncomplete()); TransactionManager.TxnRequestHandler nextRequestHandler = transactionManager.nextRequest(accumulator.hasIncomplete());