sudeshwasnik commented on code in PR #12784:
URL: https://github.com/apache/kafka/pull/12784#discussion_r1006768071
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java:
##########
@@ -280,16 +280,18 @@ private void commitTransaction() {
// Commit the transaction
// Blocks until all outstanding records have been sent and ack'd
- try {
- producer.commitTransaction();
- } catch (Throwable t) {
- log.error("{} Failed to commit producer transaction",
ExactlyOnceWorkerSourceTask.this, t);
- flushError.compareAndSet(null, t);
+ Throwable error = flushError.get();
+ if (error == null) {
Review Comment:
want to check for
https://github.com/apache/kafka/pull/12784/files#diff-6073f350e3c1c4ca7334407b32d162012543dad6bcc048cc970f3d11e852e5f2R274
first here.
If this was null, next error could be result of producer.commitTransaction
itself, so we need to check for that error again.
Error resulting from any of those should throw exception... that's the
intention
lmk if i am missing your suggestion 😅
--
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
