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;
    }
@Test
    public void testExecuteFailedNoTolerance() {
        RetryWithToleranceOperator<ConsumerRecord<byte[], byte[]>> retryWithToleranceOperator = genericOperator(0, NONE, errorHandlingMetrics);

        ProcessingContext<ConsumerRecord<byte[], byte[]>> context = new ProcessingContext<>(consumerRecord);
        assertThrows(ConnectException.class, () -> retryWithToleranceOperator.executeFailed(context, Stage.TASK_PUT,
            SinkTask.class, new Throwable()));
    }