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

Code before:
     * @param ownedActiveTasks    The list of owned active tasks from the request or null.
     * @param ownedStandbyTasks   The list of owned standby tasks from the request or null.
     * @param ownedWarmupTasks    The list of owned warmup tasks from the request or null.
     * @param userEndpoint        User-defined endpoint for Interactive Queries, or null.
     * @param clientTags          Used for rack-aware assignment algorithm, or null.
     * @param shutdownApplication Whether all Streams clients in the group should shut down.
     * @return A result containing the StreamsGroupHeartbeat response and a list of records to update the state machine.
     */
    private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
        String groupId,
        String memberId,
        int memberEpoch,
        String instanceId,
        String rackId,
        int rebalanceTimeoutMs,
        String clientId,
        String clientHost,
        StreamsGroupHeartbeatRequestData.Topology topology,
        List<TaskIds> ownedActiveTasks,
        List<TaskIds> ownedStandbyTasks,
        List<TaskIds> ownedWarmupTasks,
        String processId,
        Endpoint userEndpoint,
        List<KeyValue> clientTags,
        boolean shutdownApplication,
        int memberEndpointEpochInRequest
    ) throws ApiException {
        final long currentTimeMs = time.milliseconds();
        final List<CoordinatorRecord> records = new ArrayList<>();
        final List<StreamsGroupHeartbeatResponseData.Status> returnedStatus = new ArrayList<>();

        // Get or create the streams group.
        boolean isJoining = memberEpoch == 0;
        StreamsGroup group;
        if (isJoining) {
            group = getOrCreateStreamsGroup(groupId);
            throwIfStreamsGroupIsFull(group);
        } else {
            group = getStreamsGroupOrThrow(groupId);
        }

        // Get or create the member.
        StreamsGroupMember member;
        if (instanceId == null) {
            member = getOrMaybeCreateDynamicStreamsGroupMember(
                group,
                memberId,
                memberEpoch,
                ownedActiveTasks,
                ownedStandbyTasks,
                ownedWarmupTasks,

Review comment (suggestion):
```suggestion
        int memberEndpointEpoch
```

Code after:
     * @param ownedStandbyTasks   The list of owned standby tasks from the request or null.
     * @param ownedWarmupTasks    The list of owned warmup tasks from the request or null.
     * @param userEndpoint        User-defined endpoint for Interactive Queries, or null.
     * @param clientTags          Used for rack-aware assignment algorithm, or null.
     * @param shutdownApplication Whether all Streams clients in the group should shut down.
     * @param memberEndpointEpoch The last endpoint information epoch seen be the group member.
     * @return A result containing the StreamsGroupHeartbeat response and a list of records to update the state machine.
     */
    private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
        String groupId,
        String memberId,
        int memberEpoch,
        String instanceId,
        String rackId,
        int rebalanceTimeoutMs,
        String clientId,
        String clientHost,
        StreamsGroupHeartbeatRequestData.Topology topology,
        List<TaskIds> ownedActiveTasks,
        List<TaskIds> ownedStandbyTasks,
        List<TaskIds> ownedWarmupTasks,
        String processId,
        Endpoint userEndpoint,
        List<KeyValue> clientTags,
        boolean shutdownApplication,
        int memberEndpointEpoch
    ) throws ApiException {
        final long currentTimeMs = time.milliseconds();
        final List<CoordinatorRecord> records = new ArrayList<>();
        final List<StreamsGroupHeartbeatResponseData.Status> returnedStatus = new ArrayList<>();

        // Get or create the streams group.
        boolean isJoining = memberEpoch == 0;
        StreamsGroup group;
        if (isJoining) {
            group = getOrCreateStreamsGroup(groupId);
            throwIfStreamsGroupIsFull(group);
        } else {
            group = getStreamsGroupOrThrow(groupId);
        }

        // Get or create the member.
        StreamsGroupMember member;
        if (instanceId == null) {
            member = getOrMaybeCreateDynamicStreamsGroupMember(
                group,
                memberId,
                memberEpoch,
                ownedActiveTasks,
                ownedStandbyTasks,
                ownedWarmupTasks,