diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java index 11ffc392366f4..4e2c8f5b4f700 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java @@ -2520,16 +2520,15 @@ private StreamsTopology maybeUpdateTopology(final String groupId, final StreamsGroup group, final List records) { if (topology != null) { - StreamsTopology streamsTopologyFromRequest = StreamsTopology.fromHeartbeatRequest(topology); if (group.topology().isEmpty()) { log.info("[GroupId {}][MemberId {}] Member initialized the topology with epoch {}", groupId, memberId, topology.epoch()); StreamsGroupTopologyValue recordValue = convertToStreamsGroupTopologyRecord(topology); records.add(newStreamsGroupTopologyRecord(groupId, recordValue)); - return streamsTopologyFromRequest; + return StreamsTopology.fromRecord(recordValue); } else if (group.topology().get().topologyEpoch() > topology.epoch()) { log.info("[GroupId {}][MemberId {}] Member joined with stale topology epoch {}", groupId, memberId, topology.epoch()); return group.topology().get(); - } else if (!group.topology().get().equals(streamsTopologyFromRequest)) { + } else if (!group.topology().get().matches(topology)) { throw new InvalidRequestException("Topology updates are not supported yet."); } else { log.debug("[GroupId {}][MemberId {}] Member joined with currently initialized topology {}", groupId, memberId, topology.epoch()); diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java index 9d7d097f80f3d..31f9288cfdc23 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java @@ -19,12 +19,14 @@ import org.apache.kafka.common.message.StreamsGroupDescribeResponseData; import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData; import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue; +import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.CopartitionGroup; import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.Subtopology; import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.TopicInfo; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; @@ -111,7 +113,8 @@ public Set requiredTopics() { public static StreamsTopology fromRecord(StreamsGroupTopologyValue record) { return new StreamsTopology( record.epoch(), - record.subtopologies().stream().collect(Collectors.toMap(Subtopology::subtopologyId, x -> x)) + record.subtopologies().stream() + .collect(Collectors.toMap(Subtopology::subtopologyId, x -> x)) ); } @@ -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 current, List 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 current, + List 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 current, + List requested + ) { + if (current.size() != requested.size()) { + return false; + } + + boolean sameOrder = true; + for (int i = 0; i < current.size(); i++) { + if (!matchesCopartitionGroup(current.get(i), requested.get(i))) { + sameOrder = false; + break; + } + } + if (sameOrder) { + return true; + } + + // Count equivalent occurrences rather than allocating sorted copies + // or sets, which would lose duplicate groups. + for (int i = 0; i < current.size(); i++) { + CopartitionGroup group = current.get(i); + StreamsGroupHeartbeatRequestData.CopartitionGroup representative = null; + int requestedCount = 0; + for (int j = 0; j < requested.size(); j++) { + if (matchesCopartitionGroup(group, requested.get(j))) { + representative = requested.get(j); + requestedCount++; + } + } + if (representative == null) { + return false; + } + int currentCount = 0; + for (int j = 0; j < current.size(); j++) { + if (matchesCopartitionGroup(current.get(j), representative)) { + currentCount++; + } + } + if (currentCount != requestedCount) { + return false; + } + } + return true; + } + + private static boolean matchesCopartitionGroup( + CopartitionGroup current, + StreamsGroupHeartbeatRequestData.CopartitionGroup requested + ) { + return sameTopicIndexes(current.sourceTopics(), requested.sourceTopics()) + && sameTopicIndexes(current.sourceTopicRegex(), requested.sourceTopicRegex()) + && sameTopicIndexes(current.repartitionSourceTopics(), requested.repartitionSourceTopics()); + } + + private static boolean sameTopicIndexes(List current, List requested) { + // [0] != [0, 1, 2] + if (current.size() != requested.size()) { + return false; + } + + // [0, 1, 2] == [0, 1, 2] + if (current.equals(requested)) { + return true; + } + + int size = current.size(); + + // Ignore ordering while preserving duplicate counts. + // For example, [0, 0, 1] matches [1, 0, 0], but not [0, 1, 1]. + for (int i = 0; i < size; i++) { + short index = current.get(i); + int currentCount = 0; + int requestedCount = 0; + + for (int j = 0; j < size; j++) { + if (current.get(j) == index) { + currentCount++; + } + if (requested.get(j) == index) { + requestedCount++; + } + } + if (currentCount != requestedCount) { + return false; + } + } + return true; + } + public StreamsGroupDescribeResponseData.Topology asStreamsGroupDescribeTopology() { return new StreamsGroupDescribeResponseData.Topology() .setEpoch(topologyEpoch) diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java index 899c45693fb35..d396176c3135f 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java @@ -129,6 +129,7 @@ import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataKey; import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataValue; import org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMemberKey; +import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyKey; import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue; import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard; import org.apache.kafka.coordinator.group.modern.Assignment; @@ -21330,6 +21331,109 @@ barTopicName, computeTopicHash(barTopicName, metadataImage) assertRecordsEquals(expectedRecords, result.records()); } + @Test + public void testStreamsGroupMemberJoiningWithReorderedCopartitionGroups() { + // First - [[3, 2], [1, 0]] + // Second - [[1, 0], [2, 3]] + 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 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); + assertTrue(result.records().stream().noneMatch( + record -> record.key() instanceof StreamsGroupTopologyKey) + ); + } + + @Test + public void testStreamsGroupMemberJoiningWithDifferentCopartitionGroups() { + // First -> [[0, 1], [2, 3]] + // Second -> [[0, 2], [1, 3]] + String groupId = "fooup"; + String memberId = Uuid.randomUuid().toString(); + Topology topology = new Topology().setEpoch(1).setSubtopologies(List.of( + new Subtopology() + .setSubtopologyId("subtopology1") + .setSourceTopics(List.of("foo", "bar", "baz", "qux")) + .setCopartitionGroups(List.of( + new CopartitionGroup().setSourceTopics(List.of((short) 0, (short) 1)), + new CopartitionGroup().setSourceTopics(List.of((short) 2, (short) 3)) + )) + )); + 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)); + + // Keep the epoch and topics unchanged, but change which topics must be copartitioned. + Topology changedTopology = topology.duplicate(); + changedTopology.subtopologies().get(0).setCopartitionGroups(List.of( + new CopartitionGroup().setSourceTopics(List.of((short) 0, (short) 2)), + new CopartitionGroup().setSourceTopics(List.of((short) 1, (short) 3)) + )); + InvalidRequestException exception = assertThrows(InvalidRequestException.class, () -> context.streamsGroupHeartbeat( + new StreamsGroupHeartbeatRequestData() + .setGroupId(groupId) + .setMemberId(memberId) + .setMemberEpoch(0) + .setRebalanceTimeoutMs(1500) + .setTopology(changedTopology) + .setProcessId(DEFAULT_PROCESS_ID) + .setActiveTasks(List.of()) + .setStandbyTasks(List.of()) + .setWarmupTasks(List.of()))); + + assertEquals("Topology updates are not supported yet.", exception.getMessage()); + } + @Test public void testStreamsGroupMemberJoiningWithStaleTopology() { String groupId = "fooup"; diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java index 751be12df866e..a2023c52515ff 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java @@ -23,8 +23,12 @@ import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.TopicInfo; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.lang.reflect.Modifier; import java.util.Arrays; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -33,6 +37,8 @@ import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -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 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) { + // 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)); + } + + @Test + public void differentTopologyEpochShouldNotMatch() { + StreamsGroupHeartbeatRequestData.Topology request = topologyWithCopartitionGroups("source"); + StreamsTopology original = StreamsTopology.fromHeartbeatRequest(request); + StreamsGroupHeartbeatRequestData.Topology changed = request.duplicate().setEpoch(request.epoch() + 1); + + assertFalse(original.matches(changed)); + } + + + @Test + public void matchingShouldBeReviewedWhenMessageFieldsChange() { + // Count generated instance fields, including _unknownTaggedFields, so adding a tagged + // schema field also requires reviewing the explicit comparisons in StreamsTopology. + Map, Integer> expectedFieldCounts = Map.of( + StreamsGroupTopologyValue.class, 3, + StreamsGroupHeartbeatRequestData.Topology.class, 3, + Subtopology.class, 8, + StreamsGroupHeartbeatRequestData.Subtopology.class, 8, + StreamsGroupTopologyValue.CopartitionGroup.class, 4, + StreamsGroupHeartbeatRequestData.CopartitionGroup.class, 4, + TopicInfo.class, 5, + StreamsGroupHeartbeatRequestData.TopicInfo.class, 5, + StreamsGroupTopologyValue.TopicConfig.class, 3, + StreamsGroupHeartbeatRequestData.KeyValue.class, 3 + ); + expectedFieldCounts.forEach((type, expectedCount) -> { + long actualCount = Arrays.stream(type.getDeclaredFields()) + .filter(field -> !Modifier.isStatic(field.getModifiers())) + .count(); + assertEquals(expectedCount.longValue(), actualCount, + "Review StreamsTopology.matches for changes to " + type.getName()); + }); + } + + 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)) + )))); + } + + private StreamsGroupHeartbeatRequestData.CopartitionGroup copartitionGroup(String topicType, List indexes) { + StreamsGroupHeartbeatRequestData.CopartitionGroup group = new StreamsGroupHeartbeatRequestData.CopartitionGroup(); + return switch (topicType) { + case "source" -> group.setSourceTopics(indexes); + case "regex" -> group.setSourceTopicRegex(indexes); + case "repartition" -> group.setRepartitionSourceTopics(indexes); + default -> throw new IllegalArgumentException("Unknown topic type: " + topicType); + }; + } + @Test public void asStreamsGroupDescribeTopologyShouldReturnCorrectStructure() { Map subtopologies = mkMap(