C0urante commented on code in PR #13379:
URL: https://github.com/apache/kafka/pull/13379#discussion_r1135720026
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java:
##########
@@ -483,15 +483,19 @@ protected boolean
shouldCommitTransactionForRecord(SourceRecord record) {
if (transactionContext.shouldAbortOn(record)) {
log.info("Aborting transaction for record on topic
{} as requested by connector", record.topic());
log.trace("Last record in aborted transaction:
{}", record);
- abortTransaction();
+ maybeAbortTransaction();
Review Comment:
I don't think so; the only time we invoke `Producer::beginTransaction` is
[here](https://github.com/apache/kafka/blob/ccfc389a638a126c3769bdd72725bae532ca4d01/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java#L248-L249),
after which we immediately set `transactionOpen` to `true`, and the only time
we set `transactionOpen` to `false` is after either aborting a transaction
(which takes place in this method), or committing the transaction (which takes
place
[here](https://github.com/apache/kafka/blob/ccfc389a638a126c3769bdd72725bae532ca4d01/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java#L311).
In either case, if we fail the operation (aborting/committing the
transaction), the task fails, and we [skip the end-of-life offset
commit](https://github.com/apache/kafka/blob/ccfc389a638a126c3769bdd72725bae532ca4d01/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSour
ceTask.java#L219-L222).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
