Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java (Java)

Code before:
        boolean isFullRequest = subscribedTopicNames != null;
        if (memberEpoch == 0 || isFullRequest || hasAssignedPartitionsChanged(member, updatedMember)) {
            response.setAssignment(ShareGroupHeartbeatResponse.createAssignment(updatedMember.assignedPartitions()));
        }
        return new CoordinatorResult<>(
            records,
            Map.entry(
                response,
                maybeCreateInitializeShareGroupStateRequest(groupId, groupEpoch, subscriptionMetadata, records)
            )
        );
    }

    private boolean initializedAssignmentPending(ShareGroup group) {
        if (!shareGroupPartitionMetadata.containsKey(group.groupId())) {
            // No initialized share partitions for the group so nothing can be assigned.
            return false;
        }

        if (group.isEmpty()) {
            // No members then no point of computing assignment.
            return false;
        }

        // We need to check if all the group initialized share partitions are part of the group assignment.
        Map<Uuid, Set<Integer>> initializedTps = stripInitValue(shareGroupPartitionMetadata.get(group.groupId()).initializedTopics());
        Map<Uuid, Set<Integer>> currentAssigned = new HashMap<>();
        for (Assignment assignment : group.targetAssignment().values()) {
            for (Map.Entry<Uuid, Set<Integer>> tps : assignment.partitions().entrySet()) {
                currentAssigned.computeIfAbsent(tps.getKey(), k -> new HashSet<>())
                    .addAll(tps.getValue());
            }
        }

        return !initializedTps.equals(currentAssigned);
    }

    /**
     * Computes the diff between the subscribed metadata and the initialized share topic
     * partitions corresponding to a share group.
     *
     * @param groupId               The share group id for which diff is being calculated
     * @param subscriptionMetadata  The subscription metadata corresponding to the share group.
     * @return  A map of topic partitions which are subscribed by the share group but not initialized yet.
     */
    // Visibility for testing
    Map<Uuid, InitMapValue> subscribedTopicsChangeMap(String groupId, Map<String, TopicMetadata> subscriptionMetadata) {
        if (subscriptionMetadata == null || subscriptionMetadata.isEmpty()) {
            return Map.of();
        }


Review comment (suggestion):
nit: There is a mismatch in naming here. The record is `ShareGroupStatePartitionMetadataInfo` while the timeline hashmap is `shareGroupPartitionMetadata`. I wonder if `shareGroupStatePartitionMetadata` would be better.

Code after:
        boolean isFullRequest = subscribedTopicNames != null;
        if (memberEpoch == 0 || isFullRequest || hasAssignedPartitionsChanged(member, updatedMember)) {
            response.setAssignment(ShareGroupHeartbeatResponse.createAssignment(updatedMember.assignedPartitions()));
        }
        return new CoordinatorResult<>(
            records,
            Map.entry(
                response,
                maybeCreateInitializeShareGroupStateRequest(groupId, groupEpoch, subscriptionMetadata, records)
            )
        );
    }

    private boolean initializedAssignmentPending(ShareGroup group) {
        if (!shareGroupStatePartitionMetadata.containsKey(group.groupId())) {
            // No initialized share partitions for the group so nothing can be assigned.
            return false;
        }

        if (group.isEmpty()) {
            // No members then no point of computing assignment.
            return false;
        }

        // We need to check if all the group initialized share partitions are part of the group assignment.
        Map<Uuid, Set<Integer>> initializedTps = stripInitValue(shareGroupStatePartitionMetadata.get(group.groupId()).initializedTopics());
        Map<Uuid, Set<Integer>> currentAssigned = new HashMap<>();
        for (Assignment assignment : group.targetAssignment().values()) {
            for (Map.Entry<Uuid, Set<Integer>> tps : assignment.partitions().entrySet()) {
                currentAssigned.computeIfAbsent(tps.getKey(), k -> new HashSet<>())
                    .addAll(tps.getValue());
            }
        }

        return !initializedTps.equals(currentAssigned);
    }

    /**
     * Computes the diff between the subscribed metadata and the initialized share topic
     * partitions corresponding to a share group.
     *
     * @param groupId               The share group id for which diff is being calculated
     * @param subscriptionMetadata  The subscription metadata corresponding to the share group.
     * @return  A map of topic partitions which are subscribed by the share group but not initialized yet.
     */
    // Visibility for testing
    Map<Uuid, InitMapValue> subscribedTopicsChangeMap(String groupId, Map<String, TopicMetadata> subscriptionMetadata) {
        if (subscriptionMetadata == null || subscriptionMetadata.isEmpty()) {
            return Map.of();
        }