public CompletableFuture<Map<TopicIdPartition, Acknowledgements>> commitSync(
            final Map<TopicIdPartition, Acknowledgements> acknowledgementsMap,
            final long deadlineMs) {
        final AtomicInteger resultCount = new AtomicInteger();
        final CompletableFuture<Map<TopicIdPartition, Acknowledgements>> future = new CompletableFuture<>();
        final ResultHandler resultHandler = new ResultHandler(resultCount, Optional.of(future));

        final Cluster cluster = metadata.fetch();

        sessionHandlers.forEach((nodeId, sessionHandler) -> {
            Node node = cluster.nodeById(nodeId);
            if (node != null) {
                acknowledgeRequestStates.putIfAbsent(nodeId, new Pair<>(null, null));

                // Ensure there is no commitSync()/close() request already present as they are blocking calls
                // and only one request can be active at a time.
                if (acknowledgeRequestStates.get(nodeId).getSyncRequest() != null && !acknowledgeRequestStates.get(nodeId).getSyncRequest().isEmpty()) {
                    log.error("Attempt to call commitSync() when there is an existing sync request for node {}", node.id());
                    future.completeExceptionally(
                            new IllegalStateException("Attempt to call commitSync() when there is an existing sync request for node : " + node.id()));
                } else {
                    Map<TopicIdPartition, Acknowledgements> acknowledgementsMapForNode = new HashMap<>();
                    for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
                        Acknowledgements acknowledgements = acknowledgementsMap.get(tip);
                        if (acknowledgements != null) {
                            acknowledgementsMapForNode.put(tip, acknowledgements);

                            metricsManager.recordAcknowledgementSent(acknowledgements.size());
                            log.debug("Added sync acknowledge request for partition {} to node {}", tip.topicPartition(), node.id());
                            resultCount.incrementAndGet();
                        }
                    }


                    // There can only be one commitSync()/close() happening at a time. So per node, there will be one acknowledge request state representing commitSync() and close().
                    acknowledgeRequestStates.get(nodeId).setSyncRequest(new AcknowledgeRequestState(logContext,
                            ShareConsumeRequestManager.class.getSimpleName() + ":1",
                            deadlineMs,
                            retryBackoffMs,
                            retryBackoffMaxMs,
                            sessionHandler,
                            nodeId,
                            acknowledgementsMapForNode,
                            this::handleShareAcknowledgeSuccess,
                            this::handleShareAcknowledgeFailure,
                            resultHandler,
                            AcknowledgeRequestType.COMMIT_SYNC
                    ));
                }
            }
        });

        resultHandler.completeIfEmpty();
        return future;
    }
@Test
    public void testCommitSync() {
        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.commitSync(Collections.singletonMap(tip0, acknowledgements), time.milliseconds() + 2000);

        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();
    }