You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2023/12/07 10:09:45 UTC
(camel) 07/15: CAMEL-20189: camel-aws: Force marking the consumer ready sooner, in case downloading big files takes too long, causing readiness check to timeout and fail.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch ready
in repository https://gitbox.apache.org/repos/asf/camel.git
commit da023c20af4abc103763cf8c3f28001b1834bbbb
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Dec 7 10:40:45 2023 +0100
CAMEL-20189: camel-aws: Force marking the consumer ready sooner, in case downloading big files takes too long, causing readiness check to timeout and fail.
---
.../apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java | 3 +++
.../apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java | 5 +++++
.../org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java | 3 +++
.../camel/component/aws2/redshift/data/RedshiftData2Producer.java | 2 --
.../main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java | 4 ++++
.../main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java | 3 +++
6 files changed, 18 insertions(+), 2 deletions(-)
diff --git a/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java b/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java
index 1a54b29cee9..1740d05c0bf 100644
--- a/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java
+++ b/components/camel-aws/camel-aws-cloudtrail/src/main/java/org/apache/camel/component/aws/cloudtrail/CloudtrailConsumer.java
@@ -63,6 +63,9 @@ public class CloudtrailConsumer extends ScheduledBatchPollingConsumer {
LookupEventsResponse response = getClient().lookupEvents(eventsRequestBuilder.build());
+ // okay we have some response from aws so lets mark the consumer as ready
+ forceConsumerAsReady();
+
if (!response.events().isEmpty()) {
lastTime = response.events().get(0).eventTime();
}
diff --git a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java
index 62491771907..9e3aa959d0a 100644
--- a/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java
+++ b/components/camel-aws/camel-aws2-ddb/src/main/java/org/apache/camel/component/aws2/ddbstream/Ddb2StreamConsumer.java
@@ -55,7 +55,12 @@ public class Ddb2StreamConsumer extends ScheduledBatchPollingConsumer {
@Override
protected int poll() throws Exception {
int processedExchangeCount = 0;
+
Map<String, String> shardIterators = shardIteratorHandler.getShardIterators();
+
+ // okay we have some response from azure so lets mark the consumer as ready
+ forceConsumerAsReady();
+
for (Entry<String, String> shardIteratorEntry : shardIterators.entrySet()) {
int limitPerRecordsRequest = Math.max(1,
getEndpoint().getConfiguration().getMaxResultsPerRequest() / shardIterators.size());
diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
index 87178332e51..0e750f8268c 100644
--- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
+++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -113,6 +113,9 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R
});
}
+ // okay we have some response from aws so lets mark the consumer as ready
+ forceConsumerAsReady();
+
return processedExchangeCount.get();
}
diff --git a/components/camel-aws/camel-aws2-redshift/src/main/java/org/apache/camel/component/aws2/redshift/data/RedshiftData2Producer.java b/components/camel-aws/camel-aws2-redshift/src/main/java/org/apache/camel/component/aws2/redshift/data/RedshiftData2Producer.java
index 477a3c77f20..5f7e1c18052 100644
--- a/components/camel-aws/camel-aws2-redshift/src/main/java/org/apache/camel/component/aws2/redshift/data/RedshiftData2Producer.java
+++ b/components/camel-aws/camel-aws2-redshift/src/main/java/org/apache/camel/component/aws2/redshift/data/RedshiftData2Producer.java
@@ -22,8 +22,6 @@ import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
-import org.apache.camel.health.HealthCheck;
-import org.apache.camel.health.WritableHealthCheckRepository;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.URISupport;
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
index 5cd358cf8b8..ff8234de3a1 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
@@ -160,6 +160,10 @@ public class AWS2S3Consumer extends ScheduledBatchPollingConsumer {
exchanges = createExchanges(listObjects.contents());
}
+
+ // okay we have some response from azure so lets mark the consumer as ready
+ forceConsumerAsReady();
+
return processBatch(CastUtils.cast(exchanges));
}
diff --git a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
index 1981c441b23..b6f31b371f5 100644
--- a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
+++ b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
@@ -122,6 +122,9 @@ public class Sqs2Consumer extends ScheduledBatchPollingConsumer {
LOG.trace("Received {} messages", messageResult.messages().size());
}
+ // okay we have some response from aws so lets mark the consumer as ready
+ forceConsumerAsReady();
+
Queue<Exchange> exchanges = createExchanges(messageResult.messages());
return processBatch(CastUtils.cast(exchanges));
}