Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorTimerImpl.java (Java)

Code before:
        // expires, the operation is scheduled through the scheduler to be executed. This
        // ensures that the threading model of the runtime is respected.
        var task = new TimerTask(unit.toMillis(delay)) {
            @Override
            public void run() {
                var operationName = "Timeout(key=" + key + ")";

                scheduler.scheduleWriteOperation(
                    operationName,
                    () -> {
                        log.debug("Executing write event {} for timer {}.", operationName, key);

                        // If the task is different, it means that the timer has been
                        // cancelled while the event was waiting to be processed.
                        if (!tasks.remove(key, this)) {
                            throw new RejectedExecutionException("Timer " + key + " was overridden or cancelled");
                        }

                        // Execute the timeout operation.
                        return operation.generateRecords();
                    }
                ).exceptionally(ex -> {
                    // Exceptions may be wrapped in CompletionException or ExecutionException
                    // when propagated through CompletableFuture chains, so we unwrap them
                    // before checking types with instanceof.
                    ex = Errors.maybeUnwrapException(ex);

                    // Remove the task after a failure.
                    tasks.remove(key, this);

                    if (ex instanceof RejectedExecutionException) {
                        log.debug("The write event {} for the timer {} was not executed because it was " +
                            "cancelled or overridden.", operationName, key);
                        return null;
                    }

                    if (ex instanceof NotCoordinatorException || ex instanceof CoordinatorLoadInProgressException) {
                        log.debug("The write event {} for the timer {} failed due to {}. Ignoring it because " +
                            "the coordinator is not active.", operationName, key, ex.getMessage());
                        return null;
                    }

                    if (retry) {
                        log.info("The write event {} for the timer {} failed due to {}. Rescheduling it. ",
                            operationName, key, ex.getMessage());
                        schedule(key, retryBackoff, TimeUnit.MILLISECONDS, true, retryBackoff, operation);
                    } else {
                        log.error("The write event {} for the timer {} failed due to {}. Ignoring it. ",
                            operationName, key, ex.getMessage(), ex);
                    }


Review comment (suggestion):
Based on the current usage, it does not seems that we could get wrapped exceptions here but it is better to be on the safe side.

Code after:
        // expires, the operation is scheduled through the scheduler to be executed. This
        // ensures that the threading model of the runtime is respected.
        var task = new TimerTask(unit.toMillis(delay)) {
            @Override
            public void run() {
                var operationName = "Timeout(key=" + key + ")";

                scheduler.scheduleWriteOperation(
                    operationName,
                    () -> {
                        log.debug("Executing write event {} for timer {}.", operationName, key);

                        // If the task is different, it means that the timer has been
                        // cancelled while the event was waiting to be processed.
                        if (!tasks.remove(key, this)) {
                            throw new RejectedExecutionException("Timer " + key + " was overridden or cancelled");
                        }

                        // Execute the timeout operation.
                        return operation.generateRecords();
                    }
                ).exceptionally(ex -> {
                    // Exceptions may be wrapped in CompletionException when propagated
                    // through CompletableFuture chains, so we unwrap them before
                    // checking types with instanceof.
                    ex = Errors.maybeUnwrapException(ex);

                    // Remove the task after a failure.
                    tasks.remove(key, this);

                    if (ex instanceof RejectedExecutionException) {
                        log.debug("The write event {} for the timer {} was not executed because it was " +
                            "cancelled or overridden.", operationName, key);
                        return null;
                    }

                    if (ex instanceof NotCoordinatorException || ex instanceof CoordinatorLoadInProgressException) {
                        log.debug("The write event {} for the timer {} failed due to {}. Ignoring it because " +
                            "the coordinator is not active.", operationName, key, ex.getMessage());
                        return null;
                    }

                    if (retry) {
                        log.info("The write event {} for the timer {} failed due to {}. Rescheduling it. ",
                            operationName, key, ex.getMessage());
                        schedule(key, retryBackoff, TimeUnit.MILLISECONDS, true, retryBackoff, operation);
                    } else {
                        log.error("The write event {} for the timer {} failed due to {}. Ignoring it. ",
                            operationName, key, ex.getMessage(), ex);
                    }