[
https://issues.apache.org/jira/browse/SPARK-58889?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jason Teoh resolved SPARK-58889.
--------------------------------
Resolution: Won't Fix
If there's a ticket for the associated TODO then please feel free to link.
> InternalKafkaConsumerPool retains cached kafkaParams after
> invalidateKey()/reset()
> ----------------------------------------------------------------------------------
>
> Key: SPARK-58889
> URL: https://issues.apache.org/jira/browse/SPARK-58889
> Project: Spark
> Issue Type: Bug
> Components: Bug
> Affects Versions: 4.2.0
> Reporter: Jason Teoh
> Priority: Minor
> Labels: pull-request-available
>
> `InternalKafkaConsumerPool` pools consumers by `CacheKey`, which is only
> `(groupId, topicPartition)`. Because that key does not carry the Kafka
> parameters, the factory remembers them separately in
> `ObjectFactory.keyToKafkaParams` so it can build a consumer for a key.
> The two existing invalidation paths clear the pooled consumers but never
> clean up that map which can still contain outdated configs:
> * `invalidateKey(key)` calls pool.clear(key)` but leaves
> `keyToKafkaParams(key)` in place.
> * `reset()` does similar (clear pool but not the keyToKafkaParams)
> A later `borrowObject` for the same `CacheKey` whose
> `kafkaParams` differs will then pick up the stale entry and the requirement
> fails:
> ```
> java.lang.IllegalArgumentException: requirement failed: Kafka parameters for
> same cache key should be equal. old parameters: \{...} new parameters:
> \{...}
> ```
> A simple example is when kafka groupId is reused but the config differs
> between excutions.
>
> (Apologies, seems Jira does not support the code formatting - but hopefully
> still clear with the ticks)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
