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

Code before:
        return new StreamsTopology(
            record.epoch(),
            record.subtopologies().stream().collect(Collectors.toMap(Subtopology::subtopologyId, x -> x))
        );
    }

    /**
     * Creates an instance of StreamsTopology from a StreamsGroupHeartbeatRequestData request.
     *
     * @param topology The topology supplied in the request.
     * @return The instance of StreamsTopology created from the request.
     */
    public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestData.Topology topology) {
        StreamsGroupTopologyValue recordValue = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(topology);
        final Map<String, StreamsGroupTopologyValue.Subtopology> subtopologyMap = recordValue.subtopologies().stream()
            .collect(Collectors.toMap(StreamsGroupTopologyValue.Subtopology::subtopologyId, x -> x));
        return new StreamsTopology(topology.epoch(), subtopologyMap);
    }

    public StreamsGroupDescribeResponseData.Topology asStreamsGroupDescribeTopology() {
        return new StreamsGroupDescribeResponseData.Topology()
            .setEpoch(topologyEpoch)
            .setSubtopologies(
                subtopologies.entrySet().stream().map(
                    entry -> asStreamsGroupDescribeSubtopology(entry.getKey(), entry.getValue())
                ).toList()
            );
    }

    private StreamsGroupDescribeResponseData.Subtopology asStreamsGroupDescribeSubtopology(String subtopologyId, StreamsGroupTopologyValue.Subtopology subtopology) {
        return new StreamsGroupDescribeResponseData.Subtopology()
            .setSubtopologyId(subtopologyId)
            .setSourceTopics(subtopology.sourceTopics().stream().sorted().toList())
            .setRepartitionSinkTopics(subtopology.repartitionSinkTopics().stream().sorted().toList())
            .setRepartitionSourceTopics(subtopology.repartitionSourceTopics().stream()
                .map(this::asStreamsGroupDescribeTopicInfo)
                .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList())
            .setStateChangelogTopics(subtopology.stateChangelogTopics().stream()
                .map(this::asStreamsGroupDescribeTopicInfo)
                .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList());
    }

    private StreamsGroupDescribeResponseData.TopicInfo asStreamsGroupDescribeTopicInfo(StreamsGroupTopologyValue.TopicInfo topicInfo) {
        return new StreamsGroupDescribeResponseData.TopicInfo()
            .setName(topicInfo.name())
            .setPartitions(topicInfo.partitions())
            .setReplicationFactor(topicInfo.replicationFactor())
            .setTopicConfigs(
                topicInfo.topicConfigs().stream().map(
                    y -> new StreamsGroupDescribeResponseData.KeyValue()
                        .setKey(y.key())

Review comment (suggestion):
The subtopologies should be sorted by subtopology ID to ensure consistent ordering in the response, similar to how the test expects sorted subtopologies.
```suggestion
                subtopologies.entrySet().stream()
                    .sorted(Map.Entry.comparingByKey())
                    .map(entry -> asStreamsGroupDescribeSubtopology(entry.getKey(), entry.getValue()))
                    .toList()
```

Code after:
        return new StreamsTopology(
            record.epoch(),
            record.subtopologies().stream().collect(Collectors.toMap(Subtopology::subtopologyId, x -> x))
        );
    }

    /**
     * Creates an instance of StreamsTopology from a StreamsGroupHeartbeatRequestData request.
     *
     * @param topology The topology supplied in the request.
     * @return The instance of StreamsTopology created from the request.
     */
    public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestData.Topology topology) {
        StreamsGroupTopologyValue recordValue = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(topology);
        final Map<String, StreamsGroupTopologyValue.Subtopology> subtopologyMap = recordValue.subtopologies().stream()
            .collect(Collectors.toMap(StreamsGroupTopologyValue.Subtopology::subtopologyId, x -> x));
        return new StreamsTopology(topology.epoch(), subtopologyMap);
    }

    public StreamsGroupDescribeResponseData.Topology asStreamsGroupDescribeTopology() {
        return new StreamsGroupDescribeResponseData.Topology()
            .setEpoch(topologyEpoch)
            .setSubtopologies(
                subtopologies.entrySet().stream()
                    .sorted(Map.Entry.comparingByKey())
                    .map(entry -> asStreamsGroupDescribeSubtopology(entry.getKey(), entry.getValue()))
                    .toList()
            );
    }

    private StreamsGroupDescribeResponseData.Subtopology asStreamsGroupDescribeSubtopology(String subtopologyId, StreamsGroupTopologyValue.Subtopology subtopology) {
        return new StreamsGroupDescribeResponseData.Subtopology()
            .setSubtopologyId(subtopologyId)
            .setSourceTopics(subtopology.sourceTopics().stream().sorted().toList())
            .setRepartitionSinkTopics(subtopology.repartitionSinkTopics().stream().sorted().toList())
            .setRepartitionSourceTopics(subtopology.repartitionSourceTopics().stream()
                .map(this::asStreamsGroupDescribeTopicInfo)
                .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList())
            .setStateChangelogTopics(subtopology.stateChangelogTopics().stream()
                .map(this::asStreamsGroupDescribeTopicInfo)
                .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList());
    }

    private StreamsGroupDescribeResponseData.TopicInfo asStreamsGroupDescribeTopicInfo(StreamsGroupTopologyValue.TopicInfo topicInfo) {
        return new StreamsGroupDescribeResponseData.TopicInfo()
            .setName(topicInfo.name())
            .setPartitions(topicInfo.partitions())
            .setReplicationFactor(topicInfo.replicationFactor())
            .setTopicConfigs(
                topicInfo.topicConfigs().stream().map(
                    topicConfig -> new StreamsGroupDescribeResponseData.KeyValue()