public void reap(Collection<?> events) {
        Objects.requireNonNull(events, "Event queue to reap must be non-null");

        Consumer<CompletableEvent<?>> expireEvent = event -> {
            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);
            }
        };

        tracked.stream()
            .filter(e -> !e.future().isDone())
            .forEach(expireEvent);
        tracked.clear();

        events.stream()
            .filter(e -> e instanceof CompletableEvent<?>)
            .map(e -> (CompletableEvent<?>) e)
            .filter(e -> !e.future().isDone())
            .forEach(expireEvent);
        events.clear();
    }
It is possible for the {@link AsyncKafkaConsumer#close() consumer to close} before completing the processing of
all the events in the queue. In this case, we need to
{@link CompletableFuture#completeExceptionally(Throwable) expire} any remaining events.

<p/>

Check each of the {@link #add(CompletableEvent) previously-added} {@link CompletableEvent completable events},
and for any that are incomplete, expire them. Also check the core event queue for any incomplete events and
likewise expire them.

<p/>

<em>Note</em>: because this is called in the context of {@link AsyncKafkaConsumer#close() closing consumer},
don't take the deadline into consideration, just close it regardless.

@param events Events from a queue that have not yet been tracked that also need to be reviewed
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java