KAFKA-19468: Ignore unsubscribed topics when computing share assignment #20101
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
DL1231
reviewed
Jul 4, 2025
Comment on lines
2699
to
2705
| if (currentAssignedPartitions.equals(entry.getValue().partitions())) { | ||
| // The assigned and initialized partitions match, so assignment does not need to be recomputed. | ||
| continue; | ||
| } else { | ||
| // The assigned and initialized partitions do not match, so recompute the assignment. | ||
| return true; | ||
| } |
There was a problem hiding this comment.
What do you think of the implementation below?
Suggested change
| if (currentAssignedPartitions.equals(entry.getValue().partitions())) { | |
| // The assigned and initialized partitions match, so assignment does not need to be recomputed. | |
| continue; | |
| } else { | |
| // The assigned and initialized partitions do not match, so recompute the assignment. | |
| return true; | |
| } | |
| if (!currentAssignedPartitions.equals(entry.getValue().partitions())) { | |
| // The assigned and initialized partitions do not match, so recompute the assignment. | |
| return true; | |
| } |
There was a problem hiding this comment.
Code restructured slightly. I originally wrote it in a very simplistic style to make all of the cases really clear. But that was a bit verbose so tightened up now.
| @@ -2692,7 +2690,27 @@ private boolean initializedAssignmentPending(ShareGroup group) { | |||
| } | |||
| } | |||
|
|
|||
| return !initializedTps.equals(currentAssigned); | |||
| Set<String> subscribedTopicNames = group.subscribedTopicNames().keySet(); | |||
There was a problem hiding this comment.
Should we add integration tests for this scenario?
There was a problem hiding this comment.
Just added unit tests which verify that this method returns the expected value in all of the interesting cases. I suggest an integration test which verifies that the epoch does not increase continuously would be a good candidate for a future task.
DL1231
approved these changes
Jul 4, 2025
apoorvmittal10
approved these changes
Jul 4, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
When the group coordinator is processing a heartbeat from a share
consumer, it must decide whether the recompute the assignment. Part of
this decision hinges on whether the assigned partitions match the
partitions initialised by the share coordinator. However, when the set
of subscribed topics changes, there may be initialised partitions which
are not currently assigned. Topics which are not subscribed should be
omitted from the calculation about whether to recompute the assignment.
Co-authored-by: Sushant Mahajan smahajan@confluent.io
Reviewers: Lan Ding 53332773+DL1231@users.noreply.github.com, Sushant
Mahajan smahajan@confluent.io, Apoorv Mittal
apoorvmittal10@gmail.com