chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4336624634
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -122,10 +128,47 @@ public static StreamsTopology
fromRecord(StreamsGroupTopologyValue record) {
* @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);
+ return
fromRecord(StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(topology));
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology
subtopology) {
+ // Copartition constraints are independent of both the index order and
the group order.
+ // For example, showing only source topic indexes:
+ // [[3, 2], [1, 0]] -> [[2, 3], [0, 1]] -> [[0, 1], [2, 3]]
+ // The first step sorts indexes within each group; the second sorts
the groups.
+ // The referenced topic lists stay unchanged, so each index still
identifies the same topic.
+ // Copy before sorting to avoid mutating a replayed record or the
heartbeat request.
+ List<CopartitionGroup> groups = new
ArrayList<>(subtopology.copartitionGroups().size());
+ for (CopartitionGroup group : subtopology.copartitionGroups()) {
+ CopartitionGroup copy = group.duplicate();
+ Collections.sort(copy.sourceTopics());
+ Collections.sort(copy.sourceTopicRegex());
+ Collections.sort(copy.repartitionSourceTopics());
+ groups.add(copy);
+ }
+
+ groups.sort(
+ Comparator
+ .comparing(CopartitionGroup::sourceTopics,
StreamsTopology::compareTopicIndexes)
+ .thenComparing(CopartitionGroup::sourceTopicRegex,
StreamsTopology::compareTopicIndexes)
+ .thenComparing(CopartitionGroup::repartitionSourceTopics,
StreamsTopology::compareTopicIndexes)
+ );
+
+ return new Subtopology()
Review Comment:
Yes, you are right!
I've addressed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
