You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chickenchickenlove (via GitHub)" <gi...@apache.org> on 2026/10/18 03:21:16 UTC
[PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
chickenchickenlove opened a new pull request, #23855:
URL: https://github.com/apache/kafka/pull/23855
Fixes KAFKA-21302
With `group.protocol=streams`, a member with identical copartition constraints can be rejected because of differences in source topic ordering. For example, constructing the same join with source topics `["A", "B"]` and `["B", "A"]` can produce heartbeat copartition indexes of `[0, 1, 2]` and `[1, 0, 2]`, respectively.
The coordinator compares these lists in an order-sensitive manner and treats the second member’s request as a topology change. It returns `INVALID_REQUEST`, causing the client to transition to FATAL and preventing an otherwise valid instance from joining or replacing an existing member
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6288066656
Thanks for the fix! I think we should also fix this on the client. `StreamsGroupHeartbeatRequestManager#getCopartitionGroupsFromStreams` builds the index lists and the list of groups by iterating a `Set`, which is where the nondeterministic order comes from. Could you sort the indexes within each group and then the groups there too, the same way we already sort source/sink/repartition topics in `fromStreamsToHeartbeatRequest`? Clients would then always send a canonical topology. We'd still keep the broker-side normalization permanently, both on join and on replay, because of older clients and already-persisted unsorted records. A test in `StreamsGroupHeartbeatRequestManagerTest` that checks the sorted output would be good too.
It might also be worth sorting in `convertToStreamsGroupTopologyRecord`, so that newly written topology records are canonical too. The record format doesn't change, and it means a broker downgrade won't reject newer, sorted clients against a stored record. `fromRecord` would still normalize for groups created before this change.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4412318571
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested : topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject duplicates so a
+ // malformed request cannot match one stored subtopology twice and omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its lists
+ // may be unmodifiable or shared with 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) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol version
Review Comment:
I'd drop this TODO, or move it to a JIRA.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java:
##########
@@ -122,6 +132,233 @@ public void fromHeartbeatRequestShouldCreateCorrectTopology() {
assertEquals(mkSubtopology2(), topology.subtopologies().get(SUBTOPOLOGY_ID_2));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void fromRecordShouldNormalizeCopartitionGroupsWithoutMutation(String topicType) {
+ StreamsGroupTopologyValue alreadySortedExpected = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(
+ topologyWithCopartitionGroups(topicType)
+ );
+
+ StreamsGroupTopologyValue record = alreadySortedExpected.duplicate();
+ List<StreamsGroupTopologyValue.CopartitionGroup> groups = record.subtopologies().get(0).copartitionGroups();
+ Collections.reverse(groups);
+ for (StreamsGroupTopologyValue.CopartitionGroup group : groups) {
+ Collections.reverse(group.sourceTopics());
+ Collections.reverse(group.sourceTopicRegex());
+ Collections.reverse(group.repartitionSourceTopics());
+ }
+ StreamsGroupTopologyValue originalRecord = record.duplicate();
+
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+
+ Subtopology expectedSubtopology = alreadySortedExpected.subtopologies().get(0);
+ assertEquals(
+ expectedSubtopology,
+ replayed.subtopologies().get(expectedSubtopology.subtopologyId())
+ );
+ assertEquals(originalRecord, record);
+ }
+
+
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void reorderedRequestShouldMatchWithoutMutation(String topicType) {
+ StreamsGroupHeartbeatRequestData.Topology request =
+ topologyWithCopartitionGroups(topicType);
+ StreamsTopology stored = StreamsTopology.fromRecord(
+ StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request)
+ );
+
+ // Stored groups: [[0], [0, 1], [2, 3]]
+ // Request groups: [[3, 2], [1, 0], [0]]
+ StreamsGroupHeartbeatRequestData.Topology reorderedRequest = request.duplicate();
+ reorderedRequest.subtopologies().get(0).setCopartitionGroups(List.of(
+ copartitionGroup(topicType, List.of((short) 3, (short) 2)),
+ copartitionGroup(topicType, List.of((short) 1, (short) 0)),
+ copartitionGroup(topicType, List.of((short) 0))
+ ));
+ StreamsGroupHeartbeatRequestData.Topology originalRequest =
+ reorderedRequest.duplicate();
+
+ assertTrue(stored.matches(reorderedRequest));
+ assertEquals(originalRequest, reorderedRequest);
+ }
+
+
+ @Test
+ public void mixedCopartitionGroupsShouldMatchRegardlessOfOrder() {
+ // request has 'source - [1] / repartition - [3, 2]' and 'source - [1], repartition - [0]'
+ // reordered has 'source - [1], repartition - [0' and 'source - [1], repartition = [2, 3]'
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups("source");
+ request.subtopologies().get(0).setCopartitionGroups(List.of(
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 3, (short) 2)),
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 0))
+ ));
+
+ StreamsGroupTopologyValue record = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request);
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+
+ StreamsGroupHeartbeatRequestData.Topology reorderedRequest = request.duplicate();
+ Collections.reverse(reorderedRequest.subtopologies().get(0).copartitionGroups());
+ reorderedRequest.subtopologies().get(0).copartitionGroups().get(1)
+ .setRepartitionSourceTopics(List.of((short) 2, (short) 3));
+
+ assertTrue(replayed.matches(reorderedRequest));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void differentCopartitionConstraintsShouldNotMatch(String topicType) {
+ // request - [[0], [0,1], [2,3]]
+ // changed - [[0], [0,1], [1,3]]
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups(topicType);
+ StreamsTopology original = StreamsTopology.fromHeartbeatRequest(request);
+
+ StreamsGroupHeartbeatRequestData.Topology changed = request.duplicate();
+ changed.subtopologies()
+ .get(0)
+ .copartitionGroups()
+ .set(2, copartitionGroup(topicType, List.of((short) 1, (short) 3)));
+
+ assertFalse(original.matches(changed));
+ }
+
+ private static Stream<Arguments> nonCopartitionFieldChanges() {
Review Comment:
nit: this could be a plain `Stream.of(Arguments.of("source topic", (Consumer<...>) t -> ...), ...)` without the intermediate map.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested : topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject duplicates so a
+ // malformed request cannot match one stored subtopology twice and omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its lists
+ // may be unmodifiable or shared with 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) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol version
+ // requires canonical copartition ordering and older versions are no longer
+ // supported. Validate the required ordering before comparing.
+
+ // Copy before sorting to preserve the request and its index lists.
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> sortedRequested =
+ new ArrayList<>(requested.size());
+
+ for (StreamsGroupHeartbeatRequestData.CopartitionGroup group : requested) {
+ StreamsGroupHeartbeatRequestData.CopartitionGroup copy = group.duplicate();
+ Collections.sort(copy.sourceTopics());
+ Collections.sort(copy.sourceTopicRegex());
+ Collections.sort(copy.repartitionSourceTopics());
+ sortedRequested.add(copy);
+ }
+
+ sortedRequested.sort(
Review Comment:
The sorting is now duplicated for the two `CopartitionGroup` types. What about converting the request's groups to `StreamsGroupTopologyValue.CopartitionGroup` copies and reusing the sort from `normalizeCopartitionGroups`? Then we can compare the lists with `equals` and drop `matchesCopartitionGroup`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested : topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject duplicates so a
+ // malformed request cannot match one stored subtopology twice and omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its lists
+ // may be unmodifiable or shared with 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) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol version
+ // requires canonical copartition ordering and older versions are no longer
+ // supported. Validate the required ordering before comparing.
+
+ // Copy before sorting to preserve the request and its index lists.
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> sortedRequested =
+ new ArrayList<>(requested.size());
+
+ for (StreamsGroupHeartbeatRequestData.CopartitionGroup group : requested) {
+ StreamsGroupHeartbeatRequestData.CopartitionGroup copy = group.duplicate();
+ Collections.sort(copy.sourceTopics());
+ Collections.sort(copy.sourceTopicRegex());
+ Collections.sort(copy.repartitionSourceTopics());
+ sortedRequested.add(copy);
+ }
+
+ sortedRequested.sort(
+ Comparator.comparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopicRegex,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::repartitionSourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ );
+
+ // Stored groups were already sorted by fromRecord.
Review Comment:
This assumes the stored groups are sorted, but `fromHeartbeatRequest` builds a topology without going through `fromRecord`. Could it just delegate to `fromRecord(convertToStreamsGroupTopologyRecord(topology))`? Then the test setups using it get the same normalization, and removing it later becomes optional.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6494035994
@lucasbru
Sorry for the delay and thanks for taking time to look into this again!
I've addressed it based on your comments.
When you get a chance, please take another look!
Also, I've noticed that`₩StreamsTopology.fromHeartbeatRequest()` is only used in tests.
If this new commit looks good to you, I'll create a follow-up commit to remove its usage from the tests.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6313690282
Thanks for the update and for checking the `Set.of` case. I was wrong earlier: the order isn't deterministic for the same code, so this can happen within a single app during a rolling restart. I left a few more comments on the new commit.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4420446076
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested : topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject duplicates so a
+ // malformed request cannot match one stored subtopology twice and omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its lists
+ // may be unmodifiable or shared with 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) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol version
Review Comment:
I removed it and created KAFKA-21400 instead!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4411271660
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested : topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject duplicates so a
+ // malformed request cannot match one stored subtopology twice and omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its lists
+ // may be unmodifiable or shared with 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) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol version
+ // requires canonical copartition ordering and older versions are no longer
+ // supported. Validate the required ordering before comparing.
+
+ // Copy before sorting to preserve the request and its index lists.
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> sortedRequested =
+ new ArrayList<>(requested.size());
+
+ for (StreamsGroupHeartbeatRequestData.CopartitionGroup group : requested) {
+ StreamsGroupHeartbeatRequestData.CopartitionGroup copy = group.duplicate();
+ Collections.sort(copy.sourceTopics());
+ Collections.sort(copy.sourceTopicRegex());
+ Collections.sort(copy.repartitionSourceTopics());
+ sortedRequested.add(copy);
+ }
+
+ sortedRequested.sort(
+ Comparator.comparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopicRegex,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::repartitionSourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ );
+
+ // Stored groups were already sorted by fromRecord.
+ for (int i = 0; i < current.size(); i++) {
+ if (!matchesCopartitionGroup(current.get(i), sortedRequested.get(i))) {
Review Comment:
`fromHeartbeatRequest()` is currently only used in tests. Both production initialization and replay go through `fromRecord()`, which normalizes copartition groups before `matches()` is called. The self-match inconsistency exists, but it does not affect the current production path.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6291179412
Good point, I hadn't thought about the unpatched-broker case. Agreed, let's keep this PR broker-only so it can be backported on its own, and handle the client side in a separate ticket.
The same argument applies to my suggestion of writing sorted records: if a patched broker writes sorted records and is later downgraded, existing clients sending the old order would be rejected. So let's drop that and keep persisting what the client sent.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4371456652
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +131,178 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch() || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+ for (int i = 0; i < topology.subtopologies().size(); i++) {
+ StreamsGroupHeartbeatRequestData.Subtopology requested = topology.subtopologies().get(i);
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+
+ // Do not let repeated IDs match one stored subtopology while omitting another.
+ for (int j = 0; j < i; j++) {
+ if (requested.subtopologyId().equals(topology.subtopologies().get(j).subtopologyId())) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ if (!current.subtopologyId().equals(requested.subtopologyId())) {
+ return false;
+ }
+
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
Review Comment:
Thanks for doing this, and sorry, my "without allocating" comment pushed in this direction. When the order differs, `sameTopicIndexes` is O(k²) and `matchesCopartitionGroups` O(G²), and nothing bounds those sizes in the request. A client with access to the group can initialize it with a large index list (~100KB already gives ~10⁹ iterations) and then rejoin with a permutation, which keeps the coordinator thread busy on every such heartbeat.
I think sorting is both simpler and safe here: in `fromRecord`, `duplicate()` the subtopology and sort its copartition groups (indexes within each group, then the groups, with the old allocation-free `compareTopicIndexes` loop). This only runs on initialization and load, and it needs to be a copy because the handler passes the same `recordValue` it is about to write, and the lists are unmodifiable or shared with the request. `fromHeartbeatRequest` should then delegate to `fromRecord` again. In `matches`, sort copies of just the request's copartition groups and compare them with the stored ones. That's O(k log k), records stay as the client sent them, and the counting code goes away.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +131,178 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch() || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+ for (int i = 0; i < topology.subtopologies().size(); i++) {
+ StreamsGroupHeartbeatRequestData.Subtopology requested = topology.subtopologies().get(i);
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+
+ // Do not let repeated IDs match one stored subtopology while omitting another.
Review Comment:
This is O(S²) in the number of subtopologies, with S only bounded by the record size (~1e5 minimal subtopologies). Could we track the seen IDs in a `HashSet` instead?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +131,178 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch() || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+ for (int i = 0; i < topology.subtopologies().size(); i++) {
+ StreamsGroupHeartbeatRequestData.Subtopology requested = topology.subtopologies().get(i);
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+
+ // Do not let repeated IDs match one stored subtopology while omitting another.
+ for (int j = 0; j < i; j++) {
+ if (requested.subtopologyId().equals(topology.subtopologies().get(j).subtopologyId())) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ if (!current.subtopologyId().equals(requested.subtopologyId())) {
Review Comment:
nit: `current` was looked up by `requested.subtopologyId()` in a map keyed by subtopology ID, so this check is always true. I'd drop it.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java:
##########
@@ -122,6 +128,142 @@ public void fromHeartbeatRequestShouldCreateCorrectTopology() {
assertEquals(mkSubtopology2(), topology.subtopologies().get(SUBTOPOLOGY_ID_2));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void copartitionOrderShouldNotAffectMatching(String topicType) {
+ // request has - [[0], [0, 1], [2, 3]]
+ // replayed has - [[3, 2], [1, 0], [0]]
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups(topicType);
+ StreamsGroupHeartbeatRequestData.Topology originalRequest = request.duplicate();
+
+ StreamsGroupTopologyValue record = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request).duplicate();
+ List<StreamsGroupTopologyValue.CopartitionGroup> groups = record.subtopologies().get(0).copartitionGroups();
+ Collections.reverse(groups);
+ for (StreamsGroupTopologyValue.CopartitionGroup group : groups) {
+ Collections.reverse(group.sourceTopics());
+ Collections.reverse(group.sourceTopicRegex());
+ Collections.reverse(group.repartitionSourceTopics());
+ }
+ StreamsGroupTopologyValue originalRecord = record.duplicate();
+
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+ StreamsTopology received = StreamsTopology.fromHeartbeatRequest(request);
+
+ assertTrue(replayed.matches(request));
+ assertTrue(received.matches(request));
+ assertNotEquals(received, replayed);
+ assertEquals(originalRecord, record);
+ assertEquals(originalRequest, request);
+ }
+
+ @Test
+ public void mixedCopartitionGroupsShouldMatchRegardlessOfOrder() {
+ // request has 'source - [1] / repartition - [3, 2]' and 'source - [1], repartition - [0]'
+ // reordered has 'source - [1], repartition - [0' and 'source - [1], repartition = [2, 3]'
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups("source");
+ request.subtopologies().get(0).setCopartitionGroups(List.of(
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 3, (short) 2)),
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 0))
+ ));
+
+ StreamsGroupTopologyValue record = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request);
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+
+ StreamsGroupHeartbeatRequestData.Topology reorderedRequest = request.duplicate();
+ Collections.reverse(reorderedRequest.subtopologies().get(0).copartitionGroups());
+ reorderedRequest.subtopologies().get(0).copartitionGroups().get(1)
+ .setRepartitionSourceTopics(List.of((short) 2, (short) 3));
+
+ assertTrue(replayed.matches(reorderedRequest));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void differentCopartitionConstraintsShouldNotMatch(String topicType) {
Review Comment:
Since `matches` now compares every field by hand, could we add one parameterized test that changes a single non-copartition field at a time and asserts `matches` is false: source topics, sink topics, a changelog/repartition `TopicInfo` field (partitions, replication factor, configs), a missing subtopology, a duplicate subtopology ID? Otherwise a wrong comparison of an existing field would go unnoticed, and the field-count test only covers new fields.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4411100599
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +131,178 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch() || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+ for (int i = 0; i < topology.subtopologies().size(); i++) {
+ StreamsGroupHeartbeatRequestData.Subtopology requested = topology.subtopologies().get(i);
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+
+ // Do not let repeated IDs match one stored subtopology while omitting another.
Review Comment:
Thanks for your suggestion. it is much better.
I've addressed it.
One more things I want to say is that `matches()` will return `false` if the request contains duplicate subtopology IDs. This prevents a request from matching the same stored subtopology multiple times while omitting another.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4411104794
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java:
##########
@@ -122,6 +128,142 @@ public void fromHeartbeatRequestShouldCreateCorrectTopology() {
assertEquals(mkSubtopology2(), topology.subtopologies().get(SUBTOPOLOGY_ID_2));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void copartitionOrderShouldNotAffectMatching(String topicType) {
+ // request has - [[0], [0, 1], [2, 3]]
+ // replayed has - [[3, 2], [1, 0], [0]]
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups(topicType);
+ StreamsGroupHeartbeatRequestData.Topology originalRequest = request.duplicate();
+
+ StreamsGroupTopologyValue record = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request).duplicate();
+ List<StreamsGroupTopologyValue.CopartitionGroup> groups = record.subtopologies().get(0).copartitionGroups();
+ Collections.reverse(groups);
+ for (StreamsGroupTopologyValue.CopartitionGroup group : groups) {
+ Collections.reverse(group.sourceTopics());
+ Collections.reverse(group.sourceTopicRegex());
+ Collections.reverse(group.repartitionSourceTopics());
+ }
+ StreamsGroupTopologyValue originalRecord = record.duplicate();
+
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+ StreamsTopology received = StreamsTopology.fromHeartbeatRequest(request);
+
+ assertTrue(replayed.matches(request));
+ assertTrue(received.matches(request));
+ assertNotEquals(received, replayed);
+ assertEquals(originalRecord, record);
+ assertEquals(originalRequest, request);
+ }
+
+ @Test
+ public void mixedCopartitionGroupsShouldMatchRegardlessOfOrder() {
+ // request has 'source - [1] / repartition - [3, 2]' and 'source - [1], repartition - [0]'
+ // reordered has 'source - [1], repartition - [0' and 'source - [1], repartition = [2, 3]'
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups("source");
+ request.subtopologies().get(0).setCopartitionGroups(List.of(
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 3, (short) 2)),
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setSourceTopics(List.of((short) 1))
+ .setRepartitionSourceTopics(List.of((short) 0))
+ ));
+
+ StreamsGroupTopologyValue record = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request);
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+
+ StreamsGroupHeartbeatRequestData.Topology reorderedRequest = request.duplicate();
+ Collections.reverse(reorderedRequest.subtopologies().get(0).copartitionGroups());
+ reorderedRequest.subtopologies().get(0).copartitionGroups().get(1)
+ .setRepartitionSourceTopics(List.of((short) 2, (short) 3));
+
+ assertTrue(replayed.matches(reorderedRequest));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void differentCopartitionConstraintsShouldNotMatch(String topicType) {
Review Comment:
I've added another parameterized test!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6292547482
@lucasbru
I've pushed a new commit that incorporates most of your suggestions!
I left out the `matches()` optimization for now, as I think it would be better to address it in a separate issue. However, if you consider it a minor change, I'm also happy to include it in this PR. Just let me know what you think!
I also added a comment at the end of KAFKA-21302. Even when running the exact same code, the ordering of copartition groups can vary across JVM runs if the source nodes are defined using `Set.of(...)`.
I wanted to highlight this again, as I thought it might be helpful when determining the priority of this issue.
When you get a chance, please take another look! 🙇♂️
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6315658666
@lucasbru
Thanks for taking the time to review this carefully!
I reverted change for the normalization and implemented matches method and some test cases.
When you get a chance, please take another look. 🙇♂️
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4323519410
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java:
##########
@@ -122,6 +127,93 @@ public void fromHeartbeatRequestShouldCreateCorrectTopology() {
assertEquals(mkSubtopology2(), topology.subtopologies().get(SUBTOPOLOGY_ID_2));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void copartitionOrderShouldNotAffectTopologyEquality(String topicType) {
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups(topicType);
+ StreamsGroupHeartbeatRequestData.Topology originalRequest = request.duplicate();
+ StreamsGroupTopologyValue record = StreamsCoordinatorRecordHelpers.convertToStreamsGroupTopologyRecord(request).duplicate();
+ List<StreamsGroupTopologyValue.CopartitionGroup> groups = record.subtopologies().get(0).copartitionGroups();
+ Collections.reverse(groups);
+ for (StreamsGroupTopologyValue.CopartitionGroup group : groups) {
+ Collections.reverse(group.sourceTopics());
+ Collections.reverse(group.sourceTopicRegex());
+ Collections.reverse(group.repartitionSourceTopics());
+ }
+ StreamsGroupTopologyValue originalRecord = record.duplicate();
+
+ StreamsTopology replayed = StreamsTopology.fromRecord(record);
+ StreamsTopology received = StreamsTopology.fromHeartbeatRequest(request);
+
+ assertEquals(received, replayed);
+ assertEquals(received.hashCode(), replayed.hashCode());
+ assertEquals(originalRecord, record);
+ assertEquals(originalRequest, request);
+ // The lookup map must also refer to the normalized subtopology.
+ assertSame(replayed.subtopologies().get(SUBTOPOLOGY_ID_1), replayed.sourceTopicMap().get("z"));
+ assertEquals(List.of("z", "y", "x", "w"), replayed.subtopologies().get(SUBTOPOLOGY_ID_1).sourceTopics());
+ assertEquals(originalRecord.subtopologies().get(0).sourceTopicRegex(),
+ replayed.subtopologies().get(SUBTOPOLOGY_ID_1).sourceTopicRegex());
+ assertEquals(originalRecord.subtopologies().get(0).repartitionSourceTopics(),
+ replayed.subtopologies().get(SUBTOPOLOGY_ID_1).repartitionSourceTopics());
+
+ // Exercise normalization of an unsorted heartbeat as well as an unsorted stored record.
+ StreamsGroupHeartbeatRequestData.Topology reorderedRequest = request.duplicate();
+ Collections.reverse(reorderedRequest.subtopologies().get(0).copartitionGroups());
+ reorderedRequest.subtopologies().get(0).copartitionGroups().forEach(group -> {
+ Collections.reverse(group.sourceTopics());
+ Collections.reverse(group.sourceTopicRegex());
+ Collections.reverse(group.repartitionSourceTopics());
+ });
+ StreamsGroupHeartbeatRequestData.Topology originalReorderedRequest = reorderedRequest.duplicate();
+ assertEquals(received, StreamsTopology.fromHeartbeatRequest(reorderedRequest));
+ assertEquals(originalReorderedRequest, reorderedRequest);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"source", "regex", "repartition"})
+ public void differentCopartitionConstraintsShouldNotBeEqual(String topicType) {
+ StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups(topicType);
+ StreamsTopology original = StreamsTopology.fromHeartbeatRequest(request);
+ StreamsGroupHeartbeatRequestData.Topology changed = request.duplicate();
+ changed.subtopologies().get(0).copartitionGroups().set(2,
+ copartitionGroup(topicType, List.of((short) 1, (short) 3)));
+
+ assertNotEquals(original, StreamsTopology.fromHeartbeatRequest(changed));
+ assertNotEquals(original, StreamsTopology.fromHeartbeatRequest(request.duplicate().setEpoch(2)));
+ }
+
+ private StreamsGroupHeartbeatRequestData.Topology topologyWithCopartitionGroups(String topicType) {
+ return new StreamsGroupHeartbeatRequestData.Topology()
+ .setEpoch(1)
+ .setSubtopologies(List.of(new StreamsGroupHeartbeatRequestData.Subtopology()
+ .setSubtopologyId(SUBTOPOLOGY_ID_1)
+ // Keep these unsorted to verify that index targets are preserved.
+ .setSourceTopics(List.of("z", "y", "x", "w"))
+ .setSourceTopicRegex(List.of("z.*", "y.*", "x.*", "w.*"))
+ .setRepartitionSourceTopics(List.of(
+ new StreamsGroupHeartbeatRequestData.TopicInfo().setName("repartition-z"),
+ new StreamsGroupHeartbeatRequestData.TopicInfo().setName("repartition-y"),
+ new StreamsGroupHeartbeatRequestData.TopicInfo().setName("repartition-x"),
+ new StreamsGroupHeartbeatRequestData.TopicInfo().setName("repartition-w")
+ ))
+ .setCopartitionGroups(List.of(
+ copartitionGroup(topicType, List.of((short) 0)),
+ copartitionGroup(topicType, List.of((short) 0, (short) 1)),
+ copartitionGroup(topicType, List.of((short) 2, (short) 3))
Review Comment:
Could we also cover a group that mixes source and repartition indexes, e.g. `{source: [1], repartition: [0]}`? That's the typical join case, and it's where the `thenComparing` tie-breaks actually matter.
##########
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:
nit: this is a hand-rolled lexicographic compare. Not a big deal, but since the lists are sorted at this point, would something like converting to `short[]` and using `Arrays.compare` be simpler? Fine to keep as is if you prefer.
##########
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:
This deep-copies the whole subtopology, including all the TopicInfos and their configs, just to sort the copartition groups, and `fromHeartbeatRequest` now copies twice (convert + duplicate). Could we only copy the copartition groups? If we sort in `convertToStreamsGroupTopologyRecord` as suggested in the review body, the heartbeat path doesn't need a copy at all.
Also, `convertToTopicInfo` maps a null `topicConfigs` to null, and `duplicate()` would NPE on that. A deserialized request can't have null there since the field isn't nullable, so maybe that null branch is just dead code. But it's another reason not to duplicate fields we don't touch.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -21330,6 +21330,54 @@ barTopicName, computeTopicHash(barTopicName, metadataImage)
assertRecordsEquals(expectedRecords, result.records());
}
+ @Test
+ public void testStreamsGroupMemberJoiningWithReorderedCopartitionGroups() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new Subtopology()
+ .setSubtopologyId("subtopology1")
+ .setSourceTopics(List.of("foo", "bar", "baz", "qux"))
+ .setCopartitionGroups(List.of(
+ new CopartitionGroup().setSourceTopics(List.of((short) 3, (short) 2)),
+ new CopartitionGroup().setSourceTopics(List.of((short) 1, (short) 0))
+ ))
+ ));
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(Uuid.randomUuid(), "foo", 3)
+ .addTopic(Uuid.randomUuid(), "bar", 3)
+ .addTopic(Uuid.randomUuid(), "baz", 3)
+ .addTopic(Uuid.randomUuid(), "qux", 3)
+ .buildCoordinatorMetadataImage())
+ .withConfig(GroupCoordinatorConfig.STREAMS_GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, 0)
+ .build();
+ assignor.prepareGroupAssignment(Map.of(memberId, TasksTuple.EMPTY));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecord(groupId, topology));
+
+ Topology reorderedTopology = topology.duplicate();
+ reorderedTopology.subtopologies().get(0).setCopartitionGroups(List.of(
+ new CopartitionGroup().setSourceTopics(List.of((short) 1, (short) 0)),
+ new CopartitionGroup().setSourceTopics(List.of((short) 2, (short) 3))
+ ));
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(1500)
+ .setTopology(reorderedTopology)
+ .setProcessId(DEFAULT_PROCESS_ID)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of()));
+
+ assertEquals(Errors.NONE.code(), result.response().data().errorCode());
+ assertTrue(result.response().data().memberEpoch() > 0);
Review Comment:
Maybe also assert that `result.records()` contains no new topology record, so we know the request was matched against the existing topology?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4336637074
##########
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:
Definitely agree.
I think this is an area where regressions could easily be introduced if new fields are added in the future.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6291430425
Small addition after looking at how the client builds the groups: the order comes from `HashSet` iteration in `InternalTopologyBuilder#copartitionGroups`, which is deterministic for the same topology code. So instances running the same code always send the same order, and the mismatch only happens when equivalent topologies are written differently (e.g. join order swapped) and the topic names collide in the same hash bucket.
That also means existing records generally match what current clients send, which makes your point about sorting on the client even stronger. A sorting client would usually differ from the persisted order on an unpatched broker.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4412318571
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested : topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject duplicates so a
+ // malformed request cannot match one stored subtopology twice and omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its lists
+ // may be unmodifiable or shared with 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) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol version
Review Comment:
I'd drop this TODO and move it to a JIRA.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6515196511
@lucasbru
Thanks for taking the time to look into this continuously!
I've addressed them based on your comments.
When you get a chance, please take a look!
Regarding backporting, this change should be suitable for older release branches because normalization is performed on in-memory copies, without changing the persisted records or their format. Compatibility with each target branch would still need to be verified. I will prepare PR for client side, but it might not be backported.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6291355033
Thanks for the clarification!
I've created new ticket for client side. (https://issues.apache.org/jira/browse/KAFKA-21320)
Currently, I'm working on addressing your comments in this PR.
I'll be back ASAP with new commit!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23855:
URL: https://github.com/apache/kafka/pull/23855#issuecomment-6290562779
@lucasbru
Thanks for taking the time to review!
I agree that we should fix the client side as well. Given the backport implications, would it make sense to track that change in a separate ticket and PR?
AFAIK, Copartition groups are persisted, including their ordering. If an existing client is upgraded to send sorted groups while the broker remains unpatched, the broker could reject the topology because its ordering differs from the persisted record, even though the topology is semantically unchanged.
Separating the client-side change would let us backport the broker normalization independently and evaluate the client-side backport with this compatibility concern in mind.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4411202776
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +134,180 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch()
+ || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+
+ Set<String> seenIds = new HashSet<>();
+ for (StreamsGroupHeartbeatRequestData.Subtopology requested : topology.subtopologies()) {
+ // Valid Streams clients send unique subtopology IDs. Reject duplicates so a
+ // malformed request cannot match one stored subtopology twice and omit another.
+ if (!seenIds.add(requested.subtopologyId())) {
+ return false;
+ }
+
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static Subtopology normalizeCopartitionGroups(Subtopology subtopology) {
+ // Copy before sorting: the record may be persisted as-is, and its lists
+ // may be unmodifiable or shared with 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) {
+ for (int i = 0; i < Math.min(first.size(), second.size()); i++) {
+ int comparison = Short.compare(first.get(i), second.get(i));
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ return Integer.compare(first.size(), second.size());
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
+ List<CopartitionGroup> current,
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> requested
+ ) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+
+ // TODO: Consider removing request normalization if a future protocol version
+ // requires canonical copartition ordering and older versions are no longer
+ // supported. Validate the required ordering before comparing.
+
+ // Copy before sorting to preserve the request and its index lists.
+ List<StreamsGroupHeartbeatRequestData.CopartitionGroup> sortedRequested =
+ new ArrayList<>(requested.size());
+
+ for (StreamsGroupHeartbeatRequestData.CopartitionGroup group : requested) {
+ StreamsGroupHeartbeatRequestData.CopartitionGroup copy = group.duplicate();
+ Collections.sort(copy.sourceTopics());
+ Collections.sort(copy.sourceTopicRegex());
+ Collections.sort(copy.repartitionSourceTopics());
+ sortedRequested.add(copy);
+ }
+
+ sortedRequested.sort(
+ Comparator.comparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::sourceTopicRegex,
+ StreamsTopology::compareTopicIndexes
+ )
+ .thenComparing(
+ StreamsGroupHeartbeatRequestData.CopartitionGroup::repartitionSourceTopics,
+ StreamsTopology::compareTopicIndexes
+ )
+ );
+
+ // Stored groups were already sorted by fromRecord.
+ for (int i = 0; i < current.size(); i++) {
+ if (!matchesCopartitionGroup(current.get(i), sortedRequested.get(i))) {
Review Comment:
`current` is not necessarily sorted: `StreamsTopology.fromHeartbeatRequest(...)` and the public constructors still preserve the supplied ordering. For example, with a copartition group containing `[1, 0]`, `StreamsTopology.fromHeartbeatRequest(request).matches(request)` returns false because only the request side is sorted to `[0, 1]`. Please enforce normalization on every construction path or normalize both sides here, and add a self-match regression test using an unsorted request.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21302: Sort copartition groups and indexes for consistent Streams topology comparison [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4411082799
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +131,178 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic indexes are compared
+ * without regard to order, but duplicate occurrences must still match. All other lists
+ * retain their order-sensitive comparison, including the topic lists referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology) {
+ if (topologyEpoch != topology.epoch() || subtopologies.size() != topology.subtopologies().size()) {
+ return false;
+ }
+ for (int i = 0; i < topology.subtopologies().size(); i++) {
+ StreamsGroupHeartbeatRequestData.Subtopology requested = topology.subtopologies().get(i);
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+
+ // Do not let repeated IDs match one stored subtopology while omitting another.
+ for (int j = 0; j < i; j++) {
+ if (requested.subtopologyId().equals(topology.subtopologies().get(j).subtopologyId())) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current, StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ if (!current.subtopologyId().equals(requested.subtopologyId())) {
+ return false;
+ }
+
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ && current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(), requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(), requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(), requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current, List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested = requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() != fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(), fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(), requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
Review Comment:
I agree.
In that case, performance degradation is expected.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org