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

Code before:

        Set<String> deletedRegexes = new HashSet<>();
        counts.forEach((regex, count) -> {
            if (group.numSubscribedMembers(regex) == count) {
                records.add(newConsumerGroupRegularExpressionTombstone(group.groupId(), regex));
                deletedRegexes.add(regex);
            }
        });

        return deletedRegexes;
    }

    /**
     * Write tombstones for the member. The order matters here.
     *
     * @param records       The list of records to append the member assignment tombstone records.
     * @param groupId       The group id.
     * @param memberId      The member id.
     */
    private void removeMember(List<CoordinatorRecord> records, String groupId, String memberId) {
        records.add(newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId));
        records.add(newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId));
        records.add(newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId));
    }

    /**
     * Cancel all the timers of the member.
     *
     * @param groupId       The group id.
     * @param memberId      The member id.
     */
    private void cancelTimers(String groupId, String memberId) {
        cancelGroupSessionTimeout(groupId, memberId);
        cancelGroupRebalanceTimeout(groupId, memberId);
        cancelConsumerGroupJoinTimeout(groupId, memberId);
        cancelConsumerGroupSyncTimeout(groupId, memberId);
    }

    /**
     * Schedules (or reschedules) the session timeout for the member.
     *
     * @param groupId       The group id.
     * @param memberId      The member id.
     */
    private void scheduleStreamsGroupSessionTimeout(
        String groupId,
        String memberId
    ) {
        timer.schedule(
            groupSessionTimeoutKey(groupId, memberId),
            streamsGroupSessionTimeoutMs(groupId),

Review comment (bug):
There was a bug here - every time we'd fence a member, we'd reset the metadata hash to 0. We don't need to do that, otherwise we'll need to recompute the metadata hash on the next heartbeat and bump the group epoch again.

Code after:
            groupId,
            newMember
        ));
    }

    /**
     * Fences a member from a streams group.
     *
     * @param group     The group.
     * @param member    The member.
     * @param response  The response of the CoordinatorResult.
     *
     * @return The CoordinatorResult to be applied.
     */
    private <T> CoordinatorResult<T, CoordinatorRecord> streamsGroupFenceMember(
        StreamsGroup group,
        StreamsGroupMember member,
        T response
    ) {
        List<CoordinatorRecord> records = new ArrayList<>();

        records.addAll(removeStreamsMember(group.groupId(), member.memberId()));

        // We bump the group epoch.
        int groupEpoch = group.groupEpoch() + 1;
        records.add(newStreamsGroupMetadataRecord(group.groupId(), groupEpoch, group.metadataHash(), group.validatedTopologyEpoch()));

        cancelTimers(group.groupId(), member.memberId());

        return new CoordinatorResult<>(records, response);
    }

    /**
     * Write tombstones for the member.
     *
     * @param groupId       The group id.
     * @param memberId      The member id.
     */
    private List<CoordinatorRecord> removeStreamsMember(String groupId, String memberId) {
        return List.of(
            newStreamsGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
            newStreamsGroupTargetAssignmentTombstoneRecord(groupId, memberId),
            newStreamsGroupMemberTombstoneRecord(groupId, memberId)
        );
    }

    /**
     * Maybe delete the resolved regular expressions associated with the provided members
     * if they were the last ones subscribed to them.
     *
     * @param records   The record accumulator.