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

Code before:
     * @param groupId  The streams group id.
     * @param memberId The streams group member id.
     * @return The record.
     */
    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
            )
        );
    }


Review comment (suggestion):
I think we should also require assignmentConfigs to be non-null here and add a corresponding test. When we read a record, it may happen that the record does not have the field (yet) and is null. But when we write a record, we should always have assignmentConfigs.

Code after:
     * @param groupId  The streams group id.
     * @param memberId The streams group member id.
     * @return The record.
     */
    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.
     *