protected <V> V execAndHandleError(ProcessingContext<T> context, Operation<V> operation, Class<? extends Exception> tolerated) {
        try {
            V result = execAndRetry(context, operation);
            if (context.failed()) {
                markAsFailed();
                errorHandlingMetrics.recordSkipped();
            }
            return result;
        } catch (Exception e) {
            errorHandlingMetrics.recordFailure();
            markAsFailed();
            context.error(e);

            if (!tolerated.isAssignableFrom(e.getClass())) {
                throw new ConnectException("Unhandled exception in error handler", e);
            }

            if (!withinToleranceLimits()) {
                throw new ConnectException("Tolerance exceeded in error handler", e);
            }

            errorHandlingMetrics.recordSkipped();
            return null;
        }
    }
Attempt to execute an operation. Handles retriable and tolerated exceptions thrown by the operation.
<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 operation the operation to be executed.
@param tolerated the class of exceptions which can be tolerated if errors.tolerance=all
@param <V> The return type of the result of the operation.
@return the result of the operation, or null if 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