[
https://issues.apache.org/jira/browse/KAFKA-21188?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18138017#comment-18138017
]
sanghyeok An edited comment on KAFKA-21188 at 11/4/26, 2:40 PM:
----------------------------------------------------------------
[~lucasbru]
I’m working on a draft implementation to explore whether this issue can be
addressed through client-side changes alone.
During that work, I found a case where client-side validation appears
insufficient to prevent task input partitions from being omitted due to stale
metadata, and I’d appreciate your thoughts!
For example, suppose the actual partition counts are {{{}A=1, B=3,
{color:#de350b}*C=5*{color}{}}}, while the client’s cached counts are {{{}A=1,
B=3, {color:#de350b}*C=1*{color}{}}}. If the client is assigned task 2, its
input partitions should be {color:#de350b}*{B-2, C-2}*{color}, but the client
would construct only {color:#de350b}*{B-2}*{color}. Since all partition counts
are available and the input partition set is non-empty, this passes both the
null checks and the empty-input checks.
The task assignment and cached metadata alone do not allow the client to
distinguish whether C actually has one partition or whether its cached count is
stale. Even requesting another metadata update does not guarantee freshness,
because the broker handling the Metadata request may be behind the coordinator.
Therefore, addressing these omissions in this issue does not appear possible
through changes that rely solely on the current task assignment information and
the client’s metadata cache. Checking for empty input can prevent the
corresponding exception, but it cannot detect a partially incomplete input
partition set.
{{ConsumerGroupHeartbeat}} provides assignments as topic IDs and partition
numbers, and the client reports its currently owned partitions in the same
terms. It therefore does not need to reconstruct the assigned partition set
from per-topic partition counts. In contrast, {{ActiveTasks}} in
{{StreamsGroupHeartbeat}} provides subtopology IDs and task partition numbers.
That information alone is insufficient to determine the complete set of input
topic partitions for each task.
To fully address this issue, I believe the client needs either a way to verify
that its metadata is sufficiently up to date for the assignment, or additional
information that explicitly identifies each task’s input partitions. It appears
that changes to both the client and the broker are needed to resolve this issue!
What do you think? Please share your thoughts, when you get a chance!
was (Author: JIRAUSER303328):
[~lucasbru]
I’m working on a draft implementation to explore whether this issue can be
addressed through client-side changes alone.
During that work, I found a case where client-side validation appears
insufficient to prevent task input partitions from being omitted due to stale
metadata, and I’d appreciate your thoughts!
For example, suppose the actual partition counts are {{{}A=1, B=3, C=5{}}},
while the client’s cached counts are {{{}A=1, B=3, C=1{}}}. If the client is
assigned task 2, its input partitions should be {{{}{B-2, C-2}{}}}, but the
client would construct only {{{}{B-2}{}}}. Since all partition counts are
available and the input partition set is non-empty, this passes both the null
checks and the empty-input checks.
The task assignment and cached metadata alone do not allow the client to
distinguish whether C actually has one partition or whether its cached count is
stale. Even requesting another metadata update does not guarantee freshness,
because the broker handling the Metadata request may be behind the coordinator.
Therefore, addressing these omissions in KAFKA-21188 does not appear possible
through changes that rely solely on the current task assignment information and
the client’s metadata cache. Checking for empty input can prevent the
corresponding exception, but it cannot detect a partially incomplete input
partition set.
{{ConsumerGroupHeartbeat}} provides assignments as topic IDs and partition
numbers, and the client reports its currently owned partitions in the same
terms. It therefore does not need to reconstruct the assigned partition set
from per-topic partition counts. In contrast, {{ActiveTasks}} in
{{StreamsGroupHeartbeat}} provides subtopology IDs and task partition numbers.
That information alone is insufficient to determine the complete set of input
topic partitions for each task.
To fully address this issue, I believe the client needs either a way to verify
that its metadata is sufficiently up to date for the assignment, or additional
information that explicitly identifies each task’s input partitions. It appears
that changes to both the client and the broker are needed to resolve this issue!
What do you think? Please share your thoughts, when you get a chance!
> Merging topics with different partition counts leaves some records
> unprocessed under the Streams Rebalance Protocol
> -------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21188
> URL: https://issues.apache.org/jira/browse/KAFKA-21188
> Project: Kafka
> Issue Type: Bug
> Components: clients, streams
> Reporter: sanghyeok An
> Assignee: sanghyeok An
> Priority: Major
>
> With {*}group.protocol=streams{*}, merging two topics with different
> partition counts includes nonexistent partitions in the tasks’ inputs,
> leaving some records unprocessed.
> *Steps to Reproduce*
> 1. Create topic A with 1 partition({*}0{*}) and topic B with 3
> partitions({*}0, 1, 2{*}).
> 2. Run the following topology with {*}group.protocol=streams{*}:
> {code:java}
> KStream<String, String> streamA = builder.stream("A");
> KStream<String, String> streamB = builder.stream("B");
> streamA.merge(streamB).to("output");
> {code}
> 3. Produce records to {*}A-0{*}, {*}B-0{*}, {*}B-1{*}, and {*}B-2{*}.
> * *Expected:* All input records appear in `output`.
> * *Actual:* Records from `B-1` and `B-2` do not appear in `output`, and the
> following messages repeatedly reference nonexistent partitions:
> {code:java}
> Requesting the log end offset for A-1 in order to compute lag
> Requesting the log end offset for A-2 in order to compute lag
> {code}
>
> *Cause*
> StreamsMembershipManager#topicPartitionsForActiveTasks and
> DefaultStreamsRebalanceListener#toTopicPartitions apply the task’s partition
> ID to every input topic without checking each topic’s partition count.
> As a result, a task that should include only *B-1* also includes {*}A-1{*}.
> The same issue occurs for partition ID 2.
> The classic protocol’s PartitionGrouper checks each topic’s partition count
> and includes only existing partitions. The new protocol’s conversion paths
> need equivalent handling.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
