lucasbru commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4335681508
##########
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:
Since this lists the fields explicitly, a field added to `Subtopology` later
(it would have to be a tagged field) would be silently dropped here. Could we
add a small test that counts the declared instance fields of
`StreamsGroupTopologyValue.Subtopology` via reflection (8 today, including
`_unknownTaggedFields`)? A schema field count wouldn't catch tagged fields. Or
at least a comment saying this has to list every field. The same applies to
`matches()` if we go that way.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -122,10 +126,38 @@ 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.
+ Subtopology copy = subtopology.duplicate();
Review Comment:
I think `matches()` would actually replace the normalization rather than add
to it: with an order-insensitive comparison we don't need to sort or copy the
copartition groups at all, and `fromRecord`/`fromHeartbeatRequest` stay as they
were. The current version also lists the `Subtopology` fields explicitly, so
the risk of missing a field is the same either way and needs the same
reflection test. So I'd lean towards doing it in this PR. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -122,10 +126,38 @@ 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.
+ Subtopology copy = subtopology.duplicate();
+ for (CopartitionGroup group : copy.copartitionGroups()) {
+ Collections.sort(group.sourceTopics());
+ Collections.sort(group.sourceTopicRegex());
+ Collections.sort(group.repartitionSourceTopics());
+ }
+ copy.copartitionGroups().sort(
+ Comparator.comparing(CopartitionGroup::sourceTopics,
StreamsTopology::compareTopicIndexes)
+ .thenComparing(CopartitionGroup::sourceTopicRegex,
StreamsTopology::compareTopicIndexes)
+ .thenComparing(CopartitionGroup::repartitionSourceTopics,
StreamsTopology::compareTopicIndexes)
+ );
+ return copy;
+ }
+
+ private static int compareTopicIndexes(List<Short> first, List<Short>
second) {
Review Comment:
Sorry, my nit made this slightly worse: `toArray(Short[]::new)` allocates
two arrays on every comparison. The previous loop didn't allocate, so if we
keep the normalization I'd go back to 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]
