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

Code before:
            log.info("[GroupId {}][MemberId {}] Shutdown requested for the streams application.", groupId, memberId);
            shutdownRequestMemberId = Optional.of(memberId);
        }
    }

    public Optional<String> getShutdownRequestMemberId() {
        return shutdownRequestMemberId;
    }

    private void clearShutdownRequestMemberId() {
        if (shutdownRequestMemberId.isPresent()) {
            log.info("[GroupId {}] Clearing shutdown requested for the streams application.", groupId);
            shutdownRequestMemberId = Optional.empty();
        }
    }

    public int endpointInformationEpoch() {
        return endpointInformationEpoch;
    }

    public void setEndpointInformationEpoch(int endpointInformationEpoch) {
        this.endpointInformationEpoch = endpointInformationEpoch;
    }

    // Visible for testing
    Optional<StreamsGroupHeartbeatResponseData.EndpointToPartitions> getCachedEndpointToPartitions(
        String memberId
    ) {
        return Optional.ofNullable(endpointToPartitionsCache.get(memberId));
    }

    // Visible for testing
    void cacheEndpointToPartitions(
        String memberId,
        StreamsGroupHeartbeatResponseData.EndpointToPartitions endpointToPartitions
    ) {
        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(

Review comment (performance):
```suggestion
    Optional<StreamsGroupHeartbeatResponseData.EndpointToPartitions> cachedEndpointToPartitions(
```

Code after:
            log.info("[GroupId {}][MemberId {}] Shutdown requested for the streams application.", groupId, memberId);
            shutdownRequestMemberId = Optional.of(memberId);
        }
    }

    public Optional<String> getShutdownRequestMemberId() {
        return shutdownRequestMemberId;
    }

    private void clearShutdownRequestMemberId() {
        if (shutdownRequestMemberId.isPresent()) {
            log.info("[GroupId {}] Clearing shutdown requested for the streams application.", groupId);
            shutdownRequestMemberId = Optional.empty();
        }
    }

    public int endpointInformationEpoch() {
        return endpointInformationEpoch;
    }

    public void setEndpointInformationEpoch(int endpointInformationEpoch) {
        this.endpointInformationEpoch = endpointInformationEpoch;
    }

    // Visible for testing
    Optional<StreamsGroupHeartbeatResponseData.EndpointToPartitions> cachedEndpointToPartitions(
        String memberId
    ) {
        return Optional.ofNullable(endpointToPartitionsCache.get(memberId));
    }

    // Visible for testing
    void cacheEndpointToPartitions(
        String memberId,
        StreamsGroupHeartbeatResponseData.EndpointToPartitions endpointToPartitions
    ) {
        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);
    }

    /**