@Override
    ShareFetchResponse updateAndGenerateResponseData(String groupId, Uuid memberId,
                                                     LinkedHashMap<TopicIdPartition, ShareFetchResponseData.PartitionData> updates) {
        if (!isSubsequent) {
            return new ShareFetchResponse(ShareFetchResponse.toMessage(
                    Errors.NONE, 0, updates.entrySet().iterator(), Collections.emptyList()));
        } else {
            int expectedEpoch = ShareFetchMetadata.nextEpoch(reqMetadata.epoch());
            int sessionEpoch;
            synchronized (session) {
                sessionEpoch = session.epoch;
            }
            if (sessionEpoch != expectedEpoch) {
                log.debug("Subsequent share session {} expected epoch {}, but got {}. Possible duplicate request.",
                        session.key(), expectedEpoch, sessionEpoch);
                return new ShareFetchResponse(ShareFetchResponse.toMessage(Errors.INVALID_SHARE_SESSION_EPOCH,
                        0, Collections.emptyIterator(), Collections.emptyList()));
            }
            // Iterate over the update list using PartitionIterator. This will prune updates which don't need to be sent
            Iterator<Map.Entry<TopicIdPartition, ShareFetchResponseData.PartitionData>> partitionIterator = new PartitionIterator(
                    updates.entrySet().iterator(), true);
            while (partitionIterator.hasNext()) {
                partitionIterator.next();
            }
            log.debug("Subsequent share session context with session key {} returning {}", session.key(),
                    partitionsToLogString(updates.keySet()));
            return new ShareFetchResponse(ShareFetchResponse.toMessage(
                    Errors.NONE, 0, updates.entrySet().iterator(), Collections.emptyList()));
        }
    }
Iterator that goes over the given partition map and selects partitions that need to be included in the response.
If updateShareContextAndRemoveUnselected is set to true, the share context will be updated for the selected
partitions and also remove unselected ones as they are encountered.
core/src/main/java/kafka/server/share/ShareSessionContext.java