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

Code before:
        updateGroupsByTopics(groupId, oldSubscribedTopicNames, consumerGroup.subscribedTopicNames().keySet());
    }

    /**
     * Replays StreamsGroupMetadataKey/Value to update the hard state of
     * the streams group. It updates the group epoch of the Streams
     * group or deletes the streams group.
     *
     * @param key   A StreamsGroupMetadataKey key.
     * @param value A StreamsGroupMetadataValue record.
     */
    public void replay(
        StreamsGroupMetadataKey key,
        StreamsGroupMetadataValue value
    ) {
        String groupId = key.groupId();

        if (value != null) {
            StreamsGroup streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, true);
            streamsGroup.setGroupEpoch(value.epoch());
            streamsGroup.setMetadataHash(value.metadataHash());
            streamsGroup.setValidatedTopologyEpoch(value.validatedTopologyEpoch());

            if (value.assignmentConfigs() != null) {
                for (StreamsGroupMetadataValue.AssignmentConfig config : value.assignmentConfigs()) {
                    streamsGroup.setAssignmentConfig(config.key(), config.value());
                }
            }

        } else {
            StreamsGroup streamsGroup;
            try {
                streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, false);
            } catch (GroupIdNotFoundException ex) {
                // If the group does not exist, we can ignore the tombstone.
                return;
            }

            if (!streamsGroup.members().isEmpty()) {
                throw new IllegalStateException("Received a tombstone record to delete group " + groupId
                    + " but the group still has " + streamsGroup.members().size() + " members.");
            }
            if (streamsGroup.assignmentEpoch() != -1) {
                throw new IllegalStateException("Received a tombstone record to delete group " + groupId
                    + " but did not receive StreamsGroupTargetAssignmentMetadataValue tombstone.");
            }
            removeGroup(groupId);
        }

    }


Review comment (suggestion):
I think this would never "unset" a configuration. For example, if we remove a configuration, we'd probably want to also remove it from the StreamsGroup object.  I would change setAssignmentConfig to setAssignmentConfigs and pass in a whole collection. Inside the method you can clear and putAll on the internal collection. 

Code after:
        updateGroupsByTopics(groupId, oldSubscribedTopicNames, consumerGroup.subscribedTopicNames().keySet());
    }

    /**
     * Replays StreamsGroupMetadataKey/Value to update the hard state of
     * the streams group. It updates the group epoch of the Streams
     * group or deletes the streams group.
     *
     * @param key   A StreamsGroupMetadataKey key.
     * @param value A StreamsGroupMetadataValue record.
     */
    public void replay(
        StreamsGroupMetadataKey key,
        StreamsGroupMetadataValue value
    ) {
        String groupId = key.groupId();

        if (value != null) {
            StreamsGroup streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, true);
            streamsGroup.setGroupEpoch(value.epoch());
            streamsGroup.setMetadataHash(value.metadataHash());
            streamsGroup.setValidatedTopologyEpoch(value.validatedTopologyEpoch());

            if (value.lastAssignmentConfigs() != null) {
                streamsGroup.setLastAssignmentConfigs(
                    value.lastAssignmentConfigs().stream()
                        .collect(Collectors.toMap(
                            StreamsGroupMetadataValue.LastAssignmentConfig::key,
                            StreamsGroupMetadataValue.LastAssignmentConfig::value
                        ))
                );
            } else {
                streamsGroup.setLastAssignmentConfigs(Map.of());
            }

        } else {
            StreamsGroup streamsGroup;
            try {
                streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, false);
            } catch (GroupIdNotFoundException ex) {
                // If the group does not exist, we can ignore the tombstone.
                return;
            }

            if (!streamsGroup.members().isEmpty()) {
                throw new IllegalStateException("Received a tombstone record to delete group " + groupId
                    + " but the group still has " + streamsGroup.members().size() + " members.");
            }
            if (streamsGroup.assignmentEpoch() != -1) {
                throw new IllegalStateException("Received a tombstone record to delete group " + groupId
                    + " but did not receive StreamsGroupTargetAssignmentMetadataValue tombstone.");