This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 43cd5f0e70d9 CAMEL-24357: camel-aws2-s3-vectors - consumer sends a
valid topK, honors delay, and no longer loses vectors on failure
43cd5f0e70d9 is described below
commit 43cd5f0e70d9274a83aecfdde346d9b3c66c9d29
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Aug 6 12:32:42 2026 +0200
CAMEL-24357: camel-aws2-s3-vectors - consumer sends a valid topK, honors
delay, and no longer loses vectors on failure
The aws2-s3-vectors consumer had three defects rooted in consumer options
shadowing the base scheduled-poll options:
- poll() sent topK(0) because maxMessagesPerPoll was never wired (default
0),
so the consumer delivered zero messages. Now wired from configuration and
treats non-positive cap as "unlimited".
- The delay option bound to a configuration field the consumer never read;
now propagated to the consumer's scheduler.
- Vectors were marked processed before routing, so a failed exchange was
skipped forever; the de-dup set now only applies when
deleteAfterRead=false
and drops ids on failure for retry.
Adds AWS2S3VectorsConsumerTest covering the topK and delay fixes.
Closes #25369
Co-Authored-By: Claude Opus 4.8 <[email protected]>
---
components/camel-aws/camel-aws2-s3-vectors/pom.xml | 5 ++
.../aws2/s3vectors/AWS2S3VectorsConsumer.java | 41 ++++++++++--
.../aws2/s3vectors/AWS2S3VectorsEndpoint.java | 4 ++
.../aws2/s3vectors/AWS2S3VectorsConsumerTest.java | 75 ++++++++++++++++++++++
4 files changed, 121 insertions(+), 4 deletions(-)
diff --git a/components/camel-aws/camel-aws2-s3-vectors/pom.xml
b/components/camel-aws/camel-aws2-s3-vectors/pom.xml
index a39f057666b1..e8fd76d1969d 100644
--- a/components/camel-aws/camel-aws2-s3-vectors/pom.xml
+++ b/components/camel-aws/camel-aws2-s3-vectors/pom.xml
@@ -70,6 +70,11 @@
<artifactId>camel-test-spring-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- test infra -->
<dependency>
diff --git
a/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
b/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
index 9267d901a483..3a7811d62dc0 100644
---
a/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
+++
b/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
@@ -83,7 +83,7 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
.vectorBucketName(vectorBucketName)
.indexName(vectorIndexName)
.queryVector(VectorData.builder().float32(queryVector).build())
- .topK(Math.min(getMaxMessagesPerPoll(),
getConfiguration().getTopK()));
+ .topK(resolveTopK());
// Add metadata filter if configured
String metadataFilter =
getConfiguration().getConsumerMetadataFilter();
@@ -121,15 +121,18 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
message.setHeader(AWS2S3VectorsConstants.VECTOR_BUCKET_NAME, vectorBucketName);
message.setHeader(AWS2S3VectorsConstants.VECTOR_INDEX_NAME, vectorIndexName);
- // Add to processed set
- processedVectorIds.add(vectorId);
-
// Add delete callback if deleteAfterRead is enabled
if (getConfiguration().isDeleteAfterRead()) {
exchange.getExchangeExtension().addOnCompletion(
new VectorDeleteSynchronization(
getS3VectorsClient(),
vectorBucketName, vectorIndexName,
vectorId));
+ } else {
+ // 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);
+ exchange.getExchangeExtension().addOnCompletion(new
VectorDedupSynchronization(vectorId));
}
exchanges.add(exchange);
@@ -200,6 +203,18 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
return getEndpoint().getS3VectorsClient();
}
+ /**
+ * Resolves the number of nearest vectors to request from the similarity
search. The configured {@code topK} is
+ * capped by {@code maxMessagesPerPoll} only when that is a positive
value; a non-positive
+ * {@code maxMessagesPerPoll} (the "unlimited" default) leaves the
configured {@code topK} unchanged. AWS S3 Vectors
+ * requires {@code topK >= 1}.
+ */
+ private int resolveTopK() {
+ int topK = getConfiguration().getTopK();
+ int max = getMaxMessagesPerPoll();
+ return max > 0 ? Math.min(max, topK) : topK;
+ }
+
/**
* Parse comma-separated vector string into List<Float>
*
@@ -268,4 +283,22 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
LOG.trace("Exchange failed, not deleting vector [{}]", vectorId);
}
}
+
+ /**
+ * Synchronization callback that drops a vector id from the de-duplication
set when the exchange fails, so the
+ * vector can be reprocessed on a subsequent poll instead of being skipped
forever.
+ */
+ private class VectorDedupSynchronization extends SynchronizationAdapter {
+
+ private final String vectorId;
+
+ VectorDedupSynchronization(String vectorId) {
+ this.vectorId = vectorId;
+ }
+
+ @Override
+ public void onFailure(Exchange exchange) {
+ processedVectorIds.remove(vectorId);
+ }
+ }
}
diff --git
a/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsEndpoint.java
b/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsEndpoint.java
index 5c6be4a6a2cd..23dea141b1fd 100644
---
a/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsEndpoint.java
+++
b/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsEndpoint.java
@@ -85,6 +85,10 @@ public class AWS2S3VectorsEndpoint extends
ScheduledPollEndpoint implements Endp
public Consumer createConsumer(Processor processor) throws Exception {
AWS2S3VectorsConsumer consumer = new AWS2S3VectorsConsumer(this,
processor);
configureConsumer(consumer);
+ consumer.setMaxMessagesPerPoll(configuration.getMaxMessagesPerPoll());
+ // the delay is a configuration option (it would otherwise shadow
ScheduledPollEndpoint#delay and be
+ // ignored), so propagate it to the consumer's scheduler
+ consumer.setDelay(configuration.getDelay());
return consumer;
}
diff --git
a/components/camel-aws/camel-aws2-s3-vectors/src/test/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumerTest.java
b/components/camel-aws/camel-aws2-s3-vectors/src/test/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumerTest.java
new file mode 100644
index 000000000000..285bc62dc5bc
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3-vectors/src/test/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumerTest.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.s3vectors;
+
+import java.util.List;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import software.amazon.awssdk.services.s3vectors.S3VectorsClient;
+import software.amazon.awssdk.services.s3vectors.model.QueryVectorsRequest;
+import software.amazon.awssdk.services.s3vectors.model.QueryVectorsResponse;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.when;
+
+class AWS2S3VectorsConsumerTest {
+
+ @Test
+ void consumerSendsPositiveTopK() throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ S3VectorsClient client = Mockito.mock(S3VectorsClient.class);
+ ArgumentCaptor<QueryVectorsRequest> captor =
ArgumentCaptor.forClass(QueryVectorsRequest.class);
+ when(client.queryVectors(captor.capture()))
+
.thenReturn(QueryVectorsResponse.builder().vectors(List.of()).build());
+
+ AWS2S3VectorsEndpoint endpoint = context.getEndpoint(
+
"aws2-s3-vectors://test-bucket?vectorIndexName=test-index&accessKey=test&secretKey=test",
+ AWS2S3VectorsEndpoint.class);
+ endpoint.getConfiguration().setConsumerQueryVector("0.1,0.2,0.3");
+ endpoint.setS3VectorsClient(client);
+
+ AWS2S3VectorsConsumer consumer = (AWS2S3VectorsConsumer)
endpoint.createConsumer(exchange -> {
+ });
+
+ consumer.poll();
+
+ // Before the fix the topK was Math.min(maxMessagesPerPoll, topK)
where maxMessagesPerPoll was the
+ // unwired base-class field (0), so topK(0) was sent and AWS
returned nothing. It must now be >= 1.
+ assertThat(captor.getValue().topK())
+ .as("topK sent to AWS S3 Vectors must be >= 1")
+ .isEqualTo(10);
+ }
+ }
+
+ @Test
+ void delayOptionDrivesTheConsumerPollInterval() throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ AWS2S3VectorsEndpoint endpoint = context.getEndpoint(
+
"aws2-s3-vectors://test-bucket?vectorIndexName=test-index&delay=1234&accessKey=test&secretKey=test",
+ AWS2S3VectorsEndpoint.class);
+
+ AWS2S3VectorsConsumer consumer = (AWS2S3VectorsConsumer)
endpoint.createConsumer(exchange -> {
+ });
+
+ // delay used to be ignored (a shadow field on the configuration);
it must now drive the actual poll interval
+ assertThat(consumer.getDelay()).isEqualTo(1234L);
+ }
+ }
+}
