You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2026/08/26 09:52:53 UTC
(camel) branch main updated: CAMEL-24491: camel-ibm-cos - make consumer in-progress deduplication effective (#25740)
This is an automated email from the ASF dual-hosted git repository.
oscerd 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 9b26b811fd27 CAMEL-24491: camel-ibm-cos - make consumer in-progress deduplication effective (#25740)
9b26b811fd27 is described below
commit 9b26b811fd2798bda130d441538cd339d8ccf4e2
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Wed Aug 26 11:52:44 2026 +0200
CAMEL-24491: camel-ibm-cos - make consumer in-progress deduplication effective (#25740)
* CAMEL-24491: camel-ibm-cos - make consumer in-progress deduplication effective
The IBM COS consumer's in-progress deduplication never worked: createExchanges()
only checked getInProgressRepository().contains(key) without ever adding the
key, the MemoryIdempotentRepository was started only after doStart()'s early
returns (so never in the normal consuming case), and processCommit/processRollback
never removed the key. As a result overlapping polls could re-deliver an object
still being processed when deleteAfterRead/moveAfterRead were off.
Mirror the camel-aws2-s3 consumer this module was copied from: use
getInProgressRepository().add(key) as the atomic guard (skip when it returns
false), remove the key in processCommit (finally) and processRollback via a
null-safe helper, and start the in-progress repository up front in doStart()
before any early return.
Co-Authored-By: Claude Opus 4.8 <no...@anthropic.com>
Signed-off-by: Andrea Cosentino <an...@gmail.com>
* CAMEL-24491: release the in-progress key when the object fetch or exchange creation fails
Addresses review feedback: if getObject/createExchange throws after add(key)
has claimed the in-progress key, no Synchronization is attached to remove it, so
the object would be left permanently unconsumable. Release the key (and skip the
object for this poll) so a transient failure does not silently block the object.
Co-Authored-By: Claude Opus 4.8 <no...@anthropic.com>
Signed-off-by: Andrea Cosentino <an...@gmail.com>
---------
Signed-off-by: Andrea Cosentino <an...@gmail.com>
Co-authored-by: Claude Opus 4.8 <no...@anthropic.com>
---
.../camel/component/ibm/cos/IBMCOSConsumer.java | 27 ++++++++++++++++++----
.../camel/component/ibm/cos/IBMCOSEndpoint.java | 5 ++--
2 files changed, 25 insertions(+), 7 deletions(-)
diff --git a/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java b/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java
index 233723ac43f7..66de7917f054 100644
--- a/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java
+++ b/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSConsumer.java
@@ -160,15 +160,23 @@ public class IBMCOSConsumer extends ScheduledBatchPollingConsumer {
}
if (getEndpoint().getInProgressRepository() != null
- && getEndpoint().getInProgressRepository().contains(s3ObjectSummary.getKey())) {
+ && !getEndpoint().getInProgressRepository().add(s3ObjectSummary.getKey())) {
LOG.trace("Object {} is already in progress", s3ObjectSummary.getKey());
continue;
}
- S3Object s3Object = getCosClient().getObject(
- new GetObjectRequest(s3ObjectSummary.getBucketName(), s3ObjectSummary.getKey()));
- Exchange exchange = createExchange(s3Object, s3ObjectSummary.getKey());
- exchanges.add(exchange);
+ try {
+ S3Object s3Object = getCosClient().getObject(
+ new GetObjectRequest(s3ObjectSummary.getBucketName(), s3ObjectSummary.getKey()));
+ Exchange exchange = createExchange(s3Object, s3ObjectSummary.getKey());
+ exchanges.add(exchange);
+ } catch (Exception e) {
+ // Fetching the object or creating the exchange failed after we claimed the in-progress key;
+ // release it so the object is not left permanently unconsumable, and skip it for this poll.
+ LOG.warn("Error fetching object {} from bucket {}: {}. Skipping it for this poll.",
+ s3ObjectSummary.getKey(), s3ObjectSummary.getBucketName(), e.getMessage());
+ removeInProgress(s3ObjectSummary.getKey());
+ }
}
return exchanges;
@@ -246,11 +254,20 @@ public class IBMCOSConsumer extends ScheduledBatchPollingConsumer {
}
} catch (Exception e) {
LOG.warn("Error during post processing of object {} from bucket {}: {}", key, bucketName, e.getMessage());
+ } finally {
+ removeInProgress(key);
}
}
protected void processRollback(String key) {
LOG.trace("Processing failed for object with key {}", key);
+ removeInProgress(key);
+ }
+
+ private void removeInProgress(String key) {
+ if (getEndpoint().getInProgressRepository() != null) {
+ getEndpoint().getInProgressRepository().remove(key);
+ }
}
private void copyObject(String bucketName, String key) {
diff --git a/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSEndpoint.java b/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSEndpoint.java
index 63d9d5b5478b..dcbafcfd3dc9 100644
--- a/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSEndpoint.java
+++ b/components/camel-ibm/camel-ibm-cos/src/main/java/org/apache/camel/component/ibm/cos/IBMCOSEndpoint.java
@@ -115,6 +115,9 @@ public class IBMCOSEndpoint extends ScheduledPollEndpoint implements EndpointSer
protected void doStart() throws Exception {
super.doStart();
+ // Start the in-progress repository up front so consumer deduplication works even when doStart returns early
+ ServiceHelper.startService(inProgressRepository);
+
cosClient = configuration.getCosClient() != null
? configuration.getCosClient() : createCosClient();
@@ -146,8 +149,6 @@ public class IBMCOSEndpoint extends ScheduledPollEndpoint implements EndpointSer
cosClient.createBucket(bucketName);
LOG.trace("Bucket created");
}
-
- ServiceHelper.startService(inProgressRepository);
}
@Override