public CompletableFuture<Void> commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets) {
        if (offsets.isEmpty()) {
            log.debug("Skipping commit of empty offsets");
            return CompletableFuture.completedFuture(null);
        }
        OffsetCommitRequestState commitRequest = createOffsetCommitRequest(offsets, Long.MAX_VALUE);
        pendingRequests.addOffsetCommitRequest(commitRequest);

        CompletableFuture<Void> asyncCommitResult = new CompletableFuture<>();
        commitRequest.future.whenComplete((committedOffsets, error) -> {
            if (error != null) {
                asyncCommitResult.completeExceptionally(commitAsyncExceptionForError(error));
            } else {
                asyncCommitResult.complete(null);
            }
        });
        return asyncCommitResult;
    }
@Test
    public void testPollSkipIfCoordinatorUnknown() {
        CommitRequestManager commitRequestManager = create(false, 0);
        assertPoll(false, 0, commitRequestManager);

        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
        commitRequestManager.commitAsync(offsets);
        assertPoll(false, 0, commitRequestManager);
    }