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

Code before:
            group::currentStandbyTaskProcessIds,
            group::currentWarmupTaskProcessIds,
            targetAssignmentEpoch,
            targetAssignment,
            ownedActiveTasks,
            ownedStandbyTasks,
            ownedWarmupTasks,
            records
        );

        scheduleStreamsGroupSessionTimeout(groupId, memberId);
        if (shutdownApplication) {
            group.setShutdownRequestMemberId(memberId);
        }

        // Prepare the response.
        StreamsGroupHeartbeatResponseData response = new StreamsGroupHeartbeatResponseData()
            .setMemberId(updatedMember.memberId())
            .setMemberEpoch(updatedMember.memberEpoch())
            .setHeartbeatIntervalMs(streamsGroupHeartbeatIntervalMs(groupId));

        if (group.endpointInformationEpoch() > memberEndpointEpochInRequest) {
            response.setPartitionsByUserEndpoint(maybeBuildEndpointToPartitions(group));
            response.setEndpointInformationEpoch(group.endpointInformationEpoch());
        } else {
            int responseEndpoint = Math.min(group.endpointInformationEpoch(), memberEndpointEpochInRequest);
            response.setEndpointInformationEpoch(responseEndpoint);
        }


        // The assignment is only provided in the following cases:
        // 1. The member is joining.
        // 2. The member's assignment has been updated.
        if (memberEpoch == 0 || hasAssignedTasksChanged(member, updatedMember)) {
            response.setActiveTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().activeTasks()));
            response.setStandbyTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().standbyTasks()));
            response.setWarmupTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().warmupTasks()));
            group.setEndpointInformationEpoch(group.endpointInformationEpoch() + 1);
        }

        Map<String, CreatableTopic> internalTopicsToBeCreated = Collections.emptyMap();
        if (updatedConfiguredTopology.topicConfigurationException().isPresent()) {
            TopicConfigurationException exception = updatedConfiguredTopology.topicConfigurationException().get();
            internalTopicsToBeCreated = updatedConfiguredTopology.internalTopicsToBeCreated();
            returnedStatus.add(
                new StreamsGroupHeartbeatResponseData.Status()
                    .setStatusCode(exception.status().code())
                    .setStatusDetail(exception.getMessage())
            );
        }


Review comment (suggestion):
We need to account for the `GroupMetadataManager` losing the state of the endpoint epoch, in that case we'll reset the clients to use the GMM endpoint epoch.

Code after:
            group::currentStandbyTaskProcessIds,
            group::currentWarmupTaskProcessIds,
            targetAssignmentEpoch,
            targetAssignment,
            ownedActiveTasks,
            ownedStandbyTasks,
            ownedWarmupTasks,
            records
        );

        scheduleStreamsGroupSessionTimeout(groupId, memberId);
        if (shutdownApplication) {
            group.setShutdownRequestMemberId(memberId);
        }

        // Prepare the response.
        StreamsGroupHeartbeatResponseData response = new StreamsGroupHeartbeatResponseData()
            .setMemberId(updatedMember.memberId())
            .setMemberEpoch(updatedMember.memberEpoch())
            .setHeartbeatIntervalMs(streamsGroupHeartbeatIntervalMs(groupId));
        // The assignment is only provided in the following cases:
        // 1. The member is joining.
        // 2. The member's assignment has been updated.
        if (memberEpoch == 0 || hasAssignedTasksChanged(member, updatedMember)) {
            response.setActiveTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().activeTasks()));
            response.setStandbyTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().standbyTasks()));
            response.setWarmupTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedTasks().warmupTasks()));
            if (memberEpoch != 0 || !updatedMember.assignedTasks().isEmpty()) {
                group.setEndpointInformationEpoch(group.endpointInformationEpoch() + 1);
            }
        }

        if (group.endpointInformationEpoch() != memberEndpointEpoch) {
            response.setPartitionsByUserEndpoint(maybeBuildEndpointToPartitions(group, updatedMember));
        }
        response.setEndpointInformationEpoch(group.endpointInformationEpoch());

        Map<String, CreatableTopic> internalTopicsToBeCreated = Collections.emptyMap();
        if (updatedConfiguredTopology.topicConfigurationException().isPresent()) {
            TopicConfigurationException exception = updatedConfiguredTopology.topicConfigurationException().get();
            internalTopicsToBeCreated = updatedConfiguredTopology.internalTopicsToBeCreated();
            returnedStatus.add(
                new StreamsGroupHeartbeatResponseData.Status()
                    .setStatusCode(exception.status().code())
                    .setStatusDetail(exception.getMessage())
            );
        }

        group.getShutdownRequestMemberId().ifPresent(requestingMemberId -> returnedStatus.add(
            new Status()
                .setStatusCode(StreamsGroupHeartbeatResponse.Status.SHUTDOWN_APPLICATION.code())