public <V> V execute(ProcessingContext<T> context, Operation<V> operation, Stage stage, Class<?> executingClass) {
        context.currentContext(stage, executingClass);
        if (context.failed()) {
            log.debug("ProcessingContext is already in failed state. Ignoring requested operation.");
            return null;
        }
        context.currentContext(stage, executingClass);
        try {
            Class<? extends Exception> ex = TOLERABLE_EXCEPTIONS.getOrDefault(context.stage(), RetriableException.class);
            return execAndHandleError(context, operation, ex);
        } finally {
            if (context.failed()) {
                errorHandlingMetrics.recordError();
                report(context);
            }
        }
    }
Attempt to execute an operation. Handles retriable and tolerated exceptions thrown by the operation. This is
used for small blocking operations which can be represented as an {@link Operation}. For operations which do
not fit this interface, see {@link #executeFailed(ProcessingContext, Stage, Class, Throwable)}

<p>If any error is already present in the context, return null without modifying the context.
<p>Retries are allowed if the operator is still running, and this operation is within the error retry timeout.
<p>Tolerable exceptions are different for each stage, and encoded in {@link #TOLERABLE_EXCEPTIONS}
<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 recoverable operation
@param stage The logical stage within the overall pipeline of the operation that has failed
@param executingClass The class containing the operation implementation that failed
@param <V> return type of the result of the operation.
@return result of the operation, or null if a prior exception occurred, or the operation only threw retriable or tolerable exceptions
@throws ConnectException wrapper if any non-tolerated exception was thrown by the operation
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java