vamossagar12 commented on code in PR #13726:
URL: https://github.com/apache/kafka/pull/13726#discussion_r1218215790
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java:
##########
@@ -458,13 +458,8 @@ boolean sendRecords() {
}
protected List<SourceRecord> poll() throws InterruptedException {
- try {
- return task.poll();
- } catch (RetriableException |
org.apache.kafka.common.errors.RetriableException e) {
- log.warn("{} failed to poll records from SourceTask. Will retry
operation.", this, e);
- // Do nothing. Let the framework poll whenever it's ready.
- return null;
- }
+ retryWithToleranceOperator.reset();
+ return retryWithToleranceOperator.execute(task::poll, Stage.TASK_POLL,
this.getClass());
Review Comment:
Also, for task.poll() there is not records.skipped that is applicable. We
either get records or don't get them, so there is not question of skipping
them. Rest of the metrics, IMHO still point to operation level failures and are
usable as is. Let me know what do you think.
--
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
