[
https://issues.apache.org/jira/browse/KAFKA-20887?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18107378#comment-18107378
]
md tanwir edited comment on KAFKA-20887 at 8/29/26 10:40 AM:
-------------------------------------------------------------
https://github.com/apache/kafka/pull/23255
was (Author: JIRAUSER314232):
https://github.com/apache/kafka/pull/21674
> 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
> Assignee: md tanwir
> Priority: Critical
> Labels: needs-kip
>
> 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. We can fix this "for free" for the "streams"
> case when we do this ticket. Cf the linked ticket for "classic" which must
> get its own fix independently. Of course, the client side code (which is
> shared across both protocols) must be setup such that is works for both
> "streams" and "classic" case. Thus, as long as it's not fixed for "classic",
> we need to still ensure that the code does not crash for "classic" if we fix
> it for "streams".
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
