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

Code before:
        endpointToPartitionsCache.put(memberId, endpointToPartitions);
    }

    private void clearEndpointToPartitionsCache() {
        endpointToPartitionsCache.clear();
    }

    /**
     * Builds the endpoint-to-partitions list for all members, using the cache where possible.
     *
     * @param updatedMember The member that was just updated (may have a stale entry in the members map).
     * @param metadataImage The current metadata image for resolving topic partitions.
     * @return The list of endpoint-to-partitions mappings for all members with endpoints.
     */
    public List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> buildEndpointToPartitions(
        StreamsGroupMember updatedMember,
        CoordinatorMetadataImage metadataImage
    ) {
        List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> endpointToPartitionsList = new ArrayList<>();
        for (Map.Entry<String, StreamsGroupMember> entry : members.entrySet()) {
            if (updatedMember != null && entry.getKey().equals(updatedMember.memberId())) {
                continue;
            }
            getOrComputeEndpointToPartitions(entry.getValue(), metadataImage)
                .ifPresent(endpointToPartitionsList::add);
        }
        if (updatedMember != null) {
            getOrComputeEndpointToPartitions(updatedMember, metadataImage)
                .ifPresent(endpointToPartitionsList::add);
        }
        return endpointToPartitionsList;
    }

    private Optional<StreamsGroupHeartbeatResponseData.EndpointToPartitions> getOrComputeEndpointToPartitions(
        StreamsGroupMember member,
        CoordinatorMetadataImage metadataImage
    ) {
        if (member.userEndpoint().isEmpty()) {
            return Optional.empty();
        }

        String memberId = member.memberId();

        StreamsGroupHeartbeatResponseData.EndpointToPartitions cached = endpointToPartitionsCache.get(memberId);
        if (cached != null) {
            return Optional.of(cached);
        }

        Optional<StreamsGroupHeartbeatResponseData.EndpointToPartitions> computed =
            EndpointToPartitionsManager.maybeEndpointToPartitions(member, this, metadataImage);
        computed.ifPresent(endpointToPartitions ->

Review comment (bug):
Would it be simpler to start this method with
```
List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> endpointToPartitionsList = new ArrayList<>();
if (updatedMember == null) {
  return endpointToPartitionsList;
}
```

This should allow us to remove `updatedMember != null` in the body, making it easier to read the code?

But looking into the code, it seems `updateMember` should never be `null`, so why do we handle this case to begin with? If we want a guard, should we rather throw an exception (not sure if this might be too harsh crashing the GC...) or at least surface this error somehow else (log an ERROR log, with the request to file a JIRA ticket), to surface the problem, and to avoid hidden bugs?

Code after:
        endpointToPartitionsCache.put(memberId, endpointToPartitions);
    }

    /**
     * Invalidates the cached endpoint-to-partitions entry for the given member.
     * This should be called when a member's assigned tasks change during reconciliation,
     * before record replay has had a chance to call updateMember().
     *
     * @param memberId The member ID whose cache entry should be invalidated.
     */
    public void invalidateCachedEndpointToPartitions(String memberId) {
        endpointToPartitionsCache.remove(memberId);
    }

    /**
     * Builds the endpoint-to-partitions list for all members, using the cache where possible.
     *
     * @param updatedMember The member that was just updated (may have a stale entry in the members map).
     * @param metadataImage The current metadata image for resolving topic partitions.
     * @return The list of endpoint-to-partitions mappings for all members with endpoints.
     */
    public List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> buildEndpointToPartitions(
        StreamsGroupMember updatedMember,
        CoordinatorMetadataImage metadataImage
    ) {
        List<StreamsGroupHeartbeatResponseData.EndpointToPartitions> endpointToPartitionsList = new ArrayList<>();
        if (updatedMember == null) {
            log.error("[GroupId {}] updatedMember is unexpectedly null in buildEndpointToPartitions. " +
                "This is a bug, please file a JIRA ticket.", groupId);
            return endpointToPartitionsList;
        }
        for (Map.Entry<String, StreamsGroupMember> entry : members.entrySet()) {
            if (entry.getKey().equals(updatedMember.memberId())) {
                continue;
            }
            getOrComputeEndpointToPartitions(entry.getValue(), metadataImage)
                .ifPresent(endpointToPartitionsList::add);
        }
        getOrComputeEndpointToPartitions(updatedMember, metadataImage)
            .ifPresent(endpointToPartitionsList::add);
        return endpointToPartitionsList;
    }

    private Optional<StreamsGroupHeartbeatResponseData.EndpointToPartitions> getOrComputeEndpointToPartitions(
        StreamsGroupMember member,
        CoordinatorMetadataImage metadataImage
    ) {
        if (member.userEndpoint().isEmpty()) {
            return Optional.empty();
        }