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

Code before:
     */
    public static CoordinatorRecord newStreamsGroupMemberTombstoneRecord(
        String groupId,
        String memberId
    ) {
        Objects.requireNonNull(groupId, "groupId should not be null here");
        Objects.requireNonNull(memberId, "memberId should not be null here");

        return CoordinatorRecord.tombstone(
            new StreamsGroupMemberMetadataKey()
                .setGroupId(groupId)
                .setMemberId(memberId)
        );
    }

    public static CoordinatorRecord newStreamsGroupMetadataRecord(
        String groupId,
        int newGroupEpoch,
        long metadataHash,
        int validatedTopologyEpoch,
        Map<String, String> assignmentConfigs
    ) {
        Objects.requireNonNull(groupId, "groupId should not be null here");

        List<StreamsGroupMetadataValue.AssignmentConfig> assignmentConfigList = new ArrayList<>();
        if (assignmentConfigs != null && !assignmentConfigs.isEmpty()) {
            assignmentConfigList = assignmentConfigs.entrySet().stream()
                .map(entry -> new StreamsGroupMetadataValue.AssignmentConfig()
                    .setKey(entry.getKey())
                    .setValue(entry.getValue()))
                .sorted(Comparator.comparing(StreamsGroupMetadataValue.AssignmentConfig::key))
                .toList();
        }

        return CoordinatorRecord.record(
            new StreamsGroupMetadataKey()
                .setGroupId(groupId),
            new ApiMessageAndVersion(
                new StreamsGroupMetadataValue()
                    .setEpoch(newGroupEpoch)
                    .setMetadataHash(metadataHash)
                    .setValidatedTopologyEpoch(validatedTopologyEpoch)
                    .setAssignmentConfigs(assignmentConfigList),
                (short) 0
            )
        );
    }

    /**
     * Creates a StreamsGroupMetadata tombstone.
     *

Review comment (suggestion):
If we require non-null above, remove the check here.

Code after:
     */
    public static CoordinatorRecord newStreamsGroupMemberTombstoneRecord(
        String groupId,
        String memberId
    ) {
        Objects.requireNonNull(groupId, "groupId should not be null here");
        Objects.requireNonNull(memberId, "memberId should not be null here");

        return CoordinatorRecord.tombstone(
            new StreamsGroupMemberMetadataKey()
                .setGroupId(groupId)
                .setMemberId(memberId)
        );
    }

    public static CoordinatorRecord newStreamsGroupMetadataRecord(
        String groupId,
        int newGroupEpoch,
        long metadataHash,
        int validatedTopologyEpoch,
        Map<String, String> assignmentConfigs
    ) {
        Objects.requireNonNull(groupId, "groupId should not be null here");
        Objects.requireNonNull(assignmentConfigs, "assignmentConfigs should not be null here");

        List<StreamsGroupMetadataValue.LastAssignmentConfig> assignmentConfigList = assignmentConfigs.entrySet().stream()
            .map(entry -> new StreamsGroupMetadataValue.LastAssignmentConfig()
                .setKey(entry.getKey())
                .setValue(entry.getValue()))
            .toList();

        return CoordinatorRecord.record(
            new StreamsGroupMetadataKey()
                .setGroupId(groupId),
            new ApiMessageAndVersion(
                new StreamsGroupMetadataValue()
                    .setEpoch(newGroupEpoch)
                    .setMetadataHash(metadataHash)
                    .setValidatedTopologyEpoch(validatedTopologyEpoch)
                    .setLastAssignmentConfigs(assignmentConfigList),
                (short) 0
            )
        );
    }

    /**
     * Creates a StreamsGroupMetadata tombstone.
     *
     * @param groupId The streams group id.
     * @return The record.
     */