chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4325405019
##########
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:
@lucasbru
Thank you for the thoughtful suggestion. I agree that, considering cases
such as rejoin storms, optimizing this path with `matches()` is important for
the broker.
Would you be open to first applying your initial suggestion to copy only the
copartition groups, then landing and backporting the correctness fix before
moving on to `matches()`? (I'm happy to take on the `matches()` optimization
separately.)
My concern is that introducing a new comparison implementation covering the
current schema could inadvertently introduce new correctness issues. With that
in mind, I would prefer to keep this PR narrowly focused on handling the
ordering differences between client requests and the broker’s stored topology,
and complete the backports first(even client side!).
If you’re comfortable with that approach, could we address the `matches()`
optimization in a separate ticket, with dedicated review and testing, including
the schema field-coverage test you suggested?
--
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]
