C0urante commented on a change in pull request #11323:
URL: https://github.com/apache/kafka/pull/11323#discussion_r720395810
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
##########
@@ -570,22 +512,21 @@ public boolean commitOffsets() {
// could look a little confusing.
} catch (InterruptedException e) {
log.warn("{} Flush of offsets interrupted, cancelling", this);
- finishFailedFlush();
+ offsetWriter.cancelFlush();
recordCommitFailure(time.milliseconds() - started, e);
return false;
} catch (ExecutionException e) {
log.error("{} Flush of offsets threw an unexpected exception: ",
this, e);
- finishFailedFlush();
+ offsetWriter.cancelFlush();
recordCommitFailure(time.milliseconds() - started, e);
return false;
} catch (TimeoutException e) {
log.error("{} Timed out waiting to flush offsets to storage",
this);
- finishFailedFlush();
+ offsetWriter.cancelFlush();
Review comment:
I think a retry notice is fine. The term "new offsets" may be a little
misleading since the offsets that we failed to commit may still be partially or
entirely present the next time we try to commit (they're tracked by the offset
writer class and retained when `cancelFlush` is invoked).
WDYT about "latest" instead of "new"?
--
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
