davsclaus commented on code in PR #25369:
URL: https://github.com/apache/camel/pull/25369#discussion_r3726127336
##########
components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java:
##########
@@ -121,15 +121,18 @@ protected int poll() throws Exception {
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);
Review 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.
--
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]
