public void commitAsync(final Map<TopicIdPartition, Acknowledgements> acknowledgementsMap) {
        final Cluster cluster = metadata.fetch();
        final ResultHandler resultHandler = new ResultHandler(Optional.empty());

        sessionHandlers.forEach((nodeId, sessionHandler) -> {
            Node node = cluster.nodeById(nodeId);
            if (node != null) {
                Map<TopicIdPartition, Acknowledgements> acknowledgementsMapForNode = new HashMap<>();

                acknowledgeRequestStates.putIfAbsent(nodeId, new Pair<>(null, null));

                for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
                    Acknowledgements acknowledgements = acknowledgementsMap.get(tip);
                    if (acknowledgements != null) {
                        acknowledgementsMapForNode.put(tip, acknowledgements);

                        metricsManager.recordAcknowledgementSent(acknowledgements.size());
                        log.debug("Added async acknowledge request for partition {} to node {}", tip.topicPartition(), node.id());
                        AcknowledgeRequestState asyncRequestState = acknowledgeRequestStates.get(nodeId).getAsyncRequest();
                        if (asyncRequestState == null) {
                            acknowledgeRequestStates.get(nodeId).setAsyncRequest(new AcknowledgeRequestState(logContext,
                                    ShareConsumeRequestManager.class.getSimpleName() + ":2",
                                    Long.MAX_VALUE,
                                    retryBackoffMs,
                                    retryBackoffMaxMs,
                                    sessionHandler,
                                    nodeId,
                                    acknowledgementsMapForNode,
                                    this::handleShareAcknowledgeSuccess,
                                    this::handleShareAcknowledgeFailure,
                                    resultHandler,
                                    AcknowledgeRequestType.COMMIT_ASYNC
                            ));
                        } else {
                            Acknowledgements prevAcks = asyncRequestState.acknowledgementsToSend.putIfAbsent(tip, acknowledgements);
                            if (prevAcks != null) {
                                asyncRequestState.acknowledgementsToSend.get(tip).merge(acknowledgements);
                            }
                        }
                    }
                }
            }
        });

        resultHandler.completeIfEmpty();
    }
@Test
    public void testCommitAsync() {
        buildRequestManager();

        assignFromSubscribed(Collections.singleton(tp0));

        // normal fetch
        assertEquals(1, sendFetches());
        assertFalse(shareConsumeRequestManager.hasCompletedFetches());

        client.prepareResponse(fullFetchResponse(tip0, records, acquiredRecords, Errors.NONE));
        networkClientDelegate.poll(time.timer(0));
        assertTrue(shareConsumeRequestManager.hasCompletedFetches());

        Acknowledgements acknowledgements = Acknowledgements.empty();
        acknowledgements.add(1L, AcknowledgeType.ACCEPT);
        acknowledgements.add(2L, AcknowledgeType.ACCEPT);
        acknowledgements.add(3L, AcknowledgeType.REJECT);

        shareConsumeRequestManager.commitAsync(Collections.singletonMap(tip0, acknowledgements));

        assertEquals(1, shareConsumeRequestManager.sendAcknowledgements());

        client.prepareResponse(fullAcknowledgeResponse(tip0, Errors.NONE));
        networkClientDelegate.poll(time.timer(0));
        assertTrue(shareConsumeRequestManager.hasCompletedFetches());

        assertEquals(Collections.singletonMap(tip0, acknowledgements), completedAcknowledgements.get(0));
        completedAcknowledgements.clear();
    }