@Override
        public boolean hasNext() {
            while ((nextElement == null) && iterator.hasNext()) {
                Map.Entry<TopicIdPartition, ShareFetchResponseData.PartitionData> element = iterator.next();
                TopicIdPartition topicPart = element.getKey();
                ShareFetchResponseData.PartitionData respData = element.getValue();
                synchronized (session) {
                    CachedSharePartition cachedPart = session.partitionMap().find(new CachedSharePartition(topicPart));
                    boolean mustRespond = cachedPart.maybeUpdateResponseData(respData, updateShareContextAndRemoveUnselected);
                    if (mustRespond) {
                        nextElement = element;
                        if (updateShareContextAndRemoveUnselected && ShareFetchResponse.recordsSize(respData) > 0) {
                            // Session.partitionMap is of type ImplicitLinkedHashCollection<> which tracks the order of insertion of elements.
                            // Since, we are updating an element in this case, we need to perform a remove and then a mustAdd to maintain the correct order
                            session.partitionMap().remove(cachedPart);
                            session.partitionMap().mustAdd(cachedPart);
                        }
                    } else {
                        if (updateShareContextAndRemoveUnselected) {
                            iterator.remove();
                        }
                    }
                }
            }
            return nextElement != null;
        }
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