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

Code before:
                    ));

            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topology, metadataImage);
            long elapsedMs = time.milliseconds() - startTimeMs;
            if (!internalTopicsToCreate.isEmpty()) {
                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
                    "Internal topics are missing: " + summarizeTopics(internalTopicsToCreate.keySet())
                ));
                log.info("Valid topic configuration found in {}ms, but internal topics are missing for topology epoch {}: {}",
                    elapsedMs, topology.topologyEpoch(), summarizeTopics(internalTopicsToCreate.keySet()));
            } else {
                log.info("Valid topic configuration found in {}ms, topology epoch {} is now initialized.",
                    elapsedMs, topology.topologyEpoch());
            }

            return new ConfiguredTopology(
                topology.topologyEpoch(),
                metadataHash,
                Optional.of(configuredSubtopologies),
                internalTopicsToCreate,
                topicConfigurationException
            );

        } catch (TopicConfigurationException e) {
            long elapsedMs = time.milliseconds() - startTimeMs;
            log.warn("Topic configuration failed for topology epoch {} in {} ms: {}",
                topology.topologyEpoch(), elapsedMs, e.getMessage());
            return new ConfiguredTopology(
                topology.topologyEpoch(),
                metadataHash,
                Optional.empty(),
                Map.of(),
                Optional.of(e)
            );
        }
    }

    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
                                                   final CoordinatorMetadataImage metadataImage) {
        TreeSet<String> sortedMissingTopics = new TreeSet<>();
        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
            for (String sourceTopic : subtopology.sourceTopics()) {
                if (metadataImage.topicMetadata(sourceTopic).isEmpty()) {
                    sortedMissingTopics.add(sourceTopic);
                }
            }
        }
        if (!sortedMissingTopics.isEmpty()) {
            throw TopicConfigurationException.missingSourceTopics(
                "Source topics " + summarizeTopics(sortedMissingTopics) + " are missing.");
        }

Review comment (bug):
```suggestion
            log.warn("Topic configuration failed for topology epoch {} in {}ms: {}",
```

Code after:
                    ));

            Map<String, CreatableTopic> internalTopicsToCreate = missingInternalTopics(configuredSubtopologies, topology, metadataImage);
            long elapsedMs = time.milliseconds() - startTimeMs;
            if (!internalTopicsToCreate.isEmpty()) {
                topicConfigurationException = Optional.of(TopicConfigurationException.missingInternalTopics(
                    "Internal topics are missing: " + summarizeTopics(internalTopicsToCreate.keySet())
                ));
                log.info("Valid topic configuration found in {}ms, but internal topics are missing for topology epoch {}: {}",
                    elapsedMs, topology.topologyEpoch(), summarizeTopics(internalTopicsToCreate.keySet()));
            } else {
                log.info("Valid topic configuration found in {}ms, topology epoch {} is now initialized.",
                    elapsedMs, topology.topologyEpoch());
            }

            return new ConfiguredTopology(
                topology.topologyEpoch(),
                metadataHash,
                Optional.of(configuredSubtopologies),
                internalTopicsToCreate,
                topicConfigurationException
            );

        } catch (TopicConfigurationException e) {
            long elapsedMs = time.milliseconds() - startTimeMs;
            log.warn("Topic configuration failed for topology epoch {} in {}ms: {}",
                topology.topologyEpoch(), elapsedMs, e.getMessage());
            return new ConfiguredTopology(
                topology.topologyEpoch(),
                metadataHash,
                Optional.empty(),
                Map.of(),
                Optional.of(e)
            );
        }
    }

    private static void throwOnMissingSourceTopics(final StreamsTopology topology,
                                                   final CoordinatorMetadataImage metadataImage) {
        TreeSet<String> sortedMissingTopics = new TreeSet<>();
        for (StreamsGroupTopologyValue.Subtopology subtopology : topology.subtopologies().values()) {
            for (String sourceTopic : subtopology.sourceTopics()) {
                if (metadataImage.topicMetadata(sourceTopic).isEmpty()) {
                    sortedMissingTopics.add(sourceTopic);
                }
            }
        }
        if (!sortedMissingTopics.isEmpty()) {
            throw TopicConfigurationException.missingSourceTopics(
                "Source topics " + summarizeTopics(sortedMissingTopics) + " are missing.");
        }