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