[
https://issues.apache.org/jira/browse/CAMEL-23996?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-23996.
---------------------------------
Resolution: Fixed
Fixed via https://github.com/apache/camel/pull/24691
> camel-kafka: KafkaIdempotentRepository and SingleNodeKafkaResumeStrategy
> correctness issues
> -------------------------------------------------------------------------------------------
>
> Key: CAMEL-23996
> URL: https://issues.apache.org/jira/browse/CAMEL-23996
> Project: Camel
> Issue Type: Bug
> Components: camel-kafka
> Reporter: Federico Mariani
> Assignee: Claus Ibsen
> Priority: Major
> Fix For: 4.22.0
>
>
> Code review of the Kafka idempotent repository and the single-node resume
> strategy found several correctness bugs. Filing together as one cleanup of
> the two classes. All line numbers refer to current {{main}} (4.22.0-SNAPSHOT).
> h3. KafkaIdempotentRepository
> *1. add() commits to the local cache before the Kafka broadcast; a broadcast
> failure is not rolled back → message wrongly filtered as duplicate* (lines
> 473-498)
> {{add(key)}} does {{cache.put(key, key)}} and then {{broadcastAction(key,
> add)}}, which throws {{RuntimeCamelException}} if the synchronous
> {{producer.send(...).get()}} fails. The key stays in the local cache. With an
> eager {{IdempotentConsumer}} (the default), the exchange fails, and on
> redelivery {{add()}} hits {{cache.containsKey(key) == true}} → returns false
> → the message is filtered as a "duplicate" and never processed. Peers also
> never received the add, so the cluster is inconsistent with this node.
> Pattern unchanged since CAMEL-10927.
> Fix: broadcast first and only cache on successful send, or
> {{cache.remove(key)}} in the failure path before rethrowing.
> *2. add() check-then-act is not atomic* (lines 473-484)
> {{if (cache.containsKey(key)) ... else cache.put(key, key)}} — two concurrent
> exchanges with the same key (multi-partition consumer, parallel routes) both
> pass. Other repositories synchronize {{add}} because {{IdempotentConsumer}}
> relies on atomicity. The cache is a thread-safe {{SimpleLRUCache}}: use
> {{cache.putIfAbsent(key, key) == null}} and only broadcast when insertion won.
> *3. doStop() closes the KafkaConsumer while the poller thread may still be
> inside poll() → ConcurrentModificationException, leaked consumer* (lines
> 383-393)
> {{doStop}} calls {{ExecutorServiceManager.shutdown(executorService)}}
> (non-blocking), then {{stopService(poller)}}, then immediately
> {{IOHelper.close(consumer)}}. The poller thread is typically blocked inside
> {{consumer.poll(100ms)}}; {{KafkaConsumer}} is not thread-safe and
> {{close()}} from a second thread throws {{ConcurrentModificationException}},
> which {{IOHelper.close}} does not swallow — it propagates out of {{doStop()}}
> and the consumer is left unclosed (leaked network threads). The window is
> nearly the whole poll interval, so this fires with high probability on every
> context stop (logged as WARN, hence unnoticed). Structure from CAMEL-20682.
> Fix: await poller termination before closing the consumer, or close the
> consumer from the poller thread itself.
> *4. clear() never clears the local cache; with startupOnly=true it never
> takes effect at all* (lines 522-525)
> {{clear()}} only broadcasts and relies on the poller consuming the record
> back; with {{startupOnly=true}} there is no running poller, so a JMX/managed
> {{clear()}} silently does nothing until restart. Fix: also {{cache.clear()}}
> locally, mirroring {{add}}/{{remove}}.
> *Minor:* a null-valued record on the topic throws NPE from
> {{CacheAction.valueOf(null)}} — during {{populateCache()}} this aborts
> CamelContext startup (line 451); the "unexpected action" warning logs the key
> instead of the value (lines 465-468); {{doStart}} mutates caller-supplied
> {{consumerConfig}}/{{producerConfig}} Properties in place.
> h3. SingleNodeKafkaResumeStrategy
> *5. Duplicate subscribe discards the rebalance listener that fills the cache*
> (lines 209-212)
> {code:java}
> subscribe(consumer); // subscribes with a listener that seeks back
> cache.capacity() records
> LOG.debug("Loading records from topic {}",
> resumeStrategyConfiguration.getTopic());
> consumer.subscribe(Collections.singletonList(resumeStrategyConfiguration.getTopic()));
> // replaces subscription + listener!
> {code}
> A second {{subscribe}} replaces the previous subscription *and* its
> {{ConsumerRebalanceListener}} with a no-op. With {{FillPolicy.MINIMIZING}}
> ({{auto.offset.reset=latest}} per the builder, which always generates a fresh
> random group id) the consumer starts at log end, the rewind never runs, the
> resume cache stays empty → the route silently reprocesses everything. In
> CAMEL-18356 there were two consumers (one subscribe each); a later
> consolidation merged them and both subscribe calls survived. Fix: delete the
> plain {{consumer.subscribe(...)}} line.
> *6. Initialization wait is inverted/dead, so resume() can run before offsets
> are loaded* (lines 359-366)
> {{getAdapter()}} calls {{waitForInitialization()}} only when {{adapter ==
> null}}. In the normal flow {{DefaultRoute}} sets the adapter *before*
> {{ResumeStrategyHelper.resume()}} runs, so {{getAdapter()}} returns
> immediately and {{resume()}} races the async topic load ({{initLatch}} is
> never awaited). Conversely, waiting when the adapter is null is pointless —
> the refresh thread never sets it. Guard added in CAMEL-18688. Fix: wait on
> {{initLatch}} (when non-null, i.e. after {{loadCache()}}) regardless of
> adapter nullity.
> *7. stop() unlocks a lock it may not hold → IllegalMonitorStateException,
> skipped consumer shutdown* (lines 394-409)
> If {{writeLock.tryLock(1, SECONDS)}} returns false (e.g. another thread is
> inside {{updateLastOffset}} blocked in {{producer.send}} on metadata for up
> to {{max.block.ms}}) or throws {{InterruptedException}}, execution still
> reaches {{finally \{ writeLock.unlock(); \}}} →
> {{IllegalMonitorStateException}} out of {{stop()}}, and the consumer-shutdown
> block below (wakeup + executor shutdown) is skipped → refresh thread and
> consumer leak. From CAMEL-18362. Fix: only unlock when acquired.
> *Minor:* {{produce()}} send-failures are only logged with the default null
> callback (resume state silently lost); javadoc still references the
> {{producerErrors}} counter removed in CAMEL-18148, leaving {{RecordError}} an
> unused public class; {{consume(int, Consumer)}} has no callers;
> {{consumer}}/{{initLatch}} fields are written by the executor thread and read
> by {{stop()}} without volatile.
> ----
> _This issue was researched and written by Claude Code on behalf of Federico
> Mariani (GitHub: Croway). Findings were verified against the source and git
> history before filing._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
