TheKnowles commented on a change in pull request #11382:
URL: https://github.com/apache/kafka/pull/11382#discussion_r763062207
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
##########
@@ -366,7 +367,11 @@ private boolean sendRecords() {
if (e != null) {
log.error("{} failed to send record to {}: ",
WorkerSourceTask.this, topic, e);
Review comment:
The error retry handling infrastructure predominantly concerns itself
with the sink side of the house. Insofar that any refactoring I would want to
do would probably necessitate a KIP on its own. To that end, I have added an
additional executeFailed() function to RetryWithToleranceOperator to allow the
source worker to handle error logging with all of the existing
infrastructure/configuration that exists for sink tasks.
I toy'ed around with the idea of having the new executeFailed() fire without
a tolerance type check. This would work for failing/ignoring as expected, but
with no mechanism to then decide if we should call commitRecord(). We could
block on the future from executeFailed() and then check withinToleranceLimits()
but that introduces non determinism with interrupt/execution exceptions.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
