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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
