KAFKA-18518: Add processor to handle rebalance events #18527

cadonna · 2025-01-14T14:48:33Z

This commit adds a processor named
StreamsRebalanceEventsProcessor that handles the rebalance events sent from the background thread of the async consumer to the stream thread when an task
assignment changes. It also adds the corresponding rebalance events.

Additionally, this commit adds StreamsRebalanceData that maintains the data that is exchanges for the Streams rebalance protocol.

All of these are used by the Streams heartbeat request manager and the Streams membership manager that will be added in a future commit.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

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

cadonna · 2025-01-14T14:48:55Z

Call for review: @aliehsaeedii

lucasbru

Thanks for the PR @cadonna ! I think StreamsRebalanceData need some more work, rest is looking fine

lucasbru · 2025-01-20T13:49:51Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+
+        public static final Assignment EMPTY = new Assignment();
+
+        public final Set<TaskId> activeTasks = new HashSet<>();


The collections should be unmodifiable, and probably private with getters?

Right now, I can do Assignment.EMPTY.activeTasks.add(new TaskID("a",1)) and wreak havoc.

If you want to make the class mutable, you cannot define EMPTY as a static variable.

lucasbru · 2025-01-20T13:50:34Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+
+        public final Set<TaskId> warmupTasks = new HashSet<>();
+
+        public Assignment() {


If you are calling this constructor, you should just be using EMPTY. It does not make sense to have both.

lucasbru · 2025-01-20T13:52:27Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+            return subtopologyId;
+        }
+
+        public TaskId(final String subtopologyId, final int partitionId) {


nit: first constructor, then getters

lucasbru · 2025-01-20T13:54:49Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+
+        @Override
+        public int compareTo(TaskId taskId) {
+            if (subtopologyId.equals(taskId.subtopologyId)) {


In Java 8, it woul be more idiomatic to use soemthing like:

Comparator.comparing(TaskId::subtopologyId) .thenComparingInt(TaskId::partitionId);

lucasbru · 2025-01-20T13:56:27Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+
+    public static class Subtopology {
+
+        public final Set<String> sourceTopics;


We may want to make these collections unmodifiable, non-null, private and have getters.

lucasbru · 2025-01-20T13:58:49Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+    private final AtomicReference<Assignment> reconciledAssignment = new AtomicReference<>(Assignment.EMPTY);
+
+    public StreamsRebalanceData(Map<String, Subtopology> subtopologies) {
+        this.subtopologies = subtopologies;


non-null, unmodifiable?

lucasbru · 2025-01-20T14:00:48Z

...c/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessor.java

+     */
+    public StreamsRebalanceEventsProcessor(StreamsRebalanceData streamsRebalanceData,
+                                           StreamsGroupRebalanceCallbacks rebalanceCallbacks) {
+        this.streamsRebalanceData = streamsRebalanceData;


lucasbru · 2025-01-20T14:05:04Z

...c/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessor.java

+
+    private StreamsOnTasksRevokedCallbackCompletedEvent invokeOnTasksRevokedCallback(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke,
+                                                                                     final CompletableFuture<Void> future) {
+        final Optional<Exception> exceptionFromCallback = rebalanceCallbacks.onTasksRevoked(activeTasksToRevoke);


invokeOnTasksRevokedCallback has a different implementation than invokeOnTasksAssignedCallback and invokeOnAllTasksLostCallback. It's probably easier to read if all three use either the .map stile or the isPresent style. I don't mind which one, isPresent is probably a little easier to follow.

In this case I just applied IDEA's recommendation to transform it to functional style. However, I agree with you regarding readability.

lucasbru · 2025-01-20T14:05:37Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java

@@ -27,7 +27,9 @@
 public abstract class BackgroundEvent {

    public enum Type {
-        ERROR, CONSUMER_REBALANCE_LISTENER_CALLBACK_NEEDED, SHARE_ACKNOWLEDGEMENT_COMMIT_CALLBACK
+        ERROR, CONSUMER_REBALANCE_LISTENER_CALLBACK_NEEDED, SHARE_ACKNOWLEDGEMENT_COMMIT_CALLBACK,


nit: one per line will simplify conflict resolution

lucasbru · 2025-01-20T14:08:57Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java

+/**
+ * This class holds the data that is needed to participate in the Streams rebalance protocol.
+ */
+public class StreamsRebalanceData {


This class and its inner classes need unit tests.

This commit adds a processor named StreamsRebalanceEventsProcessor that handles the rebalance events sent from the background thread of the async consumer to the stream thread when an task assignment changes. It also adds the corresponding rebalance events. Additionally, this commit adds StreamsRebalanceData that maintains the data that is exchanges for the Streams rebalance protocol. All of these are used by the Streams heartbeat request manager and the Streams membership manager that will be added in a future commit.

cadonna · 2025-01-21T14:55:39Z

@lucasbru Thank you for your review!
Frankly, I do not know why I missed out on so many things in the Streams rebalance data. 🤷

Please re-review.

lucasbru

LGTM, thanks for the updates!

This commit adds a processor named StreamsRebalanceEventsProcessor that handles the rebalance events sent from the background thread of the async consumer to the stream thread when an task assignment changes. It also adds the corresponding rebalance events. Additionally, this commit adds StreamsRebalanceData that maintains the data that is exchanges for the Streams rebalance protocol. All of these are used by the Streams heartbeat request manager and the Streams membership manager that will be added in a future commit. Reviewer: Lucas Brutschy <lbrutschy@confluent.io>

cadonna added streams consumer KIP-1071 PRs related to KIP-1071 labels Jan 14, 2025

cadonna requested review from bbejeck, lucasbru and mjsax January 14, 2025 14:48

github-actions bot added triage PRs from the community clients labels Jan 14, 2025

cadonna mentioned this pull request Jan 15, 2025

KAFKA-18538: Add Streams membership manager #18551

Open

3 tasks

cadonna removed the triage PRs from the community label Jan 15, 2025

cadonna force-pushed the streams_rebalance_interface branch from d34ad1a to fc91305 Compare January 15, 2025 21:24

lucasbru reviewed Jan 20, 2025

View reviewed changes

cadonna added 12 commits January 21, 2025 15:54

Add license headers

dc845c0

Apply spotless

21c4bee

Add javadocs

29dd4e0

Improve TaskId

e7ac06e

Improve Assignment

c653ddd

Add null checks to TaskId

88d016c

Improve subtopology

cc8e0b5

Improve topic info

de35989

Improve streams rebalance data

31f170d

Put types of background event on separate lines

eadecd3

Harmonize implementations of callbacks invocations

a319b91

cadonna force-pushed the streams_rebalance_interface branch from fc91305 to a319b91 Compare January 21, 2025 14:55

fix checkstyle issues

2015d24

lucasbru approved these changes Jan 21, 2025

View reviewed changes

cadonna merged commit 239708f into apache:trunk Jan 22, 2025
9 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18518: Add processor to handle rebalance events #18527

KAFKA-18518: Add processor to handle rebalance events #18527

cadonna commented Jan 14, 2025

cadonna commented Jan 14, 2025

lucasbru left a comment •

edited

Loading

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

cadonna Jan 21, 2025

lucasbru Jan 20, 2025

lucasbru Jan 20, 2025

cadonna commented Jan 21, 2025

lucasbru left a comment


		public static final Assignment EMPTY = new Assignment();

		public final Set<TaskId> activeTasks = new HashSet<>();


		public final Set<TaskId> warmupTasks = new HashSet<>();

		public Assignment() {


		public static class Subtopology {

		public final Set<String> sourceTopics;

KAFKA-18518: Add processor to handle rebalance events #18527

KAFKA-18518: Add processor to handle rebalance events #18527

Conversation

cadonna commented Jan 14, 2025

Committer Checklist (excluded from commit message)

cadonna commented Jan 14, 2025

lucasbru left a comment • edited Loading

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

lucasbru left a comment

Choose a reason for hiding this comment

lucasbru left a comment •

edited

Loading