KAFKA-20782: IQ bug under streams protocol

[UladzislauBlok]

Looks like IQ is broken when:

• New streams protocol is used
• number of threads is more than 1

Streams result:

Gradle Test Run :streams:integration-tests:test > Gradle Test Executor 7 > IQv2EndpointToPartitionsIntegrationTest > shouldGetCorrectHostPartitionInformation(String, boolean, int, String) > STREAMS protocol No standby FAILED
    org.opentest4j.AssertionFailedError: Condition not met within timeout 60000. Kafka Streams one didn't get 4 tasks ==> expected: <true> but was: <false>

Classic result (still fail but overall number of tasks are correct, see error message):

Gradle Test Run :streams:integration-tests:test > Gradle Test Executor 11 > IQv2EndpointToPartitionsIntegrationTest > shouldGetCorrectHostPartitionInformation(String, boolean, int, String) > CLASSIC protocol No standby FAILED
    org.opentest4j.AssertionFailedError: Condition not met within timeout 15000. KafkaStreams one never released active tasks and received standby task ==> expected: <true> but was: <false>

Root cause:
partitionsByHost.put(new StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port()), endpointPartitions);

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java

Line 830 in 8c0ca4a

private static Map<StreamsRebalanceData.HostInfo, StreamsRebalanceData.EndpointPartitions> convertHostInfoMap(

As I understand under classic protocol entire host is considered as one member, so it's okay to override metadata view by host + port
The new protocol conciser each stream-thread as independent member (correct me if I'm wrong), so we need to union all threads metadata instead of overriding it (map#merge ??)
Now effect is that metadataForAllStreamsClients() returns only 1 / n threads metadata
(to be confirmed)

Reviewers: Alieh Saeedi asaeedi@confluent.io

[UladzislauBlok]

@lucasbru @mjsax
Hey guys,
I think we have a regression in IQ with new protocol enabled. When setting number of threads is more than 1 we see only last updated thread metadata instead of merged view

[aliehsaeedii]

Thanks @UladzislauBlok for informing us. I'm checking the code...

[aliehsaeedii]

Thanks @UladzislauBlok — I checked the code and can confirm your root cause analysis.

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. The classic failure you saw is unrelated to the metadata: the test asserts on a single thread via metadataForLocalThreads().iterator().next() and expects it to hold 2 active tasks, which cannot hold once the instance has 3 threads — those assertions would need to aggregate across threads for a multi-threaded setup.

FWIW, the bug has existed since IQ support for KIP-1071 landed on trunk (KAFKA-19135), so all releases from 4.1.0 onwards are affected. Are you interested in pushing the fix?

[UladzislauBlok]

@aliehsaeedii
Would love to. While fix doesn't seem to be too complicated (not too confident about that, I will need to dig deeper), I'm curious if we need to prepare patch releases: 4.2.2 and 4.3.2 (and ofc merge to trunk and include in 4.4.0)?