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)
