[
https://issues.apache.org/jira/browse/KAFKA-21188?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
sanghyeok An updated KAFKA-21188:
---------------------------------
Component/s: clients
streams
> 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
(v8.20.10#820010)
