KAFKA-20885: Fix IQ metadata for source topics with fewer partitions - #23050
Merged
Conversation
mjsax
force-pushed
the
kafka-20885-fix-streams-iq-metadata
branch
from
August 3, 2026 00:28
cc75170 to
62df680
Compare
EndpointToPartitionsManager clamped the partitions it reports for a member by comparing the topic's partition count against the *number* of tasks the member owns, and then keeping the lowest task IDs. That is only correct when a single member owns the whole subtopology. A subtopology has as many tasks as its source topic with the most partitions, so a smaller source topic has no partition for the higher task IDs. When a member owns a "non-prefix" task set, the size-based check does not fire and the task IDs are reported verbatim as partitions of that smaller topic, advertising partitions that do not exist. Filter the task IDs by value, and omit a source topic entirely when none of the member's tasks map to one of its partitions. This mirrors PartitionGrouper#partitionGroups, which computes the same mapping for the classic protocol.
mjsax
force-pushed
the
kafka-20885-fix-streams-iq-metadata
branch
from
August 3, 2026 00:38
62df680 to
a10822a
Compare
bbejeck
reviewed
Aug 3, 2026
| // fewer partitions therefore has no partition for the higher task IDs, mirroring the client-side | ||
| // grouping in PartitionGrouper#partitionGroups. | ||
| List<Integer> partitions = taskSet.stream() | ||
| .filter(taskId -> taskId < numPartitionsForTopic) |
Member
There was a problem hiding this comment.
nit: maybe change taskId -> partitionId
mjsax
commented
Aug 3, 2026
bbejeck
approved these changes
Aug 3, 2026
mjsax
added a commit
that referenced
this pull request
Aug 4, 2026
…23050) EndpointToPartitionsManager clamped the partitions it reports for a member by comparing the topic's partition count against the *number* of tasks the member owns, and then keeping the lowest task IDs. That is only correct when a single member owns the whole subtopology. A subtopology has as many tasks as its source topic with the most partitions, so a smaller source topic has no partition for the higher task IDs. When a member owns a "non-prefix" task set, the size-based check does not fire and the task IDs are reported verbatim as partitions of that smaller topic, advertising partitions that do not exist. Filter the task IDs by value, and omit a source topic entirely when none of the member's tasks map to one of its partitions. This mirrors PartitionGrouper#partitionGroups, which computes the same mapping for the classic protocol. Reviewers: Bill Bejeck <bbejeck@apache.org>
mjsax
added a commit
that referenced
this pull request
Aug 4, 2026
…23050) EndpointToPartitionsManager clamped the partitions it reports for a member by comparing the topic's partition count against the *number* of tasks the member owns, and then keeping the lowest task IDs. That is only correct when a single member owns the whole subtopology. A subtopology has as many tasks as its source topic with the most partitions, so a smaller source topic has no partition for the higher task IDs. When a member owns a "non-prefix" task set, the size-based check does not fire and the task IDs are reported verbatim as partitions of that smaller topic, advertising partitions that do not exist. Filter the task IDs by value, and omit a source topic entirely when none of the member's tasks map to one of its partitions. This mirrors PartitionGrouper#partitionGroups, which computes the same mapping for the classic protocol. Reviewers: Bill Bejeck <bbejeck@apache.org>
Member
Author
|
Merged to |
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
EndpointToPartitionsManager clamped the partitions it reports for a
member by comparing the topic's partition count against the number of
tasks the member owns, and then keeping the lowest task IDs. That is
only correct when a single member owns the whole subtopology.
A subtopology has as many tasks as its source topic with the most
partitions, so a smaller source topic has no partition for the higher
task IDs. When a member owns a "non-prefix" task set, the size-based
check does not fire and the task IDs are reported verbatim as partitions
of that smaller topic, advertising partitions that do not exist.
Filter the task IDs by value, and omit a source topic entirely when none
of the member's tasks map to one of its partitions. This mirrors
PartitionGrouper#partitionGroups, which computes the same mapping for
the classic protocol.
Reviewers: Bill Bejeck bbejeck@apache.org