[
https://issues.apache.org/jira/browse/KAFKA-20782?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Uladzislau Blok updated KAFKA-20782:
------------------------------------
Description:
[https://github.com/apache/kafka/pull/22778]
The broker side: Since with {{group.protocol=streams}} every stream thread is
an individual group member, all threads of the same instance report the same
{{application.server}} endpoint — so the heartbeat response contains multiple
entries with an identical host:port.
On the client side, {{convertHostInfoMap()}} then puts those entries into a
{{HashMap}} keyed by host:port, so the last entry wins and the partitions of
all other threads of that instance are dropped. The map flows via
{{StreamsRebalanceData}} into {{StreamsMetadataState#onChange()}} (see
{{{}StreamThread{}}}), which backs {{metadataForAllStreamsClients()}} /
{{{}queryMetadataForKey(){}}}. So with n threads, IQ metadata reflects only one
member's share of the instance's partitions — exactly what you observed.
Regarding the classic protocol: it is indeed not affected, because there the
leader aggregates all tasks of a process (across all its threads) in
{{StreamsPartitionAssignor#populatePartitionsByHostMaps()}} BEFORE keying by
host, so no duplicate host entries ever exist.
was:https://github.com/apache/kafka/pull/22778
> IQ returns 1/N metadata when Streams protocol and N threads
> -----------------------------------------------------------
>
> Key: KAFKA-20782
> URL: https://issues.apache.org/jira/browse/KAFKA-20782
> Project: Kafka
> Issue Type: Bug
> Components: group-coordinator, streams
> Affects Versions: 4.2.1, 4.3.1
> Reporter: Uladzislau Blok
> Assignee: Uladzislau Blok
> Priority: Major
>
> [https://github.com/apache/kafka/pull/22778]
> The broker side: Since with {{group.protocol=streams}} every stream thread is
> an individual group member, all threads of the same instance report the same
> {{application.server}} endpoint — so the heartbeat response contains multiple
> entries with an identical host:port.
> On the client side, {{convertHostInfoMap()}} then puts those entries into a
> {{HashMap}} keyed by host:port, so the last entry wins and the partitions of
> all other threads of that instance are dropped. The map flows via
> {{StreamsRebalanceData}} into {{StreamsMetadataState#onChange()}} (see
> {{{}StreamThread{}}}), which backs {{metadataForAllStreamsClients()}} /
> {{{}queryMetadataForKey(){}}}. So with n threads, IQ metadata reflects only
> one member's share of the instance's partitions — exactly what you observed.
> Regarding the classic protocol: it is indeed not affected, because there the
> leader aggregates all tasks of a process (across all its threads) in
> {{StreamsPartitionAssignor#populatePartitionsByHostMaps()}} BEFORE keying by
> host, so no duplicate host entries ever exist.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
