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 2020/06/30 12:28:59 UTC
[camel] 01/05: CAMEL-15191 - AWS2-Kinesis-Firehose: Support for
RecordBatchRequest
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 7a16cc4fad0fccd00264656f1192a41bdf9ebd9a
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jun 30 14:07:39 2020 +0200
CAMEL-15191 - AWS2-Kinesis-Firehose: Support for RecordBatchRequest
---
.../KinesisFirehose2ComponentConfigurer.java | 3 ++
.../KinesisFirehose2EndpointConfigurer.java | 3 ++
.../aws2/firehose/aws2-kinesis-firehose.json | 2 +
.../main/docs/aws2-kinesis-firehose-component.adoc | 6 ++-
.../firehose/KinesisFirehose2Configuration.java | 12 ++++-
.../aws2/firehose/KinesisFirehose2Constants.java | 1 +
...stants.java => KinesisFirehose2Operations.java} | 4 +-
.../aws2/firehose/KinesisFirehose2Producer.java | 53 ++++++++++++++++++++++
.../KinesisFirehoseComponentIntegrationTest.java | 24 +++++++++-
...Aws2KinesisFirehoseComponentBuilderFactory.java | 15 ++++++
.../KinesisFirehose2EndpointBuilderFactory.java | 35 ++++++++++++++
11 files changed, 152 insertions(+), 6 deletions(-)
diff --git a/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentConfigurer.java b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentConfigurer.java
index 177dd52..97ea445 100644
--- a/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentConfigurer.java
+++ b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentConfigurer.java
@@ -35,6 +35,7 @@ public class KinesisFirehose2ComponentConfigurer extends PropertyConfigurerSuppo
case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "operation": getOrCreateConfiguration(target).setOperation(property(camelContext, org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations.class, value)); return true;
case "proxyhost":
case "proxyHost": getOrCreateConfiguration(target).setProxyHost(property(camelContext, java.lang.String.class, value)); return true;
case "proxyport":
@@ -56,6 +57,7 @@ public class KinesisFirehose2ComponentConfigurer extends PropertyConfigurerSuppo
answer.put("basicPropertyBinding", boolean.class);
answer.put("configuration", org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration.class);
answer.put("lazyStartProducer", boolean.class);
+ answer.put("operation", org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations.class);
answer.put("proxyHost", java.lang.String.class);
answer.put("proxyPort", java.lang.Integer.class);
answer.put("proxyProtocol", software.amazon.awssdk.core.Protocol.class);
@@ -77,6 +79,7 @@ public class KinesisFirehose2ComponentConfigurer extends PropertyConfigurerSuppo
case "configuration": return target.getConfiguration();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
+ case "operation": return getOrCreateConfiguration(target).getOperation();
case "proxyhost":
case "proxyHost": return getOrCreateConfiguration(target).getProxyHost();
case "proxyport":
diff --git a/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2EndpointConfigurer.java b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2EndpointConfigurer.java
index 36ac274..8359a42 100644
--- a/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2EndpointConfigurer.java
+++ b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2EndpointConfigurer.java
@@ -27,6 +27,7 @@ public class KinesisFirehose2EndpointConfigurer extends PropertyConfigurerSuppor
case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "operation": target.getConfiguration().setOperation(property(camelContext, org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations.class, value)); return true;
case "proxyhost":
case "proxyHost": target.getConfiguration().setProxyHost(property(camelContext, java.lang.String.class, value)); return true;
case "proxyport":
@@ -48,6 +49,7 @@ public class KinesisFirehose2EndpointConfigurer extends PropertyConfigurerSuppor
answer.put("amazonKinesisFirehoseClient", software.amazon.awssdk.services.firehose.FirehoseClient.class);
answer.put("basicPropertyBinding", boolean.class);
answer.put("lazyStartProducer", boolean.class);
+ answer.put("operation", org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations.class);
answer.put("proxyHost", java.lang.String.class);
answer.put("proxyPort", java.lang.Integer.class);
answer.put("proxyProtocol", software.amazon.awssdk.core.Protocol.class);
@@ -69,6 +71,7 @@ public class KinesisFirehose2EndpointConfigurer extends PropertyConfigurerSuppor
case "basicPropertyBinding": return target.isBasicPropertyBinding();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
+ case "operation": return target.getConfiguration().getOperation();
case "proxyhost":
case "proxyHost": return target.getConfiguration().getProxyHost();
case "proxyport":
diff --git a/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json b/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
index 3b0df6b..f4df3dd 100644
--- a/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
+++ b/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
@@ -24,6 +24,7 @@
"amazonKinesisFirehoseClient": { "kind": "property", "displayName": "Amazon Kinesis Firehose Client", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.firehose.FirehoseClient", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "Amazon Kinesis Firehose client to use for all requests [...]
"configuration": { "kind": "property", "displayName": "Configuration", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "deprecated": false, "secret": false, "description": "Component configuration" },
"lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the r [...]
+ "operation": { "kind": "property", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations", "enum": [ "sendBatchRecord" ], "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "The operation to do in case the user don't [...]
"proxyHost": { "kind": "property", "displayName": "Proxy Host", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "To define a proxy host when instantiating the Kinesis Firehose client" },
"proxyPort": { "kind": "property", "displayName": "Proxy Port", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "To define a proxy port when instantiating the Kinesis Firehose client" },
"proxyProtocol": { "kind": "property", "displayName": "Proxy Protocol", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "To define a proxy protocol when instantiating the [...]
@@ -36,6 +37,7 @@
"streamName": { "kind": "path", "displayName": "Stream Name", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "Name of the stream" },
"amazonKinesisFirehoseClient": { "kind": "parameter", "displayName": "Amazon Kinesis Firehose Client", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.firehose.FirehoseClient", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "Amazon Kinesis Firehose client to use for all requests [...]
"lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the [...]
+ "operation": { "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "producer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations", "enum": [ "sendBatchRecord" ], "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "The operation to do in case the user don't [...]
"proxyHost": { "kind": "parameter", "displayName": "Proxy Host", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "To define a proxy host when instantiating the Kinesis Firehose client" },
"proxyPort": { "kind": "parameter", "displayName": "Proxy Port", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "To define a proxy port when instantiating the Kinesis Firehose client" },
"proxyProtocol": { "kind": "parameter", "displayName": "Proxy Protocol", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "To define a proxy protocol when instantiating the [...]
diff --git a/components/camel-aws2-kinesis/src/main/docs/aws2-kinesis-firehose-component.adoc b/components/camel-aws2-kinesis/src/main/docs/aws2-kinesis-firehose-component.adoc
index 1b7b60c..f423ce9 100644
--- a/components/camel-aws2-kinesis/src/main/docs/aws2-kinesis-firehose-component.adoc
+++ b/components/camel-aws2-kinesis/src/main/docs/aws2-kinesis-firehose-component.adoc
@@ -41,7 +41,7 @@ The stream needs to be created prior to it being used. +
// component options: START
-The AWS 2 Kinesis Firehose component supports 10 options, which are listed below.
+The AWS 2 Kinesis Firehose component supports 11 options, which are listed below.
@@ -51,6 +51,7 @@ The AWS 2 Kinesis Firehose component supports 10 options, which are listed below
| *amazonKinesisFirehoseClient* (producer) | Amazon Kinesis Firehose client to use for all requests for this endpoint | | FirehoseClient
| *configuration* (producer) | Component configuration | | KinesisFirehose2Configuration
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
+| *operation* (producer) | The operation to do in case the user don't want to send only a record. The value can be one of: sendBatchRecord | | KinesisFirehose2Operations
| *proxyHost* (producer) | To define a proxy host when instantiating the Kinesis Firehose client | | String
| *proxyPort* (producer) | To define a proxy port when instantiating the Kinesis Firehose client | | Integer
| *proxyProtocol* (producer) | To define a proxy protocol when instantiating the Kinesis Firehose client. The value can be one of: HTTP, HTTPS | HTTPS | Protocol
@@ -86,7 +87,7 @@ with the following path and query parameters:
|===
-=== Query Parameters (10 parameters):
+=== Query Parameters (11 parameters):
[width="100%",cols="2,5,^1,2",options="header"]
@@ -94,6 +95,7 @@ with the following path and query parameters:
| Name | Description | Default | Type
| *amazonKinesisFirehoseClient* (producer) | Amazon Kinesis Firehose client to use for all requests for this endpoint | | FirehoseClient
| *lazyStartProducer* (producer) | Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during starting and cause the route to fail being started. By deferring this startup to be lazy then the startup failure can be handled during routing messages via Camel's routing error handlers. Beware that when the first message is processed then creating and [...]
+| *operation* (producer) | The operation to do in case the user don't want to send only a record. The value can be one of: sendBatchRecord | | KinesisFirehose2Operations
| *proxyHost* (producer) | To define a proxy host when instantiating the Kinesis Firehose client | | String
| *proxyPort* (producer) | To define a proxy port when instantiating the Kinesis Firehose client | | Integer
| *proxyProtocol* (producer) | To define a proxy protocol when instantiating the Kinesis Firehose client. The value can be one of: HTTP, HTTPS | HTTPS | Protocol
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
index 04c8f8a..03a5693 100644
--- a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
@@ -45,6 +45,8 @@ public class KinesisFirehose2Configuration implements Cloneable {
private String proxyHost;
@UriParam(description = "To define a proxy port when instantiating the Kinesis Firehose client")
private Integer proxyPort;
+ @UriParam(label = "producer", description = "The operation to do in case the user don't want to send only a record")
+ private KinesisFirehose2Operations operation;
public void setAmazonKinesisFirehoseClient(FirehoseClient client) {
this.amazonKinesisFirehoseClient = client;
@@ -110,11 +112,19 @@ public class KinesisFirehose2Configuration implements Cloneable {
this.proxyPort = proxyPort;
}
+ public KinesisFirehose2Operations getOperation() {
+ return operation;
+ }
+
+ public void setOperation(KinesisFirehose2Operations operation) {
+ this.operation = operation;
+ }
+
// *************************************************
//
// *************************************************
- public KinesisFirehose2Configuration copy() {
+ public KinesisFirehose2Configuration copy() {
try {
return (KinesisFirehose2Configuration)super.clone();
} catch (CloneNotSupportedException e) {
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java
index 72ed001..da68897 100644
--- a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java
@@ -19,4 +19,5 @@ package org.apache.camel.component.aws2.firehose;
public interface KinesisFirehose2Constants {
String RECORD_ID = "CamelAwsKinesisFirehoseRecordId";
+ String KINESIS_FIREHOSE_OPERATION = "CamelAwsKinesisFirehoseOperation";
}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Operations.java
similarity index 89%
copy from components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java
copy to components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Operations.java
index 72ed001..3fee7ae 100644
--- a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Operations.java
@@ -16,7 +16,7 @@
*/
package org.apache.camel.component.aws2.firehose;
-public interface KinesisFirehose2Constants {
+public enum KinesisFirehose2Operations {
- String RECORD_ID = "CamelAwsKinesisFirehoseRecordId";
+ sendBatchRecord
}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
index 8111ffc..26b8abc 100644
--- a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
@@ -17,13 +17,18 @@
package org.apache.camel.component.aws2.firehose;
import java.nio.ByteBuffer;
+import java.util.Collection;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.firehose.FirehoseClient;
+import software.amazon.awssdk.services.firehose.model.PutRecordBatchRequest;
+import software.amazon.awssdk.services.firehose.model.PutRecordBatchResponse;
import software.amazon.awssdk.services.firehose.model.PutRecordRequest;
import software.amazon.awssdk.services.firehose.model.PutRecordResponse;
import software.amazon.awssdk.services.firehose.model.Record;
@@ -43,6 +48,38 @@ public class KinesisFirehose2Producer extends DefaultProducer {
@Override
public void process(Exchange exchange) throws Exception {
+ KinesisFirehose2Operations operation = determineOperation(exchange);
+ if (ObjectHelper.isEmpty(operation)) {
+ processSingleRecord(exchange);
+ } else {
+ switch (operation) {
+ case sendBatchRecord:
+ sendBatchRecord(getClient(), exchange);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported operation");
+ }
+ }
+ }
+
+ private void sendBatchRecord(FirehoseClient client, Exchange exchange) {
+ if (exchange.getIn().getBody() instanceof Iterable) {
+ Iterable c = exchange.getIn().getBody(Iterable.class);
+ PutRecordBatchRequest.Builder batchRequest = PutRecordBatchRequest.builder();
+ batchRequest.deliveryStreamName(getEndpoint().getConfiguration().getStreamName());
+ batchRequest.records((Collection<Record>) c);
+ PutRecordBatchResponse result = client.putRecordBatch(batchRequest.build());
+ Message message = getMessageForResponse(exchange);
+ message.setBody(result);
+ } else {
+ PutRecordBatchRequest req = exchange.getIn().getBody(PutRecordBatchRequest.class);
+ PutRecordBatchResponse result = client.putRecordBatch(req);
+ Message message = getMessageForResponse(exchange);
+ message.setBody(result);
+ }
+ }
+
+ public void processSingleRecord(final Exchange exchange) {
PutRecordRequest request = createRequest(exchange);
LOG.trace("Sending request [{}] from exchange [{}]...", request, exchange);
PutRecordResponse putRecordResult = getEndpoint().getClient().putRecord(request);
@@ -65,4 +102,20 @@ public class KinesisFirehose2Producer extends DefaultProducer {
public static Message getMessageForResponse(final Exchange exchange) {
return exchange.getMessage();
}
+
+ protected FirehoseClient getClient() {
+ return getEndpoint().getClient();
+ }
+
+ protected KinesisFirehose2Configuration getConfiguration() {
+ return getEndpoint().getConfiguration();
+ }
+
+ private KinesisFirehose2Operations determineOperation(Exchange exchange) {
+ KinesisFirehose2Operations operation = exchange.getIn().getHeader(KinesisFirehose2Constants.KINESIS_FIREHOSE_OPERATION, KinesisFirehose2Operations.class);
+ if (operation == null) {
+ operation = getConfiguration().getOperation();
+ }
+ return operation;
+ }
}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
index 80e349d..8b39e6f 100644
--- a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
@@ -25,10 +25,17 @@ import org.apache.camel.component.aws2.firehose.KinesisFirehose2Constants;
import org.apache.camel.test.junit5.CamelTestSupport;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
+
+import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.firehose.FirehoseClient;
+import software.amazon.awssdk.services.firehose.model.Record;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+
@Disabled("Must be manually tested.")
public class KinesisFirehoseComponentIntegrationTest extends CamelTestSupport {
@@ -44,13 +51,28 @@ public class KinesisFirehoseComponentIntegrationTest extends CamelTestSupport {
});
assertNotNull(exchange.getIn().getHeader(KinesisFirehose2Constants.RECORD_ID));
}
+
+ @Test
+ public void testFirehoseBatchRouting() throws Exception {
+ Exchange exchange = template.send("direct:start", ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ List<Record> recs = new ArrayList<Record>();
+ Record rec = Record.builder().data(SdkBytes.fromString(("Test1"), Charset.defaultCharset())).build();
+ Record rec1 = Record.builder().data(SdkBytes.fromString(("Test2"), Charset.defaultCharset())).build();
+ recs.add(rec);
+ recs.add(rec1);
+ exchange.getIn().setBody(recs);
+ }
+ });
+ assertNotNull(exchange.getIn().getBody());
+ }
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:start").to("aws2-kinesis-firehose://mystream?amazonKinesisFirehoseClient=#FirehoseClient");
+ from("direct:start").to("aws2-kinesis-firehose://cc?amazonKinesisFirehoseClient=#FirehoseClient&operation=sendBatchRecord");
}
};
}
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisFirehoseComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisFirehoseComponentBuilderFactory.java
index b65e853..1db48df 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisFirehoseComponentBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisFirehoseComponentBuilderFactory.java
@@ -98,6 +98,20 @@ public interface Aws2KinesisFirehoseComponentBuilderFactory {
return this;
}
/**
+ * The operation to do in case the user don't want to send only a
+ * record.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations</code> type.
+ *
+ * Group: producer
+ */
+ default Aws2KinesisFirehoseComponentBuilder operation(
+ org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations operation) {
+ doSetProperty("operation", operation);
+ return this;
+ }
+ /**
* To define a proxy host when instantiating the Kinesis Firehose
* client.
*
@@ -218,6 +232,7 @@ public interface Aws2KinesisFirehoseComponentBuilderFactory {
case "amazonKinesisFirehoseClient": getOrCreateConfiguration((KinesisFirehose2Component) component).setAmazonKinesisFirehoseClient((software.amazon.awssdk.services.firehose.FirehoseClient) value); return true;
case "configuration": ((KinesisFirehose2Component) component).setConfiguration((org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration) value); return true;
case "lazyStartProducer": ((KinesisFirehose2Component) component).setLazyStartProducer((boolean) value); return true;
+ case "operation": getOrCreateConfiguration((KinesisFirehose2Component) component).setOperation((org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations) value); return true;
case "proxyHost": getOrCreateConfiguration((KinesisFirehose2Component) component).setProxyHost((java.lang.String) value); return true;
case "proxyPort": getOrCreateConfiguration((KinesisFirehose2Component) component).setProxyPort((java.lang.Integer) value); return true;
case "proxyProtocol": getOrCreateConfiguration((KinesisFirehose2Component) component).setProxyProtocol((software.amazon.awssdk.core.Protocol) value); return true;
diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KinesisFirehose2EndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KinesisFirehose2EndpointBuilderFactory.java
index 5132c82..0cae392 100644
--- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KinesisFirehose2EndpointBuilderFactory.java
+++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/KinesisFirehose2EndpointBuilderFactory.java
@@ -112,6 +112,33 @@ public interface KinesisFirehose2EndpointBuilderFactory {
return this;
}
/**
+ * The operation to do in case the user don't want to send only a
+ * record.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations</code> type.
+ *
+ * Group: producer
+ */
+ default KinesisFirehose2EndpointBuilder operation(
+ KinesisFirehose2Operations operation) {
+ doSetProperty("operation", operation);
+ return this;
+ }
+ /**
+ * The operation to do in case the user don't want to send only a
+ * record.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations</code> type.
+ *
+ * Group: producer
+ */
+ default KinesisFirehose2EndpointBuilder operation(String operation) {
+ doSetProperty("operation", operation);
+ return this;
+ }
+ /**
* To define a proxy host when instantiating the Kinesis Firehose
* client.
*
@@ -284,6 +311,14 @@ public interface KinesisFirehose2EndpointBuilderFactory {
}
/**
+ * Proxy enum for
+ * <code>org.apache.camel.component.aws2.firehose.KinesisFirehose2Operations</code> enum.
+ */
+ enum KinesisFirehose2Operations {
+ sendBatchRecord;
+ }
+
+ /**
* Proxy enum for <code>software.amazon.awssdk.core.Protocol</code> enum.
*/
enum Protocol {