[kafka-connect] Records silently dropped in RetryWithToleranceOperator

[NiteshDhanpal]

Even when the errors.tolerance=none param is set, RetryWithToleranceOperator returns a null and thereby drops the record silently in execAndRetry() fn

kafka/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java

Line 190 in 54eff6a

return null;

do {
            try {
                attempt++;
                return operation.call();
            } catch (RetriableException e) {
                log.trace("Caught a retriable exception while executing {} operation with {}", context.stage(), context.executingClass());
                errorHandlingMetrics.recordFailure();
                if (time.milliseconds() < deadline) {
                    backoff(attempt, deadline);
                    errorHandlingMetrics.recordRetry();
                } else {
                    log.trace("Can't retry. start={}, attempt={}, deadline={}", startTime, attempt, deadline);
                    context.error(e);
                    return null;
                }
                if (stopping) {
                    log.trace("Shutdown has been scheduled. Marking operation as failed.");
                    context.error(e);
                    return null;
                }