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

Code before:
                return group.topology().get();
            }
        } else if (group.topology().isPresent()) {
            return group.topology().get();
        } else {
            throw new IllegalStateException("The topology is null and the group topology is also null.");
        }
    }

    private List<StreamsGroupHeartbeatResponseData.TaskIds> createStreamsGroupHeartbeatResponseTaskIds(final Map<String, Set<Integer>> taskIds) {
        return taskIds.entrySet().stream()
            .map(entry -> new StreamsGroupHeartbeatResponseData.TaskIds()
                .setSubtopologyId(entry.getKey())
                .setPartitions(entry.getValue().stream().sorted().toList()))
            .collect(Collectors.toList());
    }

    private List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> maybeBuildEndpointToPartitions(StreamsGroup group,
                                                                                                        StreamsGroupMember updatedMember) {
        List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> endpointToPartitionsList = new ArrayList<>();
        final Map<String, StreamsGroupMember> members = group.members();
        for (Map.Entry<String, StreamsGroupMember> entry : members.entrySet()) {
            final String memberIdForAssignment = entry.getKey();
            final Optional<StreamsGroupMemberMetadataValue.Endpoint> endpointOptional = members.get(memberIdForAssignment).userEndpoint();
            StreamsGroupMember groupMember = updatedMember != null && memberIdForAssignment.equals(updatedMember.memberId()) ? updatedMember : members.get(memberIdForAssignment);
            if (endpointOptional.isPresent()) {
                final StreamsGroupMemberMetadataValue.Endpoint endpoint = endpointOptional.get();
                final StreamsGroupHeartbeatResponseData.Endpoint responseEndpoint = new StreamsGroupHeartbeatResponseData.Endpoint();
                responseEndpoint.setHost(endpoint.host());
                responseEndpoint.setPort(endpoint.port());
                StreamsGroupHeartbeatResponseData.EndpointToPartitions endpointToPartitions = EndpointToPartitionsManager.endpointToPartitions(groupMember, responseEndpoint, group);
                endpointToPartitionsList.add(endpointToPartitions);
            }
        }
        return endpointToPartitionsList.isEmpty() ? null : endpointToPartitionsList;
    }

    /**
     * Handles a regular heartbeat from a consumer group member. It mainly consists of
     * three parts:
     * 1) The member is created or updated. The group epoch is bumped if the member
     *    has been created or updated.
     * 2) The target assignment for the consumer group is updated if the group epoch
     *    is larger than the current target assignment epoch.
     * 3) The member's assignment is reconciled with the target assignment.
     *
     * @param context               The request context.
     * @param groupId               The group id from the request.
     * @param memberId              The member id from the request.
     * @param memberEpoch           The member epoch from the request.
     * @param instanceId            The instance id from the request or null.

Review comment (suggestion):
This is necessary as when the streams member gets updated, it returns a new `StreamsMember` instance with the updated assignment.

Code after:
                log.debug("[GroupId {}][MemberId {}] Member joined with currently initialized topology {}", groupId, memberId, topology.epoch());
                return group.topology().get();
            }
        } else if (group.topology().isPresent()) {
            return group.topology().get();
        } else {
            throw new IllegalStateException("The topology is null and the group topology is also null.");
        }
    }

    private List<StreamsGroupHeartbeatResponseData.TaskIds> createStreamsGroupHeartbeatResponseTaskIds(final Map<String, Set<Integer>> taskIds) {
        return taskIds.entrySet().stream()
            .map(entry -> new StreamsGroupHeartbeatResponseData.TaskIds()
                .setSubtopologyId(entry.getKey())
                .setPartitions(entry.getValue().stream().sorted().toList()))
            .collect(Collectors.toList());
    }

    private List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> maybeBuildEndpointToPartitions(StreamsGroup group,
                                                                                                        StreamsGroupMember updatedMember) {
        List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> endpointToPartitionsList = new ArrayList<>();
        final Map<String, StreamsGroupMember> members = group.members();
        for (Map.Entry<String, StreamsGroupMember> entry : members.entrySet()) {
            final String memberIdForAssignment = entry.getKey();
            final Optional<StreamsGroupMemberMetadataValue.Endpoint> endpointOptional = members.get(memberIdForAssignment).userEndpoint();
            StreamsGroupMember groupMember = updatedMember != null && memberIdForAssignment.equals(updatedMember.memberId()) ? updatedMember : members.get(memberIdForAssignment);
            if (endpointOptional.isPresent()) {
                final StreamsGroupMemberMetadataValue.Endpoint endpoint = endpointOptional.get();
                final StreamsGroupHeartbeatResponseData.Endpoint responseEndpoint = new StreamsGroupHeartbeatResponseData.Endpoint();
                responseEndpoint.setHost(endpoint.host());
                responseEndpoint.setPort(endpoint.port());
                StreamsGroupHeartbeatResponseData.EndpointToPartitions endpointToPartitions = EndpointToPartitionsManager.endpointToPartitions(groupMember, responseEndpoint, group);
                endpointToPartitionsList.add(endpointToPartitions);
            }
        }
        return endpointToPartitionsList.isEmpty() ? null : endpointToPartitionsList;
    }

    /**
     * Handles a regular heartbeat from a consumer group member. It mainly consists of
     * three parts:
     * 1) The member is created or updated. The group epoch is bumped if the member
     *    has been created or updated.
     * 2) The target assignment for the consumer group is updated if the group epoch
     *    is larger than the current target assignment epoch.
     * 3) The member's assignment is reconciled with the target assignment.
     *
     * @param context               The request context.
     * @param groupId               The group id from the request.
     * @param memberId              The member id from the request.
     * @param memberEpoch           The member epoch from the request.