private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final OffsetFetchRequestState request) {
            Optional<OffsetFetchRequestState> dupe =
                    unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
            Optional<OffsetFetchRequestState> inflight =
                    inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();

            if (dupe.isPresent() || inflight.isPresent()) {
                log.debug("Duplicated unsent offset fetch request found for partitions: {}", request.requestedPartitions);
                dupe.orElseGet(inflight::get).chainFuture(request.future);
            } else {
                log.debug("Enqueuing offset fetch request for partitions: {}", request.requestedPartitions);
                this.unsentOffsetFetches.add(request);
            }
            return request.future;
        }
<p>Adding an offset fetch request to the outgoing buffer.  If the same request was made, we chain the future
to the existing one.

<p>If the request is new, it invokes a callback to remove itself from the {@code inflightOffsetFetches}
upon completion.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java