Jason Teoh created SPARK-58889:
----------------------------------
Summary: 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
`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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
