public void replay(
        ShareGroupMemberMetadataKey key,
        ShareGroupMemberMetadataValue value
    ) {
        String groupId = key.groupId();
        String memberId = key.memberId();

        ShareGroup shareGroup = getOrMaybeCreatePersistedShareGroup(groupId, value != null);
        Set<String> oldSubscribedTopicNames = new HashSet<>(shareGroup.subscribedTopicNames().keySet());

        if (value != null) {
            ShareGroupMember oldMember = shareGroup.getOrMaybeCreateMember(memberId, true);
            shareGroup.updateMember(new ShareGroupMember.Builder(oldMember)
                .updateWith(value)
                .build());
        } else {
            ShareGroupMember oldMember = shareGroup.getOrMaybeCreateMember(memberId, false);
            if (oldMember.memberEpoch() != LEAVE_GROUP_MEMBER_EPOCH) {
                throw new IllegalStateException("Received a tombstone record to delete member " + memberId
                    + " with invalid leave group epoch.");
            }
            if (shareGroup.targetAssignment().containsKey(memberId)) {
                throw new IllegalStateException("Received a tombstone record to delete member " + memberId
                    + " but member exists in target assignment.");
            }
            shareGroup.removeMember(memberId);
        }

        updateGroupsByTopics(groupId, oldSubscribedTopicNames, shareGroup.subscribedTopicNames().keySet());
    }
Replays ShareGroupMemberMetadataKey/Value to update the hard state of
the share group. It updates the subscription part of the member or
delete the member.

@param key   A ShareGroupMemberMetadataKey key.
@param value A ShareGroupMemberMetadataValue record.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java