[
https://issues.apache.org/jira/browse/CAMEL-26740?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-26740.
---------------------------------
Resolution: Fixed
Merged in https://github.com/apache/camel/pull/29434 (a44c49fa3420) for 4.24.0.
_Claude Code on behalf of davsclaus_
> camel-kafka - KafkaKeyValueRepository keys() and size() fail once an entry
> has expired, and putIfAbsent can report an insert that did not happen
> ------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-26740
> URL: https://issues.apache.org/jira/browse/CAMEL-26740
> Project: Camel
> Issue Type: Bug
> Components: camel-kafka
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> Two problems with the expiry handling of {{KafkaKeyValueRepository}}
> (camel-kafka, since 4.23):
> *1. keys() and size() throw UnsupportedOperationException once an entry has
> expired*
> {{keys()}} and {{size()}} (also the JMX attribute {{Size}}) first call
> {{evictExpired()}}, which removes expired entries with {{Iterator.remove()}}
> on {{cache.entrySet().iterator()}}. The cache comes from {{LRUCacheFactory}},
> by default a {{SimpleLRUCache}}, whose {{entrySet()}} is an unmodifiable copy
> ({{Collectors.toUnmodifiableSet()}}). So as soon as one entry with a time to
> live has expired, every call of {{keys()}} and {{size()}} fails with
> {{UnsupportedOperationException: remove}}, and so do the callers of
> {{keys()}} (for example the {{keys}} and {{size}} operations of the
> state-store component). The class Javadoc says that entries replayed from the
> topic on startup "will appear as expired immediately", so after a restart
> this can happen at once.
> {code:java}
> repo.put("a", "1", Duration.ofSeconds(1));
> // one second later
> repo.keys(); // UnsupportedOperationException: remove
> {code}
> *2. putIfAbsent can return null ("stored") although nothing was stored or
> sent*
> {{putIfAbsent}} reads the entry, removes it if it has expired, then calls
> {{cache.putIfAbsent}}. When another writer stored an entry for the key in
> between (another thread, or the topic poller applying a record whose expiry
> time has already passed, which it does when it lags behind by more than the
> time to live), and that entry has expired, the method returns {{null}}. It
> has not stored the caller's value and has not sent it to the topic. The
> interface Javadoc, the method's Javadoc and the state-store docs all say
> {{null}} means stored, so the caller acts as the owner of the key while no
> other instance (and no restart) can see its value, and the next
> {{putIfAbsent}} on that key also succeeds.
> Expected: {{keys()}} and {{size()}} skip and evict expired entries;
> {{putIfAbsent}} returns {{null}} only when it stored and sent the value.
> Suggested fix: in {{evictExpired()}} remove through the cache
> ({{cache.remove(key, entry)}}, so a key put again meanwhile is kept); in
> {{putIfAbsent}} replace an expired entry atomically ({{cache.replace(key,
> expired, newEntry)}}) and, when that fails because the key changed meanwhile,
> look again.
> Found by a code review of the Kafka repositories; both reproduce in unit
> tests with a {{MockProducer}} and no broker.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
