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 2021/04/08 06:40:40 UTC
[camel] 02/07: CAMEL-16469 - Camel-AWS2-S3 - Streaming upload:
restart from the last index when using the progressive naming strategy
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 5e2c5a8b6664015adb95800c3ab62d9c27a7f06a
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Apr 8 07:51:39 2021 +0200
CAMEL-16469 - Camel-AWS2-S3 - Streaming upload: restart from the last index when using the progressive naming strategy
---
.../camel/component/aws2/s3/AWS2S3Configuration.java | 20 +++++++++++++++++---
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 6 +++++-
2 files changed, 22 insertions(+), 4 deletions(-)
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
index 0f48a66..2d2782e 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Configuration.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.aws2.s3;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.aws2.s3.stream.AWSS3NamingStrategyEnum;
+import org.apache.camel.component.aws2.s3.stream.AWSS3RestartingPolicyEnum;
import org.apache.camel.spi.Metadata;
import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriParams;
@@ -121,6 +122,8 @@ public class AWS2S3Configuration implements Cloneable {
private AWSS3NamingStrategyEnum namingStrategy = AWSS3NamingStrategyEnum.progressive;
@UriParam(label = "producer")
private long streamingUploadTimeout;
+ @UriParam(defaultValue = "override", label = "producer")
+ private AWSS3RestartingPolicyEnum restartingPolicy = AWSS3RestartingPolicyEnum.override;
public long getPartSize() {
return partSize;
@@ -602,7 +605,7 @@ public class AWS2S3Configuration implements Cloneable {
}
/**
- * The number of messages composing a batch in stream mode
+ * The number of messages composing a batch in streaming upload mode
*/
public void setBatchMessageNumber(int batchMessageNumber) {
this.batchMessageNumber = batchMessageNumber;
@@ -613,7 +616,7 @@ public class AWS2S3Configuration implements Cloneable {
}
/**
- * The batch size (in bytes) in stream mode
+ * The batch size (in bytes) in streaming upload mode
*/
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
@@ -624,7 +627,7 @@ public class AWS2S3Configuration implements Cloneable {
}
/**
- * The naming strategy to use in stream mode
+ * The naming strategy to use in streaming upload mode
*/
public void setNamingStrategy(AWSS3NamingStrategyEnum namingStrategy) {
this.namingStrategy = namingStrategy;
@@ -641,6 +644,17 @@ public class AWS2S3Configuration implements Cloneable {
this.streamingUploadTimeout = streamingUploadTimeout;
}
+ public AWSS3RestartingPolicyEnum getRestartingPolicy() {
+ return restartingPolicy;
+ }
+
+ /**
+ * The restarting policy to use in streaming upload mode
+ */
+ public void setRestartingPolicy(AWSS3RestartingPolicyEnum restartingPolicy) {
+ this.restartingPolicy = restartingPolicy;
+ }
+
public AWS2S3Configuration copy() {
try {
return (AWS2S3Configuration) super.clone();
diff --git a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
index e83719b..c9ce9d9 100644
--- a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
+++ b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadProducer.java
@@ -76,7 +76,9 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
getConfiguration().getStreamingUploadTimeout(), getConfiguration().getStreamingUploadTimeout(),
TimeUnit.MILLISECONDS);
}
- setStartingPart();
+ if (getConfiguration().getRestartingPolicy().equals(AWSS3RestartingPolicyEnum.lastPart)) {
+ setStartingPart();
+ }
}
@Override
@@ -291,6 +293,8 @@ public class AWS2S3StreamUploadProducer extends DefaultProducer {
part.getAndSet(1);
}
}
+ } else {
+ LOG.info("lastPart restarting policy can be used only with progressive naming strategy");
}
}