UnsentRequest buildRequest(long currentTimeMs) {
            // If this is the closing request, close the share session by setting the final epoch
            if (onClose) {
                sessionHandler.notifyClose();
            }

            for (Map.Entry<TopicIdPartition, Acknowledgements> entry : acknowledgementsMap.entrySet()) {
                sessionHandler.addPartitionToFetch(entry.getKey(), entry.getValue());
            }

            ShareAcknowledgeRequest.Builder requestBuilder = sessionHandler.newShareAcknowledgeBuilder(groupId, fetchConfig);
            Node nodeToSend = metadata.fetch().nodeById(nodeId);

            nodesWithPendingRequests.add(nodeId);

            BiConsumer<ClientResponse, Throwable> responseHandler = (clientResponse, error) -> {
                if (error != null) {
                    onFailedAttempt(currentTimeMs);
                    errorHandler.handle(nodeToSend, requestBuilder.data(), this, error, currentTimeMs);
                    processingComplete();
                } else {
                    successHandler.handle(nodeToSend, requestBuilder.data(), this, clientResponse, currentTimeMs);
                }
            };

            if (requestBuilder == null) {
                log.trace("Building ShareAcknowledge request to send to node {} failed", nodeToSend.id());
                handleSessionErrorCode(Errors.SHARE_SESSION_NOT_FOUND);
                return null;
            } else {
                log.trace("Building ShareAcknowledge request to send to node {}", nodeToSend.id());
                return new UnsentRequest(requestBuilder, Optional.of(nodeToSend)).whenComplete(responseHandler);
            }
        }
Whether this is the final acknowledge request state before the consumer closes.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java