Federico Mariani created CAMEL-23996:
----------------------------------------
Summary: 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
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)
