[
https://issues.apache.org/jira/browse/KAFKA-21188?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136001#comment-18136001
]
sanghyeok An commented on KAFKA-21188:
--------------------------------------
[~lucasbru]
Thanks for the detailed explanation!
The overall approach sounds good to me.
I’ll start with the client-only bug fix, then work on a follow-up KIP for the
second step.
Considering KAFKA-20171 and our earlier discussion, I think the heartbeat
response should associate source topic IDs and their partition counts with each
subtopology. That would support both interpreting task assignments consistently
and adding broker-side regex support.
I also have an older PoC PR exploring topic ID support for
KAFKA-20171(https://github.com/apache/kafka/pull/21909).
It would need updating, but it may serve as a useful starting point for this
discussion.
I’d frame the KIP around the protocol changes and their semantics: what topic
metadata the heartbeat response carries, how changes trigger reconciliation,
and how the client behaves while some topic IDs cannot yet be resolved to
names. The partition counts would come from the coordinator, as you suggested.
The first step would address the main bug in KAFKA-21188 and allow backporting.
The KIP would then cover the remaining consistency and topic-expansion cases,
while providing the foundation for KAFKA-20171.
If that direction sounds reasonable, I’d be happy to take on the KIP as well.
What do you think?
> 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)
