@Override
    ErroneousAndValidPartitionData getErroneousAndValidTopicIdPartitions() {
        if (!isSubsequent) {
            return new ErroneousAndValidPartitionData(shareFetchData);
        }
        List<Tuple2<TopicIdPartition, PartitionData>> erroneous = new ArrayList<>();
        List<Tuple2<TopicIdPartition, ShareFetchRequest.SharePartitionData>> valid = new ArrayList<>();
        // Take the session lock and iterate over all the cached partitions.
        synchronized (session) {
            session.partitionMap().forEach(cachedSharePartition -> {
                TopicIdPartition topicIdPartition = new TopicIdPartition(cachedSharePartition.topicId(), new
                        TopicPartition(cachedSharePartition.topic(), cachedSharePartition.partition()));
                ShareFetchRequest.SharePartitionData reqData = cachedSharePartition.reqData();
                if (topicIdPartition.topic() == null) {
                    erroneous.add(new Tuple2<>(topicIdPartition, ShareFetchResponse.partitionResponse(topicIdPartition, Errors.UNKNOWN_TOPIC_ID)));
                } else {
                    valid.add(new Tuple2<>(topicIdPartition, reqData));
                }
            });
            return new ErroneousAndValidPartitionData(erroneous, valid);
        }
    }
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