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

Code before:
                              Map<String, Subtopology> subtopologies,
                              Map<String, Subtopology> sourceTopicMap) {

    /**
     * Constructor that automatically computes the sourceTopicMap from subtopologies.
     *
     * @param topologyEpoch The epoch of the topology (must be non-negative).
     * @param subtopologies The subtopologies of the topology.
     */
    public StreamsTopology(int topologyEpoch, Map<String, Subtopology> subtopologies) {
        this(topologyEpoch, subtopologies, computeSourceTopicMap(subtopologies));
    }

    public StreamsTopology {
        if (topologyEpoch < 0) {
            throw new IllegalArgumentException("Topology epoch must be non-negative.");
        }
        subtopologies = Collections.unmodifiableMap(Objects.requireNonNull(subtopologies, "Subtopologies cannot be null."));
        sourceTopicMap = Collections.unmodifiableMap(Objects.requireNonNull(sourceTopicMap, "Source topic map cannot be null."));
    }

    private static Map<String, Subtopology> computeSourceTopicMap(Map<String, Subtopology> subtopologies) {
        Objects.requireNonNull(subtopologies, "Subtopologies cannot be null.");
        Map<String, Subtopology> computedMap = new HashMap<>();
        for (Map.Entry<String, Subtopology> entry : subtopologies.entrySet()) {
            Subtopology subtopology = entry.getValue();
            // Add regular source topics
            for (String sourceTopic : subtopology.sourceTopics()) {
                computedMap.put(sourceTopic, subtopology);
            }
            // Add repartition source topics
            for (TopicInfo repartitionSourceTopic : subtopology.repartitionSourceTopics()) {
                computedMap.put(repartitionSourceTopic.name(), subtopology);
            }
        }
        return computedMap;
    }

    /**
     * Returns the set of topics that are required by the topology.
     * <p>
     * The required topics are used to determine the partition metadata on the brokers needed to configure the topology.
     *
     * @return set of topics required by the topology
     */
    public Set<String> requiredTopics() {
        return subtopologies.values().stream()
            .flatMap(x ->
                Stream.concat(
                    Stream.concat(
                        x.sourceTopics().stream(),

Review comment (refactor):
Seems we only access the values? Should we simplify the code to:
```
for (Subtopology subtopology : subtopologies.values()) {
...
}
```

Code after:
                              Map<String, Subtopology> subtopologies,
                              Map<String, Subtopology> sourceTopicMap) {

    /**
     * Constructor that automatically computes the sourceTopicMap from subtopologies.
     *
     * @param topologyEpoch The epoch of the topology (must be non-negative).
     * @param subtopologies The subtopologies of the topology.
     */
    public StreamsTopology(int topologyEpoch, Map<String, Subtopology> subtopologies) {
        this(topologyEpoch, subtopologies, computeSourceTopicMap(subtopologies));
    }

    public StreamsTopology {
        if (topologyEpoch < 0) {
            throw new IllegalArgumentException("Topology epoch must be non-negative.");
        }
        subtopologies = Collections.unmodifiableMap(Objects.requireNonNull(subtopologies, "Subtopologies cannot be null."));
        sourceTopicMap = Collections.unmodifiableMap(Objects.requireNonNull(sourceTopicMap, "Source topic map cannot be null."));
    }

    private static Map<String, Subtopology> computeSourceTopicMap(Map<String, Subtopology> subtopologies) {
        Objects.requireNonNull(subtopologies, "Subtopologies cannot be null.");
        Map<String, Subtopology> computedMap = new HashMap<>();
        for (Subtopology subtopology : subtopologies.values()) {
            // Add regular source topics
            for (String sourceTopic : subtopology.sourceTopics()) {
                computedMap.put(sourceTopic, subtopology);
            }
            // Add repartition source topics
            for (TopicInfo repartitionSourceTopic : subtopology.repartitionSourceTopics()) {
                computedMap.put(repartitionSourceTopic.name(), subtopology);
            }
        }
        return computedMap;
    }

    /**
     * Returns the set of topics that are required by the topology.
     * <p>
     * The required topics are used to determine the partition metadata on the brokers needed to configure the topology.
     *
     * @return set of topics required by the topology
     */
    public Set<String> requiredTopics() {
        return subtopologies.values().stream()
            .flatMap(x ->
                Stream.concat(
                    Stream.concat(
                        x.sourceTopics().stream(),
                        x.repartitionSourceTopics().stream().map(TopicInfo::name)