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

Code before:
    }

    /**
     * Replays ShareGroupStatePartitionMetadataKey/Value to update the hard state of
     * the share group.
     *
     * @param key   A ShareGroupStatePartitionMetadataKey key.
     * @param value A ShareGroupStatePartitionMetadataValue record.
     */
    public void replay(
        ShareGroupStatePartitionMetadataKey key,
        ShareGroupStatePartitionMetadataValue value
    ) {
        String groupId = key.groupId();

        // Update timeline structures with info about initialized/deleted topics.
        if (value == null) {
            try {
                getOrMaybeCreatePersistedShareGroup(groupId, false);
            } catch (GroupIdNotFoundException ex) {
                // Ignore tombstone if group not found.
                log.debug("ShareGroupStatePartitionMetadata tombstone for non-existent share group {}", groupId, ex);
            }
            shareGroupStatePartitionMetadata.remove(groupId);   // Should not throw any exceptions.
        } else {
            getOrMaybeCreatePersistedShareGroup(groupId, false);
            long timestamp = time.milliseconds();
            ShareGroupStatePartitionMetadataInfo info = new ShareGroupStatePartitionMetadataInfo(
                value.initializingTopics().stream()
                    .map(topicPartitionInfo -> Map.entry(
                        topicPartitionInfo.topicId(),
                        new InitMapValue(topicPartitionInfo.topicName(), new HashSet<>(topicPartitionInfo.partitions()), timestamp)))
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)),

                value.initializedTopics().stream()
                    .map(topicPartitionInfo -> Map.entry(
                        topicPartitionInfo.topicId(),
                        new InitMapValue(topicPartitionInfo.topicName(), new HashSet<>(topicPartitionInfo.partitions()), timestamp)))
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)),

                value.deletingTopics().stream()
                    .map(ShareGroupStatePartitionMetadataValue.TopicInfo::topicId)
                    .collect(Collectors.toSet())
            );

            // Init java record.
            shareGroupStatePartitionMetadata.put(groupId, info);
        }
    }

    /**

Review comment (suggestion):
I think this should be `true` for the `createIfNotExists` argument. That's the pattern that the other replay methods use and I can't think of a reason why this one would be different.

Code after:
    /**
     * Replays ShareGroupCurrentMemberAssignmentKey/Value to update the hard state of
     * the share group. It updates the assignment of a member or deletes it.
     *
     * @param key   A ShareGroupCurrentMemberAssignmentKey key.
     * @param value A ShareGroupCurrentMemberAssignmentValue record.
     */
    public void replay(
        ShareGroupCurrentMemberAssignmentKey key,
        ShareGroupCurrentMemberAssignmentValue value
    ) {
        String groupId = key.groupId();
        String memberId = key.memberId();

        ShareGroup group;
        ShareGroupMember oldMember;

        try {
            group = getOrMaybeCreatePersistedShareGroup(groupId, value != null);
            oldMember = group.getOrMaybeCreateMember(memberId, value != null);
        } catch (GroupIdNotFoundException ex) {
            log.debug("ShareGroupCurrentMemberAssignment tombstone without group - {}", groupId, ex);
            return;
        } catch (UnknownMemberIdException ex) {
            log.debug("ShareGroupCurrentMemberAssignment tombstone for groupId - {} without member - {}", groupId, memberId, ex);
            return;
        }

        if (value != null) {
            ShareGroupMember newMember = new ShareGroupMember.Builder(oldMember)
                .updateWith(value)
                .build();
            group.updateMember(newMember);
        } else {
            ShareGroupMember newMember = new ShareGroupMember.Builder(oldMember)
                .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
                .setPreviousMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH)
                .setAssignedPartitions(Map.of())
                .build();
            group.updateMember(newMember);
        }
    }

    /**
     * Replays ShareGroupStatePartitionMetadataKey/Value to update the hard state of
     * the share group.
     *
     * @param key   A ShareGroupStatePartitionMetadataKey key.
     * @param value A ShareGroupStatePartitionMetadataValue record.
     */
    public void replay(