KAFKA-18613: Add StreamsGroupHeartbeat handler in the group coordinator #19114

lucasbru · 2025-03-05T14:13:41Z

Basic streams group heartbeat handling. The main part of are the unit tests that make sure that we behave, for the most part, like a consumer group.

No support for static membership
No support for configurations (using constants instead)
No support for regular expressions

...oordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupBuilder.java

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java

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

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

…at_1

cadonna

Thanks for the PR, @lucasbru !

I did a first pass over the production code. I haven't looked at GroupMetadataTest, yet.

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

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTuple.java

…at_1

lucasbru · 2025-03-10T13:50:59Z

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

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

cadonna

Thanks for the updates @lucasbru !

I reviewed half of the tests in GroupMetadataTest.

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

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

cadonna · 2025-03-11T14:43:22Z

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

+     * @throws InvalidRequestException if the request is not valid.
+     * @throws UnsupportedAssignorException if the assignor is not supported.
+     */
+    private static void throwIfStreamsGroupHeartbeatRequestIsInvalid(


I am missing the following check from the KIP:
"Each element of ActiveTasks, StandbyTasks and WarmupTasks has to be a valid task ID in the topology initialized for the group ID."
Clearly, you cannot put it in this method, but I could also not find it in the heartbeat handler.
Do you still plan to add this check?

I think I wouldn't add this now. It seems contradict the story about topology updating (e.g. removing a subtopology). We'd add a check now, and remove it again in the next version, no? Also this PR seems to big already ;). We could consider creating a little ticket, but I don't think we should add it now.

I agree on the "not adding now part". I do not completely understand the "topology updating" part. We can still perform this check for the current topology epoch, no?

Maybe, but we have to think this through. Even if we let clients crash only when they use the same topology epoch, that would mean that scaling down the number of input partitions for a stateless application would become impossible. I am wondering if it isn't a better course of action to just ignore tasks that we do not expect. After all, the client is just saying what he thinks he owns. If we'd just ignore those tasks, the next target assignment would not include the tasks, so the client would automatically be instructed to revoke whatever tasks it thinks it currently has.

https://issues.apache.org/jira/browse/KAFKA-19044

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

…at_1

lucasbru

@cadonna Thanks for the comments! Ready for re-review.

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

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

lucasbru · 2025-03-12T20:43:18Z

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

+     * @throws InvalidRequestException if the request is not valid.
+     * @throws UnsupportedAssignorException if the assignor is not supported.
+     */
+    private static void throwIfStreamsGroupHeartbeatRequestIsInvalid(


I think I wouldn't add this now. It seems contradict the story about topology updating (e.g. removing a subtopology). We'd add a check now, and remove it again in the next version, no? Also this PR seems to big already ;). We could consider creating a little ticket, but I don't think we should add it now.

…at_1

bbejeck

Thanks for the PR @lucasbru, I a few comments, but otherwise this lgtm

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

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

…at_1

lucasbru · 2025-03-17T08:52:46Z

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

lucasbru · 2025-03-24T14:07:32Z

@cadonna @bbejeck this PR has been pending for 3 weeks. Anything I can do to push it over the line?

cadonna

Thanks for the updates, @lucasbru !

I reviewed also the second part of the tests.

Once you resolved to my comments, the PR is ready for merge.

cadonna · 2025-03-26T10:02:13Z

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

+                .withTargetAssignment(memberId, TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+                    TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
+                .withTargetAssignmentEpoch(10)
+                .withTopology(StreamsTopology.fromHeartbeatRequest(topology))


This test would be easier to understand if you added the following here:

.withPartitionMetadata(Map.of( fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 3), barTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(barTopicId, barTopicName, 3) ))

or even add a variable changedPartitionCount = 3 and then:

.withPartitionMetadata(Map.of( fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, changedPartitionCount), barTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(barTopicId, barTopicName, 3) ))

It took me quite some time to understand in the partition metadata was from no metadata to some metadata.

cadonna · 2025-03-26T10:18:59Z

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

+    }
+
+    @Test
+    public void testStreamsLeavingMemberBumpsGroupEpoch() {


Maybe a better name would be testStreamsLeavingMemberRemovesMemberAndBumpsGroupEpoch

cadonna · 2025-03-26T10:40:23Z

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

+                    TaskAssignmentTestUtil.mkTasks(subtopology1, 3, 4, 5),
+                    TaskAssignmentTestUtil.mkTasks(subtopology2, 2)))
+                .withTargetAssignmentEpoch(10))
+            .build();


