public void reap(long currentTimeMs) {
        Consumer<CompletableEvent<?>> expireEvent = event -> {
            long pastDueMs = currentTimeMs - event.deadlineMs();
            TimeoutException error = new TimeoutException(String.format("%s was %s ms past its expiration of %s", event.getClass().getSimpleName(), pastDueMs, event.deadlineMs()));

            if (event.future().completeExceptionally(error)) {
                log.debug("Event {} completed exceptionally since its expiration of {} passed {} ms ago", event, event.deadlineMs(), pastDueMs);
            } else {
                log.trace("Event {} not completed exceptionally since it was previously completed", event);
            }
        };

        // First, complete (exceptionally) any events that have passed their deadline AND aren't already complete.
        tracked.stream()
            .filter(e -> !e.future().isDone())
            .filter(e -> currentTimeMs >= e.deadlineMs())
            .forEach(expireEvent);
        // Second, remove any events that are already complete, just to make sure we don't hold references. This will
        // include any events that finished successfully as well as any events we just completed exceptionally above.
        tracked.removeIf(e -> e.future().isDone());
    }
This method performs a two-step process to "complete" {@link CompletableEvent events} that have either expired
or completed normally:

<ol>
    <li>
        For each tracked event which has exceeded its {@link CompletableEvent#deadlineMs() deadline}, an
        instance of {@link TimeoutException} is created and passed to
        {@link CompletableFuture#completeExceptionally(Throwable)}.
    </li>
    <li>
        For each tracked event of which its {@link CompletableEvent#future() future} is already in the
        {@link CompletableFuture#isDone() done} state, it will be removed from the list of tracked events.
    </li>
</ol>

<p/>

This method should be called at regular intervals, based upon the needs of the resource that owns the reaper.

@param currentTimeMs <em>Current</em> time with which to compare against the
                     <em>{@link CompletableEvent#deadlineMs() expiration time}</em>
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java