oscerd opened a new pull request, #25369: URL: https://github.com/apache/camel/pull/25369
# CAMEL-24357: camel-aws2-s3-vectors — consumer was non-functional and could lose vectors The `aws2-s3-vectors` consumer 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 with `topK(Math.min(getMaxMessagesPerPoll(), getConfiguration().getTopK()))`. `getMaxMessagesPerPoll()` is the base `ScheduledBatchPollingConsumer` field, which defaults to **0** and was never wired — `AWS2S3VectorsEndpoint.createConsumer()` only called `configureConsumer(consumer)`, never `setMaxMessagesPerPoll(..)` (unlike `AWS2S3Endpoint`). So the expression was `Math.min(0, topK) = 0` and every poll sent `topK(0)`, which AWS S3 Vectors rejects (`topK >= 1`). **The consumer delivered zero messages out of the box.** **Fix:** wire `setMaxMessagesPerPoll` from the configuration in `createConsumer`, and resolve `topK` so that a non-positive `maxMessagesPerPoll` (the "unlimited" default) means "use the configured `topK`" instead of capping to zero. ## 2. `delay` option was ignored `delay` is declared on the configuration, so `?delay=` bound to `configuration.setDelay(..)` — a value the consumer never read; the real poll interval came from the inherited `ScheduledPollEndpoint`. A user setting `?delay=60000` was silently ignored. **Fix:** propagate `configuration.getDelay()` to the consumer's scheduler in `createConsumer` (same place `maxMessagesPerPoll` is now wired). ## 3. Vectors marked processed before routing → event loss on failure `poll()` added each vector id to `processedVectorIds` at 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 `processedVectorIds` when `deleteAfterRead=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): `consumerSendsPositiveTopK` captures the `QueryVectorsRequest` and asserts `topK` is `>= 1` (fails against the old code), and `delayOptionDrivesTheConsumerPollInterval` asserts the configured `delay` reaches the consumer's scheduler. Existing producer tests still pass; full reactor build is green. No public API change. `assertj-core` added as a test dependency (project-standard). Targets `main` (4.22.0) and `camel-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_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
