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, Matthias J. Sax matthias@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)?

[mjsax]

I think the fix makes sense.

For the test, we should change the test to 2 thread, and also change the number of input topic partitions to 4 (to get one task per thread, as we have two instances).

This requires to update all kind of assertions as we get different count... Can you update the test accordingly to make it pass for both "classic" and "streams"?