[
https://issues.apache.org/jira/browse/KAFKA-21188?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134757#comment-18134757
]
sanghyeok An commented on KAFKA-21188:
--------------------------------------
Hi, [~lucasbru]!
I've been mentioning you quite often lately. I'm sorry to keep bothering you,
especially when I know you're busy. I really appreciate your time and patience.
There is an issue where a topology supported by classic groups does not work
with streams groups. I think this could be one of the blockers for online
migration (KAFKA-20172).
The classic Streams client already had partition information on the client
side, so it handled this when creating tasks. However, the Streams protocol
client does not have this information, so it includes nonexistent partitions in
a task’s inputs. Because it cannot determine the lag for those partitions, the
affected tasks cannot make progress.
To address this, I think we should include partition information for each topic
whenever an assignment is returned in the heartbeat response. Since this would
change the API schema, it might require an update to KIP-1071 or a new KIP.
I would appreciate your thoughts on this!
> 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)
