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

Code before:
        // 2. Initialize/Update the group topology.
        // If the topology is new or has changed, a StreamsGroupTopologyValue record is written to the __consumer_offsets partition to persist
        // the change. The group epoch is bumped if the topology has changed.
        StreamsTopology updatedTopology = maybeUpdateTopology(groupId, memberId, topology, group, records);
        maybeSetTopologyStaleStatus(group, updatedMember, returnedStatus);

        // 3. Determine any internal topics if needed.
        ConfiguredTopology updatedConfiguredTopology;
        boolean reconfigureTopology = group.topology().isEmpty();
        long metadataHash = group.metadataHash();
        if (reconfigureTopology || group.configuredTopology().isEmpty() || group.hasMetadataExpired(currentTimeMs)) {

            metadataHash = group.computeMetadataHash(
                metadataImage,
                topicHashCache,
                updatedTopology
            );

            if (metadataHash != group.metadataHash()) {
                log.info("[GroupId {}][MemberId {}] Computed new metadata hash: {}.",
                    groupId, memberId, metadataHash);
                bumpGroupEpoch = true;
                reconfigureTopology = true;
            }

            // Check if assignment configurations have changed
            Map<String, String> currentAssignmentConfigs = streamsGroupAssignmentConfigs(groupId);
            Map<String, String> storedAssignmentConfigs = group.assignmentConfigs();
            if (!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
                log.info("[GroupId {}][MemberId {}] Assignment configurations changed to {}. Triggering rebalance.",
                    groupId, memberId, currentAssignmentConfigs);
                bumpGroupEpoch = true;
            }

            if (reconfigureTopology || group.configuredTopology().isEmpty()) {
                log.info("[GroupId {}][MemberId {}] Configuring the topology {}", groupId, memberId, updatedTopology);
                updatedConfiguredTopology = InternalTopicManager.configureTopics(logContext, metadataHash, updatedTopology, metadataImage);
                group.setConfiguredTopology(updatedConfiguredTopology);
            } else {
                updatedConfiguredTopology = group.configuredTopology().get();
            }
        } else {
            updatedConfiguredTopology = group.configuredTopology().get();
        }

        // 3b. If the topology is validated, persist the fact that it is validated.
        int validatedTopologyEpoch = -1;
        if (updatedConfiguredTopology.isReady()) {
            validatedTopologyEpoch = updatedTopology.topologyEpoch();
            SortedMap<String, ConfiguredSubtopology> subtopologySortedMap = updatedConfiguredTopology.subtopologies().get();
            throwIfRequestContainsInvalidTasks(subtopologySortedMap, ownedActiveTasks);

Review comment (suggestion):
I think this should be moved outside of this if. We can check it right before actually bumping the group epoch, so in a sense, it would become step 3c after step 3b. 

Code after:
        // 2. Initialize/Update the group topology.
        // If the topology is new or has changed, a StreamsGroupTopologyValue record is written to the __consumer_offsets partition to persist
        // the change. The group epoch is bumped if the topology has changed.
        StreamsTopology updatedTopology = maybeUpdateTopology(groupId, memberId, topology, group, records);
        maybeSetTopologyStaleStatus(group, updatedMember, returnedStatus);

        // 3. Determine any internal topics if needed.
        ConfiguredTopology updatedConfiguredTopology;
        boolean reconfigureTopology = group.topology().isEmpty();
        long metadataHash = group.metadataHash();
        if (reconfigureTopology || group.configuredTopology().isEmpty() || group.hasMetadataExpired(currentTimeMs)) {

            metadataHash = group.computeMetadataHash(
                metadataImage,
                topicHashCache,
                updatedTopology
            );

            if (metadataHash != group.metadataHash()) {
                log.info("[GroupId {}][MemberId {}] Computed new metadata hash: {}.",
                    groupId, memberId, metadataHash);
                bumpGroupEpoch = true;
                reconfigureTopology = true;
            }

            if (reconfigureTopology || group.configuredTopology().isEmpty()) {
                log.info("[GroupId {}][MemberId {}] Configuring the topology {}", groupId, memberId, updatedTopology);
                updatedConfiguredTopology = InternalTopicManager.configureTopics(logContext, metadataHash, updatedTopology, metadataImage);
                group.setConfiguredTopology(updatedConfiguredTopology);
            } else {
                updatedConfiguredTopology = group.configuredTopology().get();
            }
        } else {
            updatedConfiguredTopology = group.configuredTopology().get();
        }

        // 3b. If the topology is validated, persist the fact that it is validated.
        int validatedTopologyEpoch = -1;
        if (updatedConfiguredTopology.isReady()) {
            validatedTopologyEpoch = updatedTopology.topologyEpoch();
            SortedMap<String, ConfiguredSubtopology> subtopologySortedMap = updatedConfiguredTopology.subtopologies().get();
            throwIfRequestContainsInvalidTasks(subtopologySortedMap, ownedActiveTasks);
            throwIfRequestContainsInvalidTasks(subtopologySortedMap, ownedStandbyTasks);
            throwIfRequestContainsInvalidTasks(subtopologySortedMap, ownedWarmupTasks);
        }
        // We validated a topology that was not validated before, so bump the group epoch as we may have to reassign tasks.
        if (validatedTopologyEpoch != group.validatedTopologyEpoch()) {
            bumpGroupEpoch = true;
        }

        // Check if assignment configurations have changed