KAFKA-18613: Improve test coverage for missing topics #19189

lucasbru · 2025-03-12T14:02:22Z

Tests for joining with missing source topics,
internal topics, incorrectly partitioned topics,
and stale topologies.

This is a stacked PR. Only review commits starting from "KAFKA-18613 : Improve test coverage for missing topics"

Basic streams group heartbeat handling. - No support for static membership - No support for configurations (using constants instead) - No support for regular expressions

…at_1

lucasbru · 2025-03-12T14:03:29Z

@cadonna Could you have a look? I'm adding three tests as a separate PR.

lucasbru · 2025-03-12T14:04:31Z

...or/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java

@@ -180,17 +178,13 @@ private static void enforceCopartitioning(final StreamsTopology topology,
                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());

-        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
-            log.info("Skipping the repartition topic validation since there are no repartition topics.");


Skipping here is actually not correct - we need to enforce copartitioning also for source topics.

…at_1

bbejeck

Thanks for the PR @lucasbru, overall this looks good, I left a couple of comments.

bbejeck · 2025-03-14T18:00:31Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

-            } else if (!updatedTopology.equals(group.topology().get())) {
+                return streamsTopologyFromRequest;
+            } else if (group.topology().get().topologyEpoch() > topology.epoch()) {
+                log.info("[GroupId {}][MemberId {}] Member joined with stake topology epoch {}", groupId, memberId, topology.epoch());


What's stake topology? did you mean state

Good catch. It was supposed to be stale.

bbejeck · 2025-03-14T18:22:11Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

                throw new InvalidRequestException("Topology updates are not supported yet.");
+            } else {
+                log.debug("[GroupId {}][MemberId {}] Member joined with currently initialized topology {}", groupId, memberId, topology.epoch());
+                return group.topology().get();


should any of the calls to group.topology().get() check for isPresent()?

I think if group.topology.isEmpty is true, the control-flow will go to the first if-block, so we won't need an extra isPresent check her.e

bbejeck · 2025-03-14T18:27:36Z

...or/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java

+        for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+            for (Set<String> copartitionGroup : copartitionGroups) {
+                decidedPartitionCountsForInternalTopics.putAll(
+                    copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));


How would this affect a topology where the user wants to expand partitions with a repartition operator? Would it throw an error in that case?

This should work as in the old protocol. I think if you use a repartition operator, the repartition topics should be included in fixedRepartitionTopics.

bbejeck · 2025-03-14T18:50:18Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+            StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, expectedMember),
+            StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecord(groupId, topology),
+            StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord(groupId, Map.of(
+                fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 6)


nit: put Map.of on a new line

bbejeck · 2025-03-14T18:51:05Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+    public void testStreamsGroupMemberJoiningWithMissingSourceTopic() {
+        String groupId = "fooup";
+        String memberId = Uuid.randomUuid().toString();
+


nit: remove extra line

bbejeck · 2025-03-14T18:51:37Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+    public void testStreamsGroupMemberJoiningWithMissingInternalTopic() {
+        String groupId = "fooup";
+        String memberId = Uuid.randomUuid().toString();
+


nit: remove extra line same for similar tests below

bbejeck · 2025-03-14T18:52:32Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+                .setWarmupTasks(List.of()));
+
+        assertEquals(
+            Map.of(barTopicName, new CreatableTopic().setName(barTopicName).setNumPartitions(6).setReplicationFactor((short) -1)),


nit: put builder calls on new line

…at_1

Tests for joining with missing source topics, internal topics, incorrectly partitioned topics, and stale topologies.

lucasbru

@bbejeck Thanks for the comments! Ready for re-review

lucasbru · 2025-03-12T14:06:30Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

-            } else if (!updatedTopology.equals(group.topology().get())) {
+                return streamsTopologyFromRequest;
+            } else if (group.topology().get().topologyEpoch() > topology.epoch()) {
+                log.info("[GroupId {}][MemberId {}] Member joined with stake topology epoch {}", groupId, memberId, topology.epoch());


I had the STALE_TOPOLOGY_EPOCH status implemented, but rejected all topology updates, which somewhat contradicts each other. I added this branch to basically ignore stale topologies.

lucasbru · 2025-03-17T08:55:07Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

-            } else if (!updatedTopology.equals(group.topology().get())) {
+                return streamsTopologyFromRequest;
+            } else if (group.topology().get().topologyEpoch() > topology.epoch()) {
+                log.info("[GroupId {}][MemberId {}] Member joined with stake topology epoch {}", groupId, memberId, topology.epoch());


Good catch. It was supposed to be stale.

lucasbru · 2025-03-17T08:56:14Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

                throw new InvalidRequestException("Topology updates are not supported yet.");
+            } else {
+                log.debug("[GroupId {}][MemberId {}] Member joined with currently initialized topology {}", groupId, memberId, topology.epoch());
+                return group.topology().get();


I think if group.topology.isEmpty is true, the control-flow will go to the first if-block, so we won't need an extra isPresent check her.e

lucasbru · 2025-03-17T08:58:09Z

...or/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java

+        for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+            for (Set<String> copartitionGroup : copartitionGroups) {
+                decidedPartitionCountsForInternalTopics.putAll(
+                    copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));


This should work as in the old protocol. I think if you use a repartition operator, the repartition topics should be included in fixedRepartitionTopics.

lucasbru · 2025-03-17T08:59:15Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+    public void testStreamsGroupMemberJoiningWithMissingSourceTopic() {
+        String groupId = "fooup";
+        String memberId = Uuid.randomUuid().toString();
+


lucasbru · 2025-03-17T08:59:58Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+            StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, expectedMember),
+            StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecord(groupId, topology),
+            StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord(groupId, Map.of(
+                fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 6)


lucasbru · 2025-03-17T09:01:51Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+    public void testStreamsGroupMemberJoiningWithMissingInternalTopic() {
+        String groupId = "fooup";
+        String memberId = Uuid.randomUuid().toString();
+


lucasbru · 2025-03-17T09:02:25Z

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

+                .setWarmupTasks(List.of()));
+
+        assertEquals(
+            Map.of(barTopicName, new CreatableTopic().setName(barTopicName).setNumPartitions(6).setReplicationFactor((short) -1)),


lucasbru added 7 commits March 6, 2025 17:59

KAFKA-18613: Add StreamsGroupHeartbeat handler in the group coordinator

1e3a1fa

Basic streams group heartbeat handling. - No support for static membership - No support for configurations (using constants instead) - No support for regular expressions

spotless

fccd6e1

fix tests

acb2b5f

Merge remote-tracking branch 'apache/trunk' into kip1071merge/heartbe…

052b64c

…at_1

refactoring to reuse StreamsTopology.fromRequest

787d8d4

comments

c291405

Merge remote-tracking branch 'apache/trunk' into kip1071merge/heartbe…

172dbdf

…at_1

lucasbru requested a review from cadonna March 12, 2025 14:02

lucasbru commented Mar 12, 2025

View reviewed changes

lucasbru added 2 commits March 12, 2025 21:44

comments addressed

537d4fe

Merge remote-tracking branch 'apache/trunk' into kip1071merge/heartbe…

954bb90

…at_1

lucasbru added streams KIP-1071 PRs related to KIP-1071 labels Mar 14, 2025

Merge remote-tracking branch 'apache/trunk' into kip1071merge/heartbe…

6482e6b

…at_1

lucasbru force-pushed the kip1071merge/heartbeat_2 branch from efe995e to a7073cc Compare March 14, 2025 11:06

bbejeck reviewed Mar 14, 2025

View reviewed changes

lucasbru added 4 commits March 17, 2025 09:49

comments

da10d3f

Merge remote-tracking branch 'apache/trunk' into kip1071merge/heartbe…

a1f24ab

…at_1

KAFKA-18613: Improve test coverage for missing topics

5a2bde6

Tests for joining with missing source topics, internal topics, incorrectly partitioned topics, and stale topologies.

comments

407a15c

lucasbru force-pushed the kip1071merge/heartbeat_2 branch from a7073cc to 407a15c Compare March 17, 2025 09:20

lucasbru commented Mar 17, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18613: Improve test coverage for missing topics #19189

KAFKA-18613: Improve test coverage for missing topics #19189

lucasbru commented Mar 12, 2025

lucasbru commented Mar 12, 2025

lucasbru Mar 12, 2025

bbejeck left a comment

bbejeck Mar 14, 2025

lucasbru Mar 17, 2025

bbejeck Mar 14, 2025

lucasbru Mar 17, 2025

bbejeck Mar 14, 2025

lucasbru Mar 17, 2025

bbejeck Mar 14, 2025

lucasbru Mar 17, 2025

bbejeck Mar 14, 2025

lucasbru Mar 17, 2025

bbejeck Mar 14, 2025

lucasbru Mar 17, 2025

bbejeck Mar 14, 2025

lucasbru Mar 17, 2025

lucasbru left a comment

lucasbru Mar 12, 2025

lucasbru Mar 17, 2025

lucasbru Mar 17, 2025

lucasbru Mar 17, 2025

lucasbru Mar 17, 2025

lucasbru Mar 17, 2025

lucasbru Mar 17, 2025

lucasbru Mar 17, 2025

KAFKA-18613: Improve test coverage for missing topics #19189

Are you sure you want to change the base?

KAFKA-18613: Improve test coverage for missing topics #19189

Conversation

lucasbru commented Mar 12, 2025

This is a stacked PR. Only review commits starting from "KAFKA-18613: Improve test coverage for missing topics"

lucasbru commented Mar 12, 2025

Choose a reason for hiding this comment

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lucasbru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

This is a stacked PR. Only review commits starting from "KAFKA-18613 : Improve test coverage for missing topics"