[jira] [Created] (KAFKA-20887) IQ metadata is incomplete during "streams" protocol rebalances, causing incorrect IQ results

["Matthias J. Sax (Jira)" <ji...@apache.org>]

Matthias J. Sax created KAFKA-20887:
---------------------------------------

             Summary: IQ metadata is incomplete during "streams" protocol rebalances, causing incorrect IQ results
                 Key: KAFKA-20887
                 URL: https://issues.apache.org/jira/browse/KAFKA-20887
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 4.1.0
            Reporter: Matthias J. Sax


Setting "Affects Version" to 4.1.0, even if "streams" is only "early access" in 4.1.0, and should not be used in production.

Using the new "streams" groups, KafkaStreams#queryMetadataForKey() can compute the wrong partition for a key *_during a rebalance_* and return the wrong host, so IQ may return a "not found" for a keys that exist. Note, this is not an exception, but an incorrectly empty query result. This problem is affecting all IQ queries across IQv1 and IQv2, leading to incorrect query results that the user cannot detect.

*Root cause*

The client infers each source topic's partition count from the reported "host endpoint information":

- StreamThread → getTopicPartitionInfo(activeHostInfoMap) → StreamsMetadataState#onChange → partitionsByTopic
- SourceTopicsInfo#maxPartitions = partitionsByTopic.get(topic).size()
- keyQueryMetadataForKey passes maxPartitions to StreamPartitioner#partitions as the modulo

This assumes every task is present in the reported "host endpoint" metadata at all times. In the "classic" protocol this invariant is guaranteed — partitionsByHost is built from the assignor's complete target (StreamsPartitionAssignor Step Three), before cooperative revocation withholds anything.

However, it does not hold for the "streams" protocol. EndpointToPartitionsManager builds each EndpointToPartitions entry from that member's current assignedTasks(), and during reconciliation a migrating task sits in the old owner's tasksPendingRevocation — in no member's assignedTasks(), so in no entry at all.

It was a deliberate improvement via KIP-1071 (and the impact on IQ was missed): if a task is not assigned, IQ metadata should not claim it is, resulting in spurious user facing exception ("task migrated to another instance"), requiring user-code retries.

*Impact*

An in-flight task on a 4-partition topic drops the count to 3, so keys hash % 3 instead of % 4. Roughly 75% of keys resolve to a wrong partition, and each maps to a host that legitimately owns that partition — live, store open — which may answers "key not found". No exception, no retry signal. Only the migrating partition's own keys should have been unavailable.

*Proposal*

Have the broker send the missing information in StreamsGroupHeartbeatResponse instead of the client inferring it — e.g. a field "unassigned partitions" or similar as part of `EndpointInformationEpoch` field, so the client's existing derivation keeps working with minimal client-side change.

Deriving the topic/partition metadata from the "host metadata" map was a deliberate choice to avoid an extra RPC from the client to the broker to fetch topic metadata, so having the client query the cluster is not the desired direction.

*Note:* The new fields must also consider tasks assigned to a client with `application.server=null`. – These tasks are filtered out from the "host endpoint" map, too.

The `application.server` issue is actually a long standing bug also affecting "classic". It's a corner case, of a mis-configured application. Applications should be configured such that all KafkaStreams clients set `application.server` or none; a mixed configuration is a wrong config – nevertheless, it should not break IQ's correctness, but rather only result in that the client with the missing config cannot be queried, ie, some tasks would be unavailable for IQ.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)