[
https://issues.apache.org/jira/browse/KAFKA-21188?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18135137#comment-18135137
]
Lucas Brutschy commented on KAFKA-21188:
----------------------------------------
Thanks for the analysis, this is a real gap. With the streams protocol, a
subtopology gets as many tasks as its source topic with the most partitions.
The client then pairs each task’s partition number with every source topic of
the subtopology, so tasks beyond the partition count of a smaller topic get a
partition that doesn’t exist. The consumer can never determine the lag of that
partition, and the whole task stays blocked, including the partitions it really
owns. Until a fix is released, setting {{max.task.idle.ms}} to -1 lets the
affected tasks process the partitions that exist.
We’d suggest to split the work into two steps.
The first is a client-only bug fix that we can also backport. Before
reconciling an assignment, the client checks that it knows the partition count
of every source topic of the assigned tasks, using its own metadata, and only
assigns partitions that exist. This is the same rule the classic
{{PartitionGrouper}} applies. If a partition count isn’t known yet, the client
requests a metadata update and reconciles those tasks once it arrives. It
follows how the consumer protocol’s membership manager resolves topic IDs:
tasks it can resolve are reconciled right away, and a partially resolved
assignment is acknowledged. Partition counts only ever increase, so this
matches the classic protocol’s behaviour. The one case it doesn’t cover is a
smaller source topic gaining partitions while its task stays assigned. The new
partition is only picked up with the next assignment change.
The second is a KIP that closes the remaining consistency gap. The client’s
metadata is refreshed independently of the heartbeat, so nothing guarantees it
describes the same topics the coordinator used for the assignment. The KIP
extends the streams group heartbeat response with the topic IDs and partition
counts of the topology’s source and repartition-source topics, taken from the
same metadata image as the assignment. The coordinator only sends them when
they change, using the group’s existing metadata hash as the version. Like the
consumer group protocol, topics are identified by ID only. The client resolves
the IDs to names through its metadata, with the same wait-and-resolve step as
the bug fix, but it takes the partition counts from the coordinator, so they’re
consistent with the assignment by construction. Because the topic metadata is
part of the target assignment, a change in partition counts alone triggers a
reconciliation, which also covers topic expansion. Topic IDs also let the
client detect a deleted and recreated topic, and they give us the foundation
for regex subscriptions, which the coordinator would resolve.
Wdyt? Both steps would share the same client structure, so the KIP mainly adds
a second source for the partition counts. Clients talking to coordinators
without the KIP fall back to the bug-fix behaviour. Would you like to pick some
of this work up? I’d suggest starting with the bug fix, since it already
resolved the main problem and can be backported, and then writing the KIP on
top of it to also support topic expansion.
> 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)
