protected <V> V execAndRetry(ProcessingContext<T> context, Operation<V> operation) throws Exception {
        int attempt = 0;
        long startTime = time.milliseconds();
        long deadline = (errorRetryTimeout >= 0) ? startTime + errorRetryTimeout : Long.MAX_VALUE;
        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;
                }
            } finally {
                context.attempt(attempt);
            }
        } while (true);
    }
Attempt to execute an operation. Handles retriable exceptions raised by the operation.
<p>Retries are allowed if the operator is still running, and this operation is within the error retry timeout.
<p>This method mutates the passed-in {@link ProcessingContext} with the number of attempts made to execute the
operation, and the last error encountered if no attempt was successful.

@param context The {@link ProcessingContext} used to hold state about this operation
@param operation the operation to be executed.
@param <V> the return type of the result of the operation.
@return the result of the operation if it succeeded, or null if the operation only threw retriable exceptions
@throws Exception rethrow if any non-retriable exception was thrown by the operation
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java