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

Code before:
            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(), Map.of()));

        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.

Review comment (suggestion):
Wait, why are dropping the assignment configuration on the floor here? We should use group.assignmentConfiguration here.

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(), group.lastAssignmentConfigs()));

        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.