[
https://issues.apache.org/jira/browse/SPARK-58889?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18106417#comment-18106417
]
Jason Teoh commented on SPARK-58889:
------------------------------------
Actually this ticket is too narrowly scoped - fixing just the
delete/invalidateKey will cause concurrency bugs and test failures
([https://github.com/apache/spark/pull/58141)]
The broader solution is captured in an existing TODO: include the kafka params
as part of the cache key. However, it seems as if there may have been some
reservations about including the kafka params first. We would need to explore
that rather than the currently-scoped ticket limited to delete/validate.
> 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]
