KAFKA-18325: Add TargetAssignmentBuilder #18676

lucasbru · 2025-01-23T09:00:39Z

A class to build a new target assignment based on the provided parameters. As a result, it yields the records that must be persisted to the log and the new member assignments as a map.

Compared to the feature branch, I extended the unit tests (testing also standby and warm-up task logic) and adopted simplifications due to the TasksTuple class.

Committer Checklist (excluded from commit message)

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

lucasbru · 2025-01-24T09:49:02Z

Rebased on latest version of streams coordinator record helpers

lucasbru · 2025-01-28T16:11:43Z

Rebased on latest trunk (and minor test fix)

lucasbru · 2025-01-30T08:59:14Z

Rebased on latest trunk

bbejeck

Thanks for the PR @lucasbru overall this looks great, I have a few comments.

bbejeck · 2025-01-30T19:14:03Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+    public TargetAssignmentBuilder removeMember(
+        String memberId
+    ) {
+        return addOrUpdateMember(memberId, null);


Why not just outright remove the member at this point? I get that in the build() method will filter it out, but does that have any advantage other than a removal here.

I believe you cannot remove the member right away, because due to the builder pattern there is not guarantee that the call to withMembers() occurs before the call to removeMembers().

bbejeck · 2025-01-30T20:17:02Z

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

+     * @return The number of partitions corresponding to the given subtopology ID, or -1 if the subtopology ID does not exist.
+     */
+    @Override
+    public int numTasks(String subtopologyId) {


Given the javadoc says number of partitions should the method be named numPartitions? It's a minor point, so I'll leave it to you to decide.

It seems we go back and forth with the question when to use partitions and when to use tasks. I believe the confusion comes from the fact that the max number of input partitions per topic of a subtopology determines the number of tasks.
I propose to call this method maxNumInputPartitions(). My rationale is that a topology does not have a notion of tasks. The task assignor uses maxNumInputPartitions() to know how many tasks to assign.
I am sorry, in case I am contradicting myself regarding an earlier comment.

Works for me.

bbejeck · 2025-01-30T20:21:43Z

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

+    }
+
+    @Override
+    public boolean isStateful(String subtopologyId) {


For the methods that use getXOrFail would it be worth adding a brief javadoc about the possibility of an IllegalStateException being thrown?

bbejeck · 2025-01-30T20:25:20Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+                new TopologyMetadata(subscriptionMetadata, topology)
+            );
+        } else {
+            newGroupAssignment = new GroupAssignment(


This branch isn't covered by the unit test.

bbejeck · 2025-01-30T20:27:32Z

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

+    private ConfiguredSubtopology getSubtopologyOrFail(String subtopologyId) {
+        final Map<String, ConfiguredSubtopology> subtopologies = getSubtopologiesOrFail();
+        if (!subtopologies.containsKey(subtopologyId)) {
+            throw new IllegalStateException(String.format("Topology does not contain subtopology %s", subtopologyId));


This branch isn't tested. Same for below in getSubtopologiesOrFail()

cadonna

Thanks for the PR, @lucasbru !

I made a pass over the production code.

cadonna · 2025-01-31T08:46:05Z

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

+        Objects.requireNonNull(topicMetadata);
+        Objects.requireNonNull(topology);


I assume the topicMetadata is mutable, because topic configs may change at any time. Since topic metadata can change at any time also topology can change at any time, right?

It's not mutable, since TopicMetadata is short-lived and passed only once to the assingor. I added unmodifiableMap, if that is what you were going for.

Shouldn't topology then also be immutable?

Made it an immutable collection of subtopologies.

cadonna · 2025-01-31T08:48:12Z

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

+ *
+ * @param topicMetadata The topic Ids mapped to their corresponding {@link TopicMetadata} object, which contains topic and partition
+ *                      metadata.
+ */


The java doc for topology is missing.

cadonna · 2025-01-31T09:00:40Z

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

+     * @return The number of partitions corresponding to the given subtopology ID, or -1 if the subtopology ID does not exist.
+     */
+    @Override
+    public int numTasks(String subtopologyId) {


It seems we go back and forth with the question when to use partitions and when to use tasks. I believe the confusion comes from the fact that the max number of input partitions per topic of a subtopology determines the number of tasks.
I propose to call this method maxNumInputPartitions(). My rationale is that a topology does not have a notion of tasks. The task assignor uses maxNumInputPartitions() to know how many tasks to assign.
I am sorry, in case I am contradicting myself regarding an earlier comment.

cadonna · 2025-01-31T10:43:08Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+    public static class TargetAssignmentResult {
+
+        /**
+         * The records that must be applied to the __streams_offsets topics to persist the new target assignment.


You probably mean __consumer_offsets? Replace error?

cadonna · 2025-01-31T10:47:04Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+
+/**
+ * Build a new Target TasksTuple based on the provided parameters. As a result, it yields the records that must be persisted to the log and
+ * the new member assignments as a map.


Could you please explicitly add that the task assignor is called that computes the new assigment?
Something like
"Build the new target member assignments based on the provided parameters by calling the task assignor. As a result, it yields the records that must be persisted to the log and the new member assignments as a map from member ID to tasks tuple."

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

cadonna · 2025-01-31T10:59:22Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+    /**
+     * Adds the subscription metadata to use.
+     *
+     * @param partitionMetadata The subscription metadata.


"subscription" or "partition"? This is a bit confusing.
Maybe topicMetadata?

cadonna · 2025-01-31T11:09:02Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+    public TargetAssignmentBuilder removeMember(
+        String memberId
+    ) {
+        return addOrUpdateMember(memberId, null);


I believe you cannot remove the member right away, because due to the builder pattern there is not guarantee that the call to withMembers() occurs before the call to removeMembers().

cadonna · 2025-01-31T11:10:46Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+            if (updatedMemberOrNull == null) {
+                memberSpecs.remove(memberId);
+            } else {
+                org.apache.kafka.coordinator.group.streams.TasksTuple assignment = targetAssignment.getOrDefault(memberId,


I believe, we do not need the full-qualified name of the TasksTuple here, right? TasksTuple is specific to the streams package.

cadonna · 2025-01-31T11:40:38Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

+    /**
+     * The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
+     */
+    public static class TargetAssignmentResult {


Could this be implemented as a record class?

lucasbru · 2025-01-31T16:03:53Z

@cadonna @bbejeck Thanks for the comments! All comments addressed, ready for re-review.

cadonna · 2025-01-31T16:08:16Z

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

+        if (subtopologies.isEmpty()) {
+            throw new IllegalStateException("Topology is not configured");
+        }


Is there a reason, you cannot make this check at construction time?

After some offline discussion, we agreed to unpack the optional inside TopologyMetadata to avoid the isPresent check altogether.

lucasbru · 2025-02-03T14:20:35Z

...inator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

@@ -265,12 +265,15 @@ public TargetAssignmentResult build() throws TaskAssignorException {
        // Compute the assignment.
        GroupAssignment newGroupAssignment;
        if (topology.isReady()) {
+            if (topology.subtopologies().isEmpty()) {
+                throw new IllegalStateException("Subtopologies must be present if topology is ready.");


Central check. This is implied by isReady, but I added the IllegalStateException just in case some future updates breaks the internal invariants of ConfiguredTopology (and to silence IDE warnings).

lucasbru · 2025-02-03T14:25:19Z

@cadonna Ready for re-review.

cadonna

Thanks for the updates, @lucasbru !

LGTM!

lucasbru force-pushed the kip1071merge/target_assignment_builder branch from 727f1a1 to 177ca42 Compare January 24, 2025 09:48

lucasbru force-pushed the kip1071merge/target_assignment_builder branch from 177ca42 to 24eea87 Compare January 28, 2025 16:11

lucasbru force-pushed the kip1071merge/target_assignment_builder branch from 24eea87 to e211eb6 Compare January 30, 2025 08:58

bbejeck reviewed Jan 30, 2025

View reviewed changes

cadonna reviewed Jan 31, 2025

View reviewed changes

lucasbru added 3 commits February 3, 2025 14:57

KAFKA-18325: Add TargetAssignmentBuilder

1dd6fa4

fixes

2019d58

fixes

ea8723f

lucasbru force-pushed the kip1071merge/target_assignment_builder branch from debeeb3 to ea8723f Compare February 3, 2025 14:13

lucasbru commented Feb 3, 2025

View reviewed changes

cadonna approved these changes Feb 3, 2025

View reviewed changes

lucasbru merged commit 4ca24a7 into apache:trunk Feb 3, 2025
9 checks passed

lucasbru added core Kafka Broker KIP-1071 PRs related to KIP-1071 streams labels Feb 5, 2025

		Objects.requireNonNull(topicMetadata);
		Objects.requireNonNull(topology);

KAFKA-18325: Add TargetAssignmentBuilder #18676

KAFKA-18325: Add TargetAssignmentBuilder #18676

Conversation

lucasbru commented Jan 23, 2025 • edited Loading

Committer Checklist (excluded from commit message)

lucasbru commented Jan 24, 2025

lucasbru commented Jan 28, 2025

lucasbru commented Jan 30, 2025

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

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

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 commented Jan 31, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lucasbru commented Feb 3, 2025

cadonna left a comment

Choose a reason for hiding this comment

lucasbru commented Jan 23, 2025 •

edited

Loading