public ShareFetchRequest.Builder newShareFetchBuilder(String groupId, FetchConfig fetchConfig) {
        List<TopicIdPartition> added = new ArrayList<>();
        List<TopicIdPartition> removed = new ArrayList<>();
        List<TopicIdPartition> replaced = new ArrayList<>();

        if (nextMetadata.isNewSession()) {
            // Add any new partitions to the session
            for (Entry<TopicPartition, TopicIdPartition> entry : nextPartitions.entrySet()) {
                TopicPartition topicPartition = entry.getKey();
                TopicIdPartition topicIdPartition = entry.getValue();
                sessionPartitions.put(topicPartition, topicIdPartition);
            }

            // If it's a new session, all the partitions must be added to the request
            added.addAll(sessionPartitions.values());
        } else {
            // Iterate over the session partitions, tallying which were added
            Iterator<Entry<TopicPartition, TopicIdPartition>> partitionIterator = sessionPartitions.entrySet().iterator();
            while (partitionIterator.hasNext()) {
                Entry<TopicPartition, TopicIdPartition> entry = partitionIterator.next();
                TopicPartition topicPartition = entry.getKey();
                TopicIdPartition prevData = entry.getValue();
                TopicIdPartition nextData = nextPartitions.remove(topicPartition);
                if (nextData != null) {
                    // If the topic ID does not match, the topic has been recreated
                    if (!prevData.equals(nextData)) {
                        nextPartitions.put(topicPartition, nextData);
                        entry.setValue(nextData);
                        replaced.add(prevData);
                    }
                } else {
                    // This partition is not in the builder, so we need to remove it from the session
                    partitionIterator.remove();
                    removed.add(prevData);
                }
            }

            // Add any new partitions to the session
            for (Entry<TopicPartition, TopicIdPartition> entry : nextPartitions.entrySet()) {
                TopicPartition topicPartition = entry.getKey();
                TopicIdPartition topicIdPartition = entry.getValue();
                sessionPartitions.put(topicPartition, topicIdPartition);
                added.add(topicIdPartition);
            }
        }

        if (log.isDebugEnabled()) {
            log.debug("Build ShareFetch {} for node {}. Added {}, removed {}, replaced {} out of {}",
                    nextMetadata, node,
                    topicIdPartitionsToLogString(added),
                    topicIdPartitionsToLogString(removed),
                    topicIdPartitionsToLogString(replaced),
                    topicIdPartitionsToLogString(sessionPartitions.values()));
        }

        // The replaced topic-partitions need to be removed, and their replacements are already added
        removed.addAll(replaced);

        Map<TopicIdPartition, List<ShareFetchRequestData.AcknowledgementBatch>> acknowledgementBatches = new HashMap<>();
        nextAcknowledgements.forEach((partition, acknowledgements) -> acknowledgementBatches.put(partition, acknowledgements.getAcknowledgementBatches()
                .stream().map(AcknowledgementBatch::toShareFetchRequest)
                .collect(Collectors.toList())));

        nextPartitions = new LinkedHashMap<>();
        nextAcknowledgements = new LinkedHashMap<>();

        return ShareFetchRequest.Builder.forConsumer(
                groupId, nextMetadata, fetchConfig.maxWaitMs,
                fetchConfig.minBytes, fetchConfig.maxBytes, fetchConfig.fetchSize,
                added, removed, acknowledgementBatches);
    }
All the partitions in the share session.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandler.java