[
https://issues.apache.org/jira/browse/KAFKA-20886?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-20886:
------------------------------------
Description:
Filing this a "minor" as we consider a "mixed configuration" as invalid, so
it's a user error triggering this issue. **
*Summary*
`application.server` is optional per instance. If some instances set it and
others don't, IQ key routing silently resolves to the wrong host — including
for keys hosted on the correctly configured instances.
*Root cause*
StreamsPartitionAssignor.populatePartitionsByHostMaps skips clients without an
endpoint:
if (hostInfo != null) \{ ... partitionsByHost.put(hostInfo, topicPartitions);
... }
Their partitions are therefore absent from partitionsByHost — which is also the
source of the partition count:
- getTopicPartitionInfo(partitionsByHost) → StreamsMetadataState#onChange →
partitionsByTopic
- SourceTopicsInfo#maxPartitions = partitionsByTopic.get(topic).size()
- keyQueryMetadataForKey passes maxPartitions to StreamPartitioner#partitions
as the modulo
An under-count changes the modulo, so keys hash to the wrong partition.
*Impact*
Permanent, not transient. If the unconfigured instances hold 1 of a topic's 4
partitions, keys are hashed % 3 instead of % 4 and ~75% resolve to the wrong
partition. The host returned for those legitimately owns that partition — live,
store open — so it answers "key not found". No exception, no retry signal.
If no instance sets `application.server,` partitionsByHost stays empty,
isInitialized() is false, and queryMetadataForKey returns
KeyQueryMetadata.NOT_AVAILABLE — correct. Only the mixed configuration is
broken. While a mixed configuration is an incorrect configuration, it should
still not break IQ's correctness, but should only imply that some tasks are
unavailable for IQ.
To fix this, we need to change our custom assignor metadata, to include
information about "not available tasks" instead of just dropping them on the
floor. This implies version bumps and upgrade/downgrade concerns. We might need
a KIP for this (need to evaluate, as we actually have "version probing" so
maybe we could also address w/o a KIP)
was:
*Summary*
`application.server` is optional per instance. If some instances set it and
others don't, IQ key routing silently resolves to the wrong host — including
for keys hosted on the correctly configured instances.
*Root cause*
StreamsPartitionAssignor.populatePartitionsByHostMaps skips clients without an
endpoint:
if (hostInfo != null) \{ ... partitionsByHost.put(hostInfo, topicPartitions);
... }
Their partitions are therefore absent from partitionsByHost — which is also the
source of the partition count:
- getTopicPartitionInfo(partitionsByHost) → StreamsMetadataState#onChange →
partitionsByTopic
- SourceTopicsInfo#maxPartitions = partitionsByTopic.get(topic).size()
- keyQueryMetadataForKey passes maxPartitions to StreamPartitioner#partitions
as the modulo
An under-count changes the modulo, so keys hash to the wrong partition.
*Impact*
Permanent, not transient. If the unconfigured instances hold 1 of a topic's 4
partitions, keys are hashed % 3 instead of % 4 and ~75% resolve to the wrong
partition. The host returned for those legitimately owns that partition — live,
store open — so it answers "key not found". No exception, no retry signal.
If no instance sets `application.server,` partitionsByHost stays empty,
isInitialized() is false, and queryMetadataForKey returns
KeyQueryMetadata.NOT_AVAILABLE — correct. Only the mixed configuration is
broken. While a mixed configuration is an incorrect configuration, it should
still not break IQ's correctness, but should only imply that some tasks are
unavailable for IQ.
To fix this, we need to change our custom assignor metadata, to include
information about "not available tasks" instead of just dropping them on the
floor. This implies version bumps and upgrade/downgrade concerns. We might need
a KIP for this (need to evaluate, as we actually have "version probing" so
maybe we could also address w/o a KIP)
> Missing `application.server` configuration silently breaks IQ key routing
> ("classic" protocol)
> ----------------------------------------------------------------------------------------------
>
> Key: KAFKA-20886
> URL: https://issues.apache.org/jira/browse/KAFKA-20886
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Reporter: Matthias J. Sax
> Priority: Minor
> Labels: needs-kip
>
> Filing this a "minor" as we consider a "mixed configuration" as invalid, so
> it's a user error triggering this issue. **
> *Summary*
> `application.server` is optional per instance. If some instances set it and
> others don't, IQ key routing silently resolves to the wrong host — including
> for keys hosted on the correctly configured instances.
> *Root cause*
> StreamsPartitionAssignor.populatePartitionsByHostMaps skips clients without
> an endpoint:
> if (hostInfo != null) \{ ... partitionsByHost.put(hostInfo, topicPartitions);
> ... }
> Their partitions are therefore absent from partitionsByHost — which is also
> the source of the partition count:
> - getTopicPartitionInfo(partitionsByHost) → StreamsMetadataState#onChange →
> partitionsByTopic
> - SourceTopicsInfo#maxPartitions = partitionsByTopic.get(topic).size()
> - keyQueryMetadataForKey passes maxPartitions to
> StreamPartitioner#partitions as the modulo
> An under-count changes the modulo, so keys hash to the wrong partition.
> *Impact*
> Permanent, not transient. If the unconfigured instances hold 1 of a topic's 4
> partitions, keys are hashed % 3 instead of % 4 and ~75% resolve to the wrong
> partition. The host returned for those legitimately owns that partition —
> live, store open — so it answers "key not found". No exception, no retry
> signal.
> If no instance sets `application.server,` partitionsByHost stays empty,
> isInitialized() is false, and queryMetadataForKey returns
> KeyQueryMetadata.NOT_AVAILABLE — correct. Only the mixed configuration is
> broken. While a mixed configuration is an incorrect configuration, it should
> still not break IQ's correctness, but should only imply that some tasks are
> unavailable for IQ.
> To fix this, we need to change our custom assignor metadata, to include
> information about "not available tasks" instead of just dropping them on the
> floor. This implies version bumps and upgrade/downgrade concerns. We might
> need a KIP for this (need to evaluate, as we actually have "version probing"
> so maybe we could also address w/o a KIP)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
