private PollResult processAcknowledgements(long currentTimeMs) {
        List<UnsentRequest> unsentRequests = new ArrayList<>();
        Iterator<AcknowledgeRequestState> iterator = acknowledgeRequestStates.iterator();
        while (iterator.hasNext()) {
            AcknowledgeRequestState acknowledgeRequestState = iterator.next();
            if (acknowledgeRequestState.isProcessed()) {
                iterator.remove();
            } else if (!acknowledgeRequestState.maybeExpire()) {
                if (nodesWithPendingRequests.contains(acknowledgeRequestState.nodeId)) {
                    log.trace("Skipping acknowledge request because previous request to {} has not been processed", acknowledgeRequestState.nodeId);
                } else {
                    if (acknowledgeRequestState.canSendRequest(currentTimeMs)) {
                        acknowledgeRequestState.onSendAttempt(currentTimeMs);
                        UnsentRequest request = acknowledgeRequestState.buildRequest(currentTimeMs);
                        if (request != null) {
                            unsentRequests.add(request);
                        }
                    }
                }
            } else {
                // Fill in TimeoutException
                for (TopicIdPartition tip : acknowledgeRequestState.acknowledgementsMap.keySet()) {
                    metricsManager.recordFailedAcknowledgements(acknowledgeRequestState.getAcknowledgementsCount(tip));
                    acknowledgeRequestState.handleAcknowledgeErrorCode(tip, Errors.REQUEST_TIMED_OUT);
                }
                iterator.remove();
            }
        }

        PollResult pollResult = null;
        if (!unsentRequests.isEmpty()) {
            pollResult = new PollResult(unsentRequests);
        } else if (!acknowledgeRequestStates.isEmpty()) {
            // Return empty result until all the acknowledgement request states are processed
            pollResult = PollResult.EMPTY;
        } else if (closing) {
            if (!closeFuture.isDone()) {
                log.trace("Completing acknowledgement on close");
                closeFuture.complete(null);
            }
            pollResult = PollResult.EMPTY;
        }

        return pollResult;
    }
Process acknowledgeRequestStates and prepares a list of acknowledgements to be sent in the poll().

@param currentTimeMs the current time in ms.

@return the PollResult containing zero or more acknowledgements.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java