IMO, you should also add the partition metadata to the group. The test relies on the order of checking first the member changes and then partition metadata changes. In the case the production code changes that order this test would fail in this specific location although it should not. That is better than the other way around, i.e., the test does not fail although it should. Nevertheless, making the test more robust to production code changes saves some debugging pain.

cadonna · 2025-03-26T11:07:20Z

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

+
+        // Member 1 joins the streams group. The request fails because the
+        // target assignment computation failed.
+        assertThrows(UnknownServerException.class, () ->


Could you please also verify the exception message?

cadonna · 2025-03-26T11:11:52Z

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

+        context.rollback();
+
+        // However, the next heartbeat should detect the divergence based on the epoch and trigger
+        // a metadata refr


nit:

Suggested change

// a metadata refr

// a metadata refresh.

cadonna · 2025-03-26T11:19:22Z

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

+        );
+
+        // Advance time past the revocation timeout.
+        List<ExpiredTimeout<Void, CoordinatorRecord>> timeouts = context.sleep(10000 + 1);


Could you please use a variable revocationTimeout instead of 10000?

bbejeck

Thanks for the update @lucasbru LGTM modulo addressing Bruno's oustanding comments.

lucasbru commented Mar 5, 2025

View reviewed changes

...oordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupBuilder.java Show resolved Hide resolved

lucasbru commented Mar 5, 2025

View reviewed changes

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java Show resolved Hide resolved

lucasbru commented Mar 5, 2025

View reviewed changes

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

lucasbru requested a review from cadonna March 5, 2025 14:20

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

lucasbru added 2 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

lucasbru force-pushed the kip1071merge/heartbeat_1 branch from b442946 to fccd6e1 Compare March 6, 2025 16:59

lucasbru added 3 commits March 7, 2025 14:41

fix tests

acb2b5f

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

052b64c

…at_1

refactoring to reuse StreamsTopology.fromRequest

787d8d4

cadonna reviewed Mar 7, 2025

View reviewed changes

cadonna added the core Kafka Broker label Mar 7, 2025

lucasbru added 2 commits March 10, 2025 14:47

comments

c291405

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

172dbdf

…at_1

cadonna reviewed Mar 11, 2025

View reviewed changes

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

cadonna reviewed 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 commented Mar 12, 2025

View reviewed changes

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

6482e6b

…at_1

bbejeck reviewed Mar 14, 2025

View reviewed changes

lucasbru added 2 commits March 17, 2025 09:49

comments

da10d3f

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

a1f24ab

…at_1

lucasbru requested a review from mjsax March 24, 2025 19:37

cadonna reviewed Mar 26, 2025

View reviewed changes

bbejeck approved these changes Mar 26, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18613: Add StreamsGroupHeartbeat handler in the group coordinator #19114

KAFKA-18613: Add StreamsGroupHeartbeat handler in the group coordinator #19114

lucasbru commented Mar 5, 2025

cadonna left a comment

lucasbru commented Mar 10, 2025

cadonna left a comment

cadonna Mar 11, 2025

lucasbru Mar 12, 2025

cadonna Mar 25, 2025

lucasbru Mar 25, 2025 •

edited

Loading

lucasbru Mar 25, 2025

lucasbru left a comment

lucasbru Mar 12, 2025

bbejeck left a comment

lucasbru commented Mar 17, 2025

lucasbru commented Mar 24, 2025

cadonna left a comment

cadonna Mar 26, 2025

cadonna Mar 26, 2025

cadonna Mar 26, 2025

cadonna Mar 26, 2025

cadonna Mar 26, 2025

cadonna Mar 26, 2025

bbejeck left a comment

KAFKA-18613: Add StreamsGroupHeartbeat handler in the group coordinator #19114

Are you sure you want to change the base?

KAFKA-18613: Add StreamsGroupHeartbeat handler in the group coordinator #19114

Conversation

lucasbru commented Mar 5, 2025

cadonna left a comment

Choose a reason for hiding this comment

lucasbru commented Mar 10, 2025

cadonna 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

lucasbru Mar 25, 2025 • edited Loading

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

bbejeck left a comment

Choose a reason for hiding this comment

lucasbru commented Mar 17, 2025

lucasbru commented Mar 24, 2025

cadonna 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

bbejeck left a comment

Choose a reason for hiding this comment

lucasbru Mar 25, 2025 •

edited

Loading