CAMEL-24248 Concurrent pollEnrich can fail to stop in-use consumers after cache eviction - #25036
Merged
Conversation
Contributor
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
davsclaus
approved these changes
Jul 23, 2026
davsclaus
left a comment
davsclaus
left a comment
Contributor
There was a problem hiding this comment.
Solid fix for a real production OOM issue. The root cause is clear: when pollEnrich uses a cacheSize smaller than the concurrency level, the LRU cache evicts a service's pool while the service is still in use. When release() is called afterward, pool.get(endpoint) returns null and the method silently does nothing — the consumer is never stopped, leaking resources.
The fix is correct and minimal:
- In
release(): when the pool is gone, stop and remove the orphaned service instead of silently ignoring it - Extract
stopAndRemove()helper to deduplicate the stop+remove logic already present inonEvict()'s else-branch
I traced the eviction flow and confirmed no double-stop risk: p.stop() drains the pool queue, but the in-use consumer was never in the queue — it's only reachable through release().
Test is well-constructed: uses CountDownLatch for deterministic concurrency (no Thread.sleep), AssertJ assertions, proper cleanup in @AfterEach.
Minor nit (non-blocking): the refactored onEvict() computes Endpoint e at the top, then the else-branch calls stopAndRemove(s) which recomputes getEndpoint.apply(s). The old code reused the cached e. Negligible since getEndpoint is a simple field accessor, but could be cleaned up by passing e as a parameter.
Note: this review does not replace specialized tools such as CodeRabbit, Sourcery, or SonarCloud.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
| stopAndRemove(s); | ||
| } | ||
| } | ||
|
|
Contributor
There was a problem hiding this comment.
This is the key fix — previously when the pool was evicted while a service was in use, release() silently did nothing and the consumer leaked. Now it properly stops and removes the orphaned service. 👍
| } catch (Exception ex) { | ||
| LOG.debug("Error removing service: {}. This exception is ignored.", s, ex); | ||
| } | ||
| stopAndRemove(s); |
Contributor
There was a problem hiding this comment.
Minor nit: onEvict() already computed Endpoint e = getEndpoint.apply(s) at line 87, but stopAndRemove(s) recomputes it. The old code reused the cached e. Since getEndpoint is typically a simple field accessor this is negligible, but you could pass e as a parameter to avoid the redundant lookup.
Contributor
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 550 tested, 29 compile-only — current: 550 all testedMaveniverse Scalpel detected 579 affected modules (current approach: 550).
|
Add deterministic coverage for concurrent dynamic pollEnrich calls when the consumer cache is smaller than the number of active consumers. Verify that an in-use consumer is stopped after its pool is evicted, with sequential eviction coverage as a control case. Co-authored-by: GPT-5.6 Sol <noreply@openai.com>
Stop and remove a service when it is released after its endpoint pool has already been evicted. This preserves lifecycle ownership for services that were still in use during eviction and prevents their resources from remaining allocated. Reuse the same cleanup path when eviction encounters a service that is no longer associated with a pool. Co-authored-by: GPT-5.6 Sol <noreply@openai.com>
gnodet
approved these changes
Jul 23, 2026
gnodet
left a comment
gnodet
left a comment
Contributor
There was a problem hiding this comment.
Clean, correct fix for a real production OOM bug. When pollEnrich uses a cacheSize smaller than the concurrency level, the LRU cache can evict a consumer's pool while the consumer is still in use. The current release() silently does nothing when the pool is gone, leaking the consumer. The fix properly stops and removes the orphaned consumer.
Concurrency analysis:
- The eviction flow is fully synchronous with
SimpleLRUCache:cache.putIfAbsent(B, B)triggerscallEviction()→onEvict(A)→p.stop()all on the same thread that calledacquire(). This guarantees that by the time the thread using consumer A callsrelease(), the pool is already removed. No TOCTOU risk in the newrelease()path. - No double-stop risk:
MultiplePool.stop()drains thequeue(which contains only idle/released consumers) and does NOT process theevictsdeque. An in-use consumer is not in the queue, so it is only stopped once — bystopAndRemove()in the newrelease()path. Even if a double-stop occurred,ServiceSupport.stop()is idempotent.
Test quality: Well-constructed with CountDownLatch and Future for deterministic concurrency control, no Thread.sleep() calls (compliant with project rules). The @AfterEach cleanup properly releases latches before JUnit5 stops the CamelContext.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Claude Code on behalf of Guillaume Nodet
davsclaus
added a commit
that referenced
this pull request
Jul 24, 2026
… stop in-use consumers after cache eviction (#25067) CAMEL-24248: Stop services released after pool eviction in ServicePool When pollEnrich processes concurrent exchanges using different dynamic endpoint URIs and its consumer cache is smaller than the concurrency level, a polling consumer's pool can be evicted while the consumer is still in use. The released consumer was never stopped, causing resource leaks and steadily increasing memory usage. Add cleanup in ServicePool.release() to stop and remove a service when its endpoint pool has already been evicted. Extract common stop-and-remove logic into a shared method. Closes #25036 Co-authored-by: Mark Snijder <snijderd@gmail.com> Co-authored-by: GPT-5.6 Sol <noreply@openai.com>
davsclaus
added a commit
that referenced
this pull request
Jul 24, 2026
… stop in-use consumers after cache eviction (#25068) CAMEL-24248: Stop services released after pool eviction in ServicePool When pollEnrich processes concurrent exchanges using different dynamic endpoint URIs and its consumer cache is smaller than the concurrency level, a polling consumer's pool can be evicted while the consumer is still in use. The released consumer was never stopped, causing resource leaks and steadily increasing memory usage. Add cleanup in ServicePool.release() to stop and remove a service when its endpoint pool has already been evicted. Extract common stop-and-remove logic into a shared method. Closes #25036 Co-authored-by: Mark Snijder <snijderd@gmail.com> Co-authored-by: GPT-5.6 Sol <noreply@openai.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
When pollEnrich processes concurrent exchanges using different dynamic endpoint URIs and its consumer cache is smaller than the concurrency level, a polling consumer’s pool can be evicted while the consumer is still in use. When that exchange finishes, the consumer is released after its pool has already been removed, so the consumer is not stopped and its resources remain allocated.
Repeated requests cause these unclosed consumers to accumulate, resulting in steadily increasing memory usage and potentially leaking component-specific resources such as connections, clients, threads, or buffers. In the attached reproducer, four concurrent SEDA consumers use dynamic polling URIs with cacheSize(1). After 400 requests, 304 polling consumers remain active and retain approximately 76 MB, whereas only the single cached consumer should remain.
We found out about this issue on production with SQS component with the setting concurrentConsumers=4 and after that a pollEnrich with cacheSize set to 1 (with sftp component). And we saw our application go OOM after about 4 days consistently.
Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.