CAMEL-24357: camel-aws2-s3-vectors - consumer returns no results (topK=0), ignores delay, and can lose vectors - #25369
Merged
Merged
Conversation
…ors delay, and no longer loses vectors on failure The aws2-s3-vectors consumer had several defects rooted in consumer options that shadow the base scheduled-poll options: - poll() sent topK(Math.min(getMaxMessagesPerPoll(), topK)) where the base maxMessagesPerPoll field was never wired (default 0), so topK(0) was sent and AWS returned nothing. maxMessagesPerPoll is now wired from the configuration in createConsumer and topK treats a non-positive cap as "unlimited". - the delay option bound to a configuration field the consumer never read; it is now propagated to the consumer's scheduler. - vectors were marked processed before routing, so a failed exchange was skipped forever; the de-dup set is now only populated when deleteAfterRead=false and an id is dropped again on failure so the vector can be retried on a later poll. Adds AWS2S3VectorsConsumerTest covering the topK and delay fixes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Andrea Cosentino <ancosen@gmail.com>
Contributor
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
davsclaus
approved these changes
Aug 6, 2026
davsclaus
left a comment
davsclaus
left a comment
Contributor
There was a problem hiding this comment.
Solid bugfix PR — all three issues are well-analyzed and the fixes are correct.
Verified:
topK(0)bug:resolveTopK()correctly treats non-positivemaxMessagesPerPollas "unlimited" instead of capping to 0. Matches theaws2-s3endpoint pattern.delayshadowing: the@UriParamon the configuration captures the value beforeconfigureConsumercan map it, so explicit propagation is needed and correct.- De-dup logic: clean separation — deletion prevents re-delivery so the set is only needed when
deleteAfterRead=false. TheVectorDedupSynchronization.onFailurecorrectly drops the id on failure for retry.
Tests follow project conventions (AssertJ, package-private visibility, descriptive names) and directly exercise the bug conditions.
One minor non-blocking observation noted inline.
Note: this review covers project rules and conventions. It does not replace specialized AI review tools (CodeRabbit, Sourcery) or static analysis (SonarCloud).
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
| // Track for de-duplication only when we are not deleting (a deleted vector cannot be | ||
| // returned again). Mark it now to avoid re-delivering it on overlapping polls, but drop it | ||
| // again if the exchange fails so it can be retried on a subsequent poll. | ||
| processedVectorIds.add(vectorId); |
Contributor
There was a problem hiding this comment.
Non-blocking / follow-up: processedVectorIds is a plain HashSet, now written from both poll() (scheduler thread) and VectorDedupSynchronization.onFailure() (exchange completion thread — potentially different with async routing). For most configurations this is fine (synchronous routing), but a ConcurrentHashMap.newKeySet() would be safer against subtle races.
Also, when deleteAfterRead=false the set grows unboundedly until doStop(). Both are pre-existing design concerns (not introduced by this PR), but worth a follow-up.
Contributor
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 9 tested, 29 compile-only — current: 9 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 9).
|
davsclaus
pushed a commit
that referenced
this pull request
Aug 6, 2026
…ctor loss Backport of #25369 to camel-4.18.x. The consumer now wires maxMessagesPerPoll so topK is valid (>= 1), propagates the delay option to the scheduler, and only marks a vector processed on success so failed exchanges can be retried. Closes #25377 Co-authored-by: Claude Opus 4.8 <noreply@anthropic.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
2 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.
CAMEL-24357: camel-aws2-s3-vectors — consumer was non-functional and could lose vectors
The
aws2-s3-vectorsconsumer had several defects, all rooted in consumer options that shadow the base scheduled-poll options.1. Consumer always sent
topK(0)— returned nothing (primary, HIGH)AWS2S3VectorsConsumer.poll()built the query withtopK(Math.min(getMaxMessagesPerPoll(), getConfiguration().getTopK())).getMaxMessagesPerPoll()is the baseScheduledBatchPollingConsumerfield, which defaults to 0 and was never wired —AWS2S3VectorsEndpoint.createConsumer()only calledconfigureConsumer(consumer), neversetMaxMessagesPerPoll(..)(unlikeAWS2S3Endpoint). So the expression wasMath.min(0, topK) = 0and every poll senttopK(0), which AWS S3 Vectors rejects (topK >= 1). The consumer delivered zero messages out of the box.Fix: wire
setMaxMessagesPerPollfrom the configuration increateConsumer, and resolvetopKso that a non-positivemaxMessagesPerPoll(the "unlimited" default) means "use the configuredtopK" instead of capping to zero.2.
delayoption was ignoreddelayis declared on the configuration, so?delay=bound toconfiguration.setDelay(..)— a value the consumer never read; the real poll interval came from the inheritedScheduledPollEndpoint. A user setting?delay=60000was silently ignored.Fix: propagate
configuration.getDelay()to the consumer's scheduler increateConsumer(same placemaxMessagesPerPollis now wired).3. Vectors marked processed before routing → event loss on failure
poll()added each vector id toprocessedVectorIdsat enqueue time, before the exchange was routed. If routing later failed, the vector stayed in the index but was in the de-dup set, so the fixed similarity query skipped it forever. The set was also cleared only on stop (unbounded growth).Fix: only track
processedVectorIdswhendeleteAfterRead=false(deletion already prevents re-delivery, so the set no longer grows in that mode), and drop the id again on failure (VectorDedupSynchronization.onFailure) so a failed exchange is retried on a later poll.Tests
New
AWS2S3VectorsConsumerTest(Mockito):consumerSendsPositiveTopKcaptures theQueryVectorsRequestand assertstopKis>= 1(fails against the old code), anddelayOptionDrivesTheConsumerPollIntervalasserts the configureddelayreaches the consumer's scheduler. Existing producer tests still pass; full reactor build is green.No public API change.
assertj-coreadded as a test dependency (project-standard). Targetsmain(4.22.0) andcamel-4.18.x(the module was added in 4.17.0; it does not exist on 4.14.x).Claude Code on behalf of oscerd