KAFKA-18323: Add StreamsGroup class #18729

lucasbru · 2025-01-28T16:18:12Z

Implements a memory model for representing streams groups in the group coordinator, as well as group count and rebalance metrics.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

lucasbru · 2025-02-03T13:19:59Z

Rebased on latest target assignment builder

Copilot

Copilot reviewed 8 out of 16 changed files in this pull request and generated 1 comment.

Files not reviewed (8)

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetrics.java: Evaluated as low risk
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java: Evaluated as low risk
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java: Evaluated as low risk
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignor.java: Evaluated as low risk
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopology.java: Evaluated as low risk
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignorTest.java: Evaluated as low risk
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java: Evaluated as low risk
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java: Evaluated as low risk

Comments suppressed due to low confidence (1)

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TopologyMetadataTest.java:126

The method name testNumTasksThrowsExceptionWhenTopologyNotConfigured is inconsistent with the actual method being tested (maxNumInputPartitions). It should be renamed to testMaxNumInputPartitionsThrowsExceptionWhenTopologyNotConfigured.

void testNumTasksThrowsExceptionWhenTopologyNotConfigured() {

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

lucasbru · 2025-02-04T09:02:22Z

Rebased on latest trunk

cadonna

Thanks for the PR, @lucasbru !

I did a first pass over the production code.

cadonna · 2025-02-04T13:51:01Z

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

+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetActiveTasksAssignment;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetStandbyTasksAssignment;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetWarmupTasksAssignment;


Could you please add some more info to the javadoc? This data structure is a bit hard to grasp.
A visualization like the following would help:

subtopology -> partition -> memberId

I am wondering if using a class like TaskId for the key would make the code simpler.

What do you think of renaming this data structures to currrentXTaskToMemberId?

Hmm, this is not used at all right now. In the consumer group, this is used to optimize the assignor, but we don't use it in our assignors. I think I will remove it for now, we can bring it back once we optimize the assignors.

cadonna · 2025-02-04T14:04:53Z

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

+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> currentActiveTaskProcessId;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentStandbyTaskProcessIds;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentWarmupTaskProcessIds;


The above comment also applies to this data structure.

What do you think of renaming this data structures to currrentXTaskToProcessId?

I agree with Bruno about a slight renaming for clarity.

cadonna · 2025-02-04T15:34:48Z

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

+    }
+
+    /**
+     * @return An immutable Map containing all the target assignment keyed by member id.


Suggested change

* @return An immutable Map containing all the target assignment keyed by member id.

* @return An immutable map containing all the target assignment keyed by member ID.

Here and elsewhere

cadonna · 2025-02-04T15:35:01Z

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

+    /**
+     * Removes the target assignment of a member.
+     *
+     * @param memberId The member id.


Suggested change

* @param memberId The member id.

* @param memberId The member ID.

Here and in other java docs.

cadonna · 2025-02-04T15:39:55Z

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

+    }
+
+    /**
+     * Returns the current processId of a task or null if the task does not have one.


Suggested change

* Returns the current processId of a task or null if the task does not have one.

* Returns the current process ID of a task or null if the task does not have one.

Here and in other places.

cadonna · 2025-02-04T15:42:07Z

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

+        if (tasks == null) {
+            return null;
+        } else {
+            return tasks.getOrDefault(taskId, null);
+        }


Does it make sense to change this to Optional?

I think this may have to checked in tight loops inside the assignor. I think it's a good place to avoid creating extra objects just for returning a value.

cadonna

I made a pass over the test code.

cadonna · 2025-02-05T08:50:01Z

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

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


I do not understand this test. The test name is testAsListedGroup() but the method asListedGroup() is never called.

Yes, you are right. This is what is being tested in the other groups as well, interestingly. I think what is being tested here is covered by other tests already. I replaced it by a test that makes sense to me.

cadonna · 2025-02-05T08:52:00Z

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

+            .setGroupType(type().toString());
+    }
+
+    public ConfiguredTopology configuredTopology() {


Would it make sense to return the Optional here?

Let's try it.

cadonna · 2025-02-05T08:52:24Z

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

+        return configuredTopology.get().orElse(null);
+    }
+
+    public StreamsTopology topology() {


Would it make sense to return the Optional here?

Let's try it

cadonna · 2025-02-05T08:58:46Z

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

+        return topology.get().orElse(null);
+    }
+
+    public void setTopology(StreamsTopology topology) {


Could you please add a test that verifies that the configured topology is updated?

cadonna · 2025-02-05T09:14:41Z

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

+     * @param groupInstanceId The group instance id.
+     * @return The member id corresponding to the given instance id or null if it does not exist
+     */
+    public String staticMemberId(String groupInstanceId) {


Maybe also here we might want to return an Optional. However, we can postpone this decision to when we will use this method in the handlers.

Yes, let's postpone it.

cadonna · 2025-02-05T11:44:35Z

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

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


Could you please also add a case where the subtopology ID is modified?

I removed the interverted assignment because we are not using it (yet)

cadonna · 2025-02-05T11:47:22Z

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

+     *
+     * @param partitionMetadata The new partition metadata.
+     */
+    public void setPartitionMetadata(


Could you add unit test for this?

cadonna · 2025-02-05T11:48:32Z

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

+     * @param clusterImage         The current metadata for the Kafka cluster.
+     * @return An immutable map of partition metadata for each topic that the streams topology is using (besides non-repartition sink topics)
+     */
+    public Map<String, TopicMetadata> computePartitionMetadata(


I assume that this will be used in the future. Could you please add unit tests?

Yes. I also realized that the rack information that we included here for future proofing (but not rack-aware assignment is currently done / planned) was removed from consumer groups due to the memory pressure it caused. I also updated this to not include the rack information. We can always bring it back in the future. This required some additional changes (but only removals).

The part of the GroupMetadataManager that calls this method is a bit weird. The group computes and returns the partition metadata so that the computed partition metadata can be passed back into the group. We should consider simplifying that part when we get to the GroupMetadataManager.

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

bbejeck

Thanks for the PR @lucasbru - I don't have any additional comments beyond what Bruno has, overall this looks good to me modulo addressing those comments.

bbejeck · 2025-02-06T16:06:33Z

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

+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> currentActiveTaskProcessId;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentStandbyTaskProcessIds;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentWarmupTaskProcessIds;


I agree with Bruno about a slight renaming for clarity.

bbejeck · 2025-02-06T17:59:32Z

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

+    /**
+     * @return The current state based on committed offset.
+     */
+    public StreamsGroupState state(long committedOffset) {


When is this method called? How often is the offset updated? From the parameter name, I'm inferring that it occurs after every commit

It's not really called. We always get the latest state, although we can get the state at every committed offset of the consumer offset topic. I removed it, we can always bring it back

lucasbru

Thanks for the reviews! Addressed all comments. Ready for re-review @cadonna @bbejeck

lucasbru · 2025-02-05T21:20:27Z

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

+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetActiveTasksAssignment;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetStandbyTasksAssignment;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetWarmupTasksAssignment;


Hmm, this is not used at all right now. In the consumer group, this is used to optimize the assignor, but we don't use it in our assignors. I think I will remove it for now, we can bring it back once we optimize the assignors.

lucasbru · 2025-02-05T21:26:19Z

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

+    private final TimelineHashMap<String, TimelineHashMap<Integer, String>> currentActiveTaskProcessId;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentStandbyTaskProcessIds;
+    private final TimelineHashMap<String, TimelineHashMap<Integer, Set<String>>> currentWarmupTaskProcessIds;


lucasbru · 2025-02-05T21:27:47Z

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

+            .setGroupType(type().toString());
+    }
+
+    public ConfiguredTopology configuredTopology() {


Let's try it.

lucasbru · 2025-02-05T21:28:44Z

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

+        return configuredTopology.get().orElse(null);
+    }
+
+    public StreamsTopology topology() {


Let's try it

lucasbru · 2025-02-05T21:36:27Z

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

+    /**
+     * Removes the target assignment of a member.
+     *
+     * @param memberId The member id.


lucasbru · 2025-02-07T10:48:50Z

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

+     * @param groupInstanceId The group instance id.
+     * @return The member id corresponding to the given instance id or null if it does not exist
+     */
+    public String staticMemberId(String groupInstanceId) {


Yes, let's postpone it.

lucasbru · 2025-02-07T10:54:39Z

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

+     *
+     * @param partitionMetadata The new partition metadata.
+     */
+    public void setPartitionMetadata(


lucasbru · 2025-02-07T11:08:28Z

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

+     * @param clusterImage         The current metadata for the Kafka cluster.
+     * @return An immutable map of partition metadata for each topic that the streams topology is using (besides non-repartition sink topics)
+     */
+    public Map<String, TopicMetadata> computePartitionMetadata(


Yes. I also realized that the rack information that we included here for future proofing (but not rack-aware assignment is currently done / planned) was removed from consumer groups due to the memory pressure it caused. I also updated this to not include the rack information. We can always bring it back in the future. This required some additional changes (but only removals).

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

cadonna

Thanks for the updates, @lucasbru !

Here my comments.

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

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

cadonna · 2025-02-07T18:03:58Z

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

+     * @param clusterImage         The current metadata for the Kafka cluster.
+     * @return An immutable map of partition metadata for each topic that the streams topology is using (besides non-repartition sink topics)
+     */
+    public Map<String, TopicMetadata> computePartitionMetadata(


The part of the GroupMetadataManager that calls this method is a bit weird. The group computes and returns the partition metadata so that the computed partition metadata can be passed back into the group. We should consider simplifying that part when we get to the GroupMetadataManager.

lucasbru

Comments addressed, ready for re-review @cadonna

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

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

cadonna

Thanks for the updates, @lucasbru !

I had one minor comment.

LGTM!

Implements a memory model for representing streams groups in the group coordinator, as well as group count and rebalance metrics. Reviewers: Bill Bejeck <bill@confluent.io>, Bruno Cadonna <bruno@confluent.io>

lucasbru force-pushed the kip1071merge/streams_group branch from b9894e7 to ad4f4f3 Compare February 3, 2025 13:19

lucasbru requested review from bbejeck, cadonna and Copilot February 3, 2025 13:20

Copilot AI reviewed Feb 3, 2025

View reviewed changes

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

lucasbru force-pushed the kip1071merge/streams_group branch from ad4f4f3 to ab70816 Compare February 3, 2025 14:30

cadonna added streams KIP-1071 PRs related to KIP-1071 core Kafka Broker labels Feb 3, 2025

lucasbru force-pushed the kip1071merge/streams_group branch from ab70816 to 8866781 Compare February 4, 2025 08:58

cadonna reviewed Feb 4, 2025

View reviewed changes

cadonna reviewed Feb 5, 2025

View reviewed changes

bbejeck reviewed Feb 6, 2025

View reviewed changes

lucasbru commented Feb 7, 2025

View reviewed changes

lucasbru added 2 commits February 7, 2025 17:28

KAFKA-18323: Add StreamsGroup class

a6dccda

updates

98bcef4

lucasbru force-pushed the kip1071merge/streams_group branch from 77921b0 to 98bcef4 Compare February 7, 2025 16:30

cadonna reviewed Feb 7, 2025

View reviewed changes

comments

b4287f3

lucasbru commented Feb 10, 2025

View reviewed changes

cadonna approved these changes Feb 11, 2025

View reviewed changes

lucasbru merged commit c70b7c4 into apache:trunk Feb 12, 2025
7 of 9 checks passed

	* @return An immutable Map containing all the target assignment keyed by member id.
	* @return An immutable map containing all the target assignment keyed by member ID.

	* @param memberId The member id.
	* @param memberId The member ID.

	* Returns the current processId of a task or null if the task does not have one.
	* Returns the current process ID of a task or null if the task does not have one.

KAFKA-18323: Add StreamsGroup class #18729

KAFKA-18323: Add StreamsGroup class #18729

Uh oh!

Conversation

lucasbru commented Jan 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

lucasbru commented Feb 3, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lucasbru commented Feb 4, 2025

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lucasbru commented Jan 28, 2025 •

edited

Loading