lucasbru commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4325079373
##########
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:
Thinking about this more: since the join heartbeat can be hot on large
clusters (e.g. during rejoin storms), could we avoid building a
`StreamsTopology` from the request at all when the group already has one?
`maybeUpdateTopology` already converts the full request on every
topology-carrying heartbeat, and with the normalization we copy it a second
time.
One option: only call `fromHeartbeatRequest` when initializing the topology,
and otherwise compare the request directly against the stored topology with
something like
`StreamsTopology#matches(StreamsGroupHeartbeatRequestData.Topology)`, matching
copartition groups regardless of order. The stored topology is already
normalized in `fromRecord`, and the groups are tiny, so this can be done
without allocating. It would need a test that fails when a field is added to
`Subtopology`, so the comparison doesn't silently miss it. I'd rather not
override `equals` for this, since `hashCode` would also need to be made
order-independent.
What do you think?
--
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]
