sanghyeok An created KAFKA-21188:
------------------------------------
Summary: 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
Reporter: sanghyeok An
Assignee: sanghyeok An
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)
