Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java (Java)

Code before:
     * For all the {@link CompletableEvent}s in the collection, if they're not already complete, invoke
     * {@link CompletableFuture#completeExceptionally(Throwable)}.
     *
     * @param events Collection of objects, assumed to be subclasses of {@link ApplicationEvent} or
     *               {@link BackgroundEvent}, but will only perform completion for any
     *               unfinished {@link CompletableEvent}s
     *
     *  @return Number of events closed
     */
    private long completeEventsExceptionallyOnClose(Collection<?> events) {
        long count = 0;

        for (Object o : events) {
            if (!(o instanceof CompletableEvent))
                continue;

            CompletableEvent<?> event = (CompletableEvent<?>) o;

            if (event.future().isDone())
                continue;

            TimeoutException error = new TimeoutException(String.format("%s could not be completed before the consumer closed", event.getClass().getSimpleName()));

            if (event.future().completeExceptionally(error)) {
                log.debug("Event {} completed exceptionally since the consumer is closing", event);
                count++;
            } else {
                log.trace("Event {} not completed exceptionally since it was completed prior to the consumer closing", event);
            }
        }

        return count;
    }
}


Review comment (suggestion):
incrementing here is not exactly the same that it used to happen before, right? 

Before the PR, the count inc would happen after the `peek(expireEvent)`, so it would count events that were expired but not completedExceptionally (the ones going through line 202, where now we don't increment the count)

This count doesn't seem used anyways, but better to be accurate and consistent with what it's done in the `reap` func (there the count includes all expired events, no matter if we completed them exceptionally or not) https://github.com/apache/kafka/blob/2b2552da1fbcdb687c99a03a81034c8f78e86871/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java#L115  

Code after:

    /**
     * For all the {@link CompletableEvent}s in the collection, if they're not already complete, invoke
     * {@link CompletableFuture#completeExceptionally(Throwable)}.
     *
     * @param events Collection of objects, assumed to be subclasses of {@link ApplicationEvent} or
     *               {@link BackgroundEvent}, but will only perform completion for any
     *               unfinished {@link CompletableEvent}s
     *
     * @return Number of events closed
     */
    private long completeEventsExceptionallyOnClose(Collection<?> events) {
        long count = 0;

        for (Object o : events) {
            if (!(o instanceof CompletableEvent))
                continue;

            CompletableEvent<?> event = (CompletableEvent<?>) o;

            if (event.future().isDone())
                continue;

            count++;

            TimeoutException error = new TimeoutException(String.format("%s could not be completed before the consumer closed", event.getClass().getSimpleName()));

            if (event.future().completeExceptionally(error)) {
                log.debug("Event {} completed exceptionally since the consumer is closing", event);
            } else {
                log.trace("Event {} not completed exceptionally since it was completed prior to the consumer closing", event);
            }
        }

        return count;
    }
}