public CompletableFuture<Void> acknowledgeOnClose(final Map<TopicIdPartition, Acknowledgements> acknowledgementsMap,
                                                      final long deadlineMs) {
        final Cluster cluster = metadata.fetch();
        final AtomicInteger resultCount = new AtomicInteger();
        final ResultHandler resultHandler = new ResultHandler(resultCount, Optional.empty());

        closing = true;

        sessionHandlers.forEach((nodeId, sessionHandler) -> {
            Node node = cluster.nodeById(nodeId);
            if (node != null) {
                Map<TopicIdPartition, Acknowledgements> acknowledgementsMapForNode = new HashMap<>();
                for (TopicIdPartition tip : sessionHandler.sessionPartitions()) {
                    Acknowledgements acknowledgements = acknowledgementsMap.getOrDefault(tip, Acknowledgements.empty());

                    if (fetchAcknowledgementsMap.get(tip) != null) {
                        acknowledgements.merge(fetchAcknowledgementsMap.remove(tip));
                    }

                    if (acknowledgements != null && !acknowledgements.isEmpty()) {
                        acknowledgementsMapForNode.put(tip, acknowledgements);

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

                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 close() when there is an existing sync request for node {}-{}", node.id(), acknowledgeRequestStates.get(nodeId).getSyncRequest());
                    closeFuture.completeExceptionally(
                            new IllegalStateException("Attempt to call close() when there is an existing sync request for node : " + node.id()));
                } else {
                    // There can only be one commitSync()/close() happening at a time. So per node, there will be one acknowledge request state.
                    acknowledgeRequestStates.get(nodeId).setSyncRequest(new AcknowledgeRequestState(logContext,
                            ShareConsumeRequestManager.class.getSimpleName() + ":3",
                            deadlineMs,
                            retryBackoffMs,
                            retryBackoffMaxMs,
                            sessionHandler,
                            nodeId,
                            acknowledgementsMapForNode,
                            this::handleShareAcknowledgeCloseSuccess,
                            this::handleShareAcknowledgeCloseFailure,
                            resultHandler,
                            AcknowledgeRequestType.CLOSE
                    ));

                }
            }
        });

        resultHandler.completeIfEmpty();
        return closeFuture;
    }
@Test
    public void testAcknowledgeOnClose() {
        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);

        // Piggyback acknowledgements
        shareConsumeRequestManager.fetch(Collections.singletonMap(tip0, acknowledgements));

        // Remaining acknowledgements sent with close().
        Acknowledgements acknowledgements2 = Acknowledgements.empty();
        acknowledgements2.add(2L, AcknowledgeType.ACCEPT);
        acknowledgements2.add(3L, AcknowledgeType.REJECT);

        shareConsumeRequestManager.acknowledgeOnClose(Collections.singletonMap(tip0, acknowledgements2),
                calculateDeadlineMs(time.timer(100)));

        assertEquals(1, shareConsumeRequestManager.sendAcknowledgements());

        client.prepareResponse(fullAcknowledgeResponse(tip0, Errors.NONE));
        networkClientDelegate.poll(time.timer(0));
        assertEquals(1, completedAcknowledgements.size());

        Acknowledgements mergedAcks = acknowledgements.merge(acknowledgements2);
        mergedAcks.setAcknowledgeErrorCode(Errors.NONE);
        // Verifying that all 3 offsets were acknowledged as part of the final ShareAcknowledge on close.
        assertEquals(mergedAcks.getAcknowledgementsTypeMap(), completedAcknowledgements.get(0).get(tip0).getAcknowledgementsTypeMap());
        assertTrue(shareConsumeRequestManager.hasCompletedFetches());
    }