shashank created CAMEL-26740:
--------------------------------
Summary: 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
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)
