Repository: apache/kafka
File: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/errors/RetryWithToleranceOperator.java (Java)

Code before:
     * @param error The error which caused the operation to fail
     * @return A future which resolves when this failure has been persisted by all {@link ErrorReporter} instances
     * @throws ConnectException if the operation is not tolerated, and the overall pipeline should stop
     */
    public Future<Void> executeFailed(ProcessingContext<T> context, Stage stage, Class<?> executingClass, Throwable error) {
        markAsFailed();
        context.currentContext(stage, executingClass);
        context.error(error);
        errorHandlingMetrics.recordFailure();
        Future<Void> errantRecordFuture = report(context);
        if (!withinToleranceLimits()) {
            errorHandlingMetrics.recordError();
            throw new ConnectException("Tolerance exceeded in error handler", error);
        }
        return errantRecordFuture;
    }

    /**
     * Report an error to all configured {@link ErrorReporter} instances.
     * @param context The context containing details of the error to report
     * @return A future which resolves when this failure has been persisted by all {@link ErrorReporter} instances
     */
    // Visible for testing
    synchronized Future<Void> report(ProcessingContext<T> context) {
        if (reporters.size() == 1) {
            return new WorkerErrantRecordReporter.ErrantRecordFuture(List.of(reporters.iterator().next().report(context)));
        }
        List<Future<RecordMetadata>> futures = reporters.stream()
                .map(r -> r.report(context))
                .filter(f -> !f.isDone())
                .collect(Collectors.toList());
        if (futures.isEmpty()) {
            return CompletableFuture.completedFuture(null);
        }
        return new WorkerErrantRecordReporter.ErrantRecordFuture(futures);
    }

    /**
     * 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

Review comment (suggestion):
```java
            return new WorkerErrantRecordReporter.ErrantRecordFuture(List.of(reporters.get(0).report(context)));
```

Code after:
     * @param error The error which caused the operation to fail
     * @return A future which resolves when this failure has been persisted by all {@link ErrorReporter} instances
     * @throws ConnectException if the operation is not tolerated, and the overall pipeline should stop
     */
    public Future<Void> executeFailed(ProcessingContext<T> context, Stage stage, Class<?> executingClass, Throwable error) {
        markAsFailed();
        context.currentContext(stage, executingClass);
        context.error(error);
        errorHandlingMetrics.recordFailure();
        Future<Void> errantRecordFuture = report(context);
        if (!withinToleranceLimits()) {
            errorHandlingMetrics.recordError();
            throw new ConnectException("Tolerance exceeded in error handler", error);
        }
        return errantRecordFuture;
    }

    /**
     * Report an error to all configured {@link ErrorReporter} instances.
     * @param context The context containing details of the error to report
     * @return A future which resolves when this failure has been persisted by all {@link ErrorReporter} instances
     */
    // Visible for testing
    synchronized Future<Void> report(ProcessingContext<T> context) {
        if (reporters.size() == 1) {
            return new WorkerErrantRecordReporter.ErrantRecordFuture(List.of(reporters.get(0).report(context)));
        }
        List<Future<RecordMetadata>> futures = reporters.stream()
                .map(r -> r.report(context))
                .filter(f -> !f.isDone())
                .collect(Collectors.toList());
        if (futures.isEmpty()) {
            return CompletableFuture.completedFuture(null);
        }
        return new WorkerErrantRecordReporter.ErrantRecordFuture(futures);
    }

    /**
     * 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