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/02/27 17:27:40 UTC
[camel] 01/13: CAMEL-14520 - Create an AWS-Kinesis component based
on SDK v2
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 056e6f7886aca7ed202bea0be5f74e12e6e4e67a
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Feb 27 16:06:52 2020 +0100
CAMEL-14520 - Create an AWS-Kinesis component based on SDK v2
---
components/camel-aws2-kinesis/pom.xml | 91 ++++++++
.../KinesisFirehose2ComponentConfigurer.java | 33 +++
.../KinesisFirehose2EndpointConfigurer.java | 41 ++++
.../aws2/kinesis/Kinesis2ComponentConfigurer.java | 35 +++
.../aws2/kinesis/Kinesis2EndpointConfigurer.java | 86 ++++++++
.../aws2/kinesis/RecordStringConverterLoader.java | 40 ++++
.../services/org/apache/camel/TypeConverterLoader | 2 +
.../services/org/apache/camel/component.properties | 7 +
.../org/apache/camel/component/aws2-kinesis | 2 +
.../apache/camel/component/aws2-kinesis-firehose | 2 +
.../apache/camel/configurer/aws2-kinesis-component | 2 +
.../apache/camel/configurer/aws2-kinesis-endpoint | 2 +
.../configurer/aws2-kinesis-firehose-component | 2 +
.../configurer/aws2-kinesis-firehose-endpoint | 2 +
.../aws2/firehose/aws2-kinesis-firehose.json | 42 ++++
.../camel/component/aws2/kinesis/aws2-kinesis.json | 67 ++++++
.../src/main/docs/aws-kinesis-component.adoc | 240 +++++++++++++++++++++
.../main/docs/aws-kinesis-firehose-component.adoc | 179 +++++++++++++++
.../aws2/firehose/KinesisFirehose2Component.java | 120 +++++++++++
...KinesisFirehose2ComponentVerifierExtension.java | 88 ++++++++
.../firehose/KinesisFirehose2Configuration.java | 126 +++++++++++
.../aws2/firehose/KinesisFirehose2Constants.java | 22 ++
.../aws2/firehose/KinesisFirehose2Endpoint.java | 123 +++++++++++
.../aws2/firehose/KinesisFirehose2Producer.java | 70 ++++++
.../component/aws2/kinesis/Kinesis2Component.java | 118 ++++++++++
.../Kinesis2ComponentVerifierExtension.java | 86 ++++++++
.../aws2/kinesis/Kinesis2Configuration.java | 179 +++++++++++++++
.../component/aws2/kinesis/Kinesis2Constants.java | 29 +++
.../component/aws2/kinesis/Kinesis2Consumer.java | 165 ++++++++++++++
.../component/aws2/kinesis/Kinesis2Endpoint.java | 141 ++++++++++++
.../component/aws2/kinesis/Kinesis2Producer.java | 67 ++++++
.../kinesis/Kinesis2ShardClosedStrategyEnum.java | 24 +++
.../aws2/kinesis/ReachedClosedStatusException.java | 30 +++
.../aws2/kinesis/RecordStringConverter.java | 48 +++++
.../KinesisFirehoseComponentConfigurationTest.java | 80 +++++++
...esisFirehoseComponentVerifierExtensionTest.java | 71 ++++++
.../aws2/firehose/KinesisFirehoseEndpointTest.java | 72 +++++++
.../KinesisFirehoseComponentIntegrationTest.java | 58 +++++
.../kinesis/KinesisComponentConfigurationTest.java | 80 +++++++
.../KinesisComponentVerifierExtensionTest.java | 71 ++++++
.../KinesisConsumerClosedShardWithFailTest.java | 106 +++++++++
.../KinesisConsumerClosedShardWithSilentTest.java | 222 +++++++++++++++++++
.../aws2/kinesis/KinesisEndpointTest.java | 111 ++++++++++
.../aws2/kinesis/RecordStringConverterTest.java | 42 ++++
.../KinesisComponentIntegrationTest.java | 91 ++++++++
.../src/test/resources/log4j2.properties | 28 +++
core/camel-allcomponents/pom.xml | 4 +
.../component/ComponentsBuilderFactory.java | 24 +++
.../dsl/Aws2KinesisComponentBuilderFactory.java | 181 ++++++++++++++++
...Aws2KinesisFirehoseComponentBuilderFactory.java | 164 ++++++++++++++
.../src/generated/resources/metadata.json | 40 ++++
51 files changed, 3756 insertions(+)
diff --git a/components/camel-aws2-kinesis/pom.xml b/components/camel-aws2-kinesis/pom.xml
new file mode 100644
index 0000000..6245cae
--- /dev/null
+++ b/components/camel-aws2-kinesis/pom.xml
@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>3.2.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-aws2-kinesis</artifactId>
+ <packaging>jar</packaging>
+
+ <name>Camel :: AWS2 Kinesis</name>
+ <description>A Camel Amazon Kinesis Web Service Component Version 2</description>
+
+ <properties>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-support</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>kinesis</artifactId>
+ <version>${aws-java-sdk2-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>firehose</artifactId>
+ <version>${aws-java-sdk2-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache-client</artifactId>
+ <version>${aws-java-sdk2-version}</version>
+ </dependency>
+
+ <!-- for testing -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
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
new file mode 100644
index 0000000..da6f21f
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentConfigurer.java
@@ -0,0 +1,33 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.aws2.firehose;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class KinesisFirehose2ComponentConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ KinesisFirehose2Component target = (KinesisFirehose2Component) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "accesskey":
+ case "accessKey": target.setAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "region": target.setRegion(property(camelContext, java.lang.String.class, value)); return true;
+ case "secretkey":
+ case "secretKey": target.setSecretKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "basicpropertybinding":
+ case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
+ case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration.class, value)); return true;
+ default: return false;
+ }
+ }
+
+}
+
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
new file mode 100644
index 0000000..29129de
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2EndpointConfigurer.java
@@ -0,0 +1,41 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.aws2.firehose;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class KinesisFirehose2EndpointConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ KinesisFirehose2Endpoint target = (KinesisFirehose2Endpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amazonkinesisfirehoseclient":
+ case "amazonKinesisFirehoseClient": target.getConfiguration().setAmazonKinesisFirehoseClient(property(camelContext, software.amazon.awssdk.services.firehose.FirehoseClient.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "proxyhost":
+ case "proxyHost": target.getConfiguration().setProxyHost(property(camelContext, java.lang.String.class, value)); return true;
+ case "proxyport":
+ case "proxyPort": target.getConfiguration().setProxyPort(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "proxyprotocol":
+ case "proxyProtocol": target.getConfiguration().setProxyProtocol(property(camelContext, software.amazon.awssdk.core.Protocol.class, value)); return true;
+ case "region": target.getConfiguration().setRegion(property(camelContext, java.lang.String.class, value)); return true;
+ case "basicpropertybinding":
+ case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
+ case "synchronous": target.setSynchronous(property(camelContext, boolean.class, value)); return true;
+ case "accesskey":
+ case "accessKey": target.getConfiguration().setAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "secretkey":
+ case "secretKey": target.getConfiguration().setSecretKey(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+}
+
diff --git a/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentConfigurer.java b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentConfigurer.java
new file mode 100644
index 0000000..298f2be
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentConfigurer.java
@@ -0,0 +1,35 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class Kinesis2ComponentConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ Kinesis2Component target = (Kinesis2Component) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "accesskey":
+ case "accessKey": target.setAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "region": target.setRegion(property(camelContext, java.lang.String.class, value)); return true;
+ case "secretkey":
+ case "secretKey": target.setSecretKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "basicpropertybinding":
+ case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
+ case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.aws2.kinesis.Kinesis2Configuration.class, value)); return true;
+ default: return false;
+ }
+ }
+
+}
+
diff --git a/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointConfigurer.java b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointConfigurer.java
new file mode 100644
index 0000000..9b4d411
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointConfigurer.java
@@ -0,0 +1,86 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class Kinesis2EndpointConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ Kinesis2Endpoint target = (Kinesis2Endpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "amazonkinesisclient":
+ case "amazonKinesisClient": target.getConfiguration().setAmazonKinesisClient(property(camelContext, software.amazon.awssdk.services.kinesis.KinesisClient.class, value)); return true;
+ case "proxyhost":
+ case "proxyHost": target.getConfiguration().setProxyHost(property(camelContext, java.lang.String.class, value)); return true;
+ case "proxyport":
+ case "proxyPort": target.getConfiguration().setProxyPort(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "proxyprotocol":
+ case "proxyProtocol": target.getConfiguration().setProxyProtocol(property(camelContext, software.amazon.awssdk.core.Protocol.class, value)); return true;
+ case "region": target.getConfiguration().setRegion(property(camelContext, java.lang.String.class, value)); return true;
+ case "bridgeerrorhandler":
+ case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true;
+ case "iteratortype":
+ case "iteratorType": target.getConfiguration().setIteratorType(property(camelContext, software.amazon.awssdk.services.kinesis.model.ShardIteratorType.class, value)); return true;
+ case "maxresultsperrequest":
+ case "maxResultsPerRequest": target.getConfiguration().setMaxResultsPerRequest(property(camelContext, int.class, value)); return true;
+ case "sendemptymessagewhenidle":
+ case "sendEmptyMessageWhenIdle": target.setSendEmptyMessageWhenIdle(property(camelContext, boolean.class, value)); return true;
+ case "sequencenumber":
+ case "sequenceNumber": target.getConfiguration().setSequenceNumber(property(camelContext, java.lang.String.class, value)); return true;
+ case "shardclosed":
+ case "shardClosed": target.getConfiguration().setShardClosed(property(camelContext, org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum.class, value)); return true;
+ case "shardid":
+ case "shardId": target.getConfiguration().setShardId(property(camelContext, java.lang.String.class, value)); return true;
+ case "exceptionhandler":
+ case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true;
+ case "exchangepattern":
+ case "exchangePattern": target.setExchangePattern(property(camelContext, org.apache.camel.ExchangePattern.class, value)); return true;
+ case "pollstrategy":
+ case "pollStrategy": target.setPollStrategy(property(camelContext, org.apache.camel.spi.PollingConsumerPollStrategy.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "basicpropertybinding":
+ case "basicPropertyBinding": target.setBasicPropertyBinding(property(camelContext, boolean.class, value)); return true;
+ case "synchronous": target.setSynchronous(property(camelContext, boolean.class, value)); return true;
+ case "backofferrorthreshold":
+ case "backoffErrorThreshold": target.setBackoffErrorThreshold(property(camelContext, int.class, value)); return true;
+ case "backoffidlethreshold":
+ case "backoffIdleThreshold": target.setBackoffIdleThreshold(property(camelContext, int.class, value)); return true;
+ case "backoffmultiplier":
+ case "backoffMultiplier": target.setBackoffMultiplier(property(camelContext, int.class, value)); return true;
+ case "delay": target.setDelay(property(camelContext, long.class, value)); return true;
+ case "greedy": target.setGreedy(property(camelContext, boolean.class, value)); return true;
+ case "initialdelay":
+ case "initialDelay": target.setInitialDelay(property(camelContext, long.class, value)); return true;
+ case "repeatcount":
+ case "repeatCount": target.setRepeatCount(property(camelContext, long.class, value)); return true;
+ case "runlogginglevel":
+ case "runLoggingLevel": target.setRunLoggingLevel(property(camelContext, org.apache.camel.LoggingLevel.class, value)); return true;
+ case "scheduledexecutorservice":
+ case "scheduledExecutorService": target.setScheduledExecutorService(property(camelContext, java.util.concurrent.ScheduledExecutorService.class, value)); return true;
+ case "scheduler": target.setScheduler(property(camelContext, java.lang.String.class, value)); return true;
+ case "schedulerproperties":
+ case "schedulerProperties": target.setSchedulerProperties(property(camelContext, java.util.Map.class, value)); return true;
+ case "startscheduler":
+ case "startScheduler": target.setStartScheduler(property(camelContext, boolean.class, value)); return true;
+ case "timeunit":
+ case "timeUnit": target.setTimeUnit(property(camelContext, java.util.concurrent.TimeUnit.class, value)); return true;
+ case "usefixeddelay":
+ case "useFixedDelay": target.setUseFixedDelay(property(camelContext, boolean.class, value)); return true;
+ case "accesskey":
+ case "accessKey": target.getConfiguration().setAccessKey(property(camelContext, java.lang.String.class, value)); return true;
+ case "secretkey":
+ case "secretKey": target.getConfiguration().setSecretKey(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+}
+
diff --git a/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/RecordStringConverterLoader.java b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/RecordStringConverterLoader.java
new file mode 100644
index 0000000..d143cc7
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/RecordStringConverterLoader.java
@@ -0,0 +1,40 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.TypeConversionException;
+import org.apache.camel.TypeConverterLoaderException;
+import org.apache.camel.spi.TypeConverterLoader;
+import org.apache.camel.spi.TypeConverterRegistry;
+import org.apache.camel.support.SimpleTypeConverter;
+import org.apache.camel.support.TypeConverterSupport;
+import org.apache.camel.util.DoubleMap;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public final class RecordStringConverterLoader implements TypeConverterLoader {
+
+ public RecordStringConverterLoader() {
+ }
+
+ @Override
+ public void load(TypeConverterRegistry registry) throws TypeConverterLoaderException {
+ try {
+ registerConverters(registry);
+ } catch (Throwable e) {
+ // ignore on load error
+ }
+ }
+
+ private void registerConverters(TypeConverterRegistry registry) {
+ addTypeConverter(registry, java.lang.String.class, software.amazon.awssdk.services.kinesis.model.Record.class, false,
+ (type, exchange, value) -> org.apache.camel.component.aws2.kinesis.RecordStringConverter.toString((software.amazon.awssdk.services.kinesis.model.Record) value));
+ }
+
+ private static void addTypeConverter(TypeConverterRegistry registry, Class<?> toType, Class<?> fromType, boolean allowNull, SimpleTypeConverter.ConversionMethod method) {
+ registry.addTypeConverter(toType, fromType, new SimpleTypeConverter(allowNull, method));
+ }
+
+}
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader
new file mode 100644
index 0000000..71b7c3c
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+org.apache.camel.component.aws2.kinesis.RecordStringConverterLoader
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component.properties b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component.properties
new file mode 100644
index 0000000..7b0f435
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component.properties
@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+components=aws2-kinesis aws2-kinesis-firehose
+groupId=org.apache.camel
+artifactId=camel-aws2-kinesis
+version=3.2.0-SNAPSHOT
+projectName=Camel :: AWS2 Kinesis
+projectDescription=A Camel Amazon Kinesis Web Service Component Version 2
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component/aws2-kinesis b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component/aws2-kinesis
new file mode 100644
index 0000000..ae9f4af
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component/aws2-kinesis
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.aws2.kinesis.Kinesis2Component
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component/aws2-kinesis-firehose b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component/aws2-kinesis-firehose
new file mode 100644
index 0000000..a9be7d4
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/component/aws2-kinesis-firehose
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.aws2.firehose.KinesisFirehose2Component
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-component b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-component
new file mode 100644
index 0000000..dda9a61
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-component
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.aws2.kinesis.Kinesis2ComponentConfigurer
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-endpoint b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-endpoint
new file mode 100644
index 0000000..555b68a
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.aws2.kinesis.Kinesis2EndpointConfigurer
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-firehose-component b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-firehose-component
new file mode 100644
index 0000000..6a3a552
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-firehose-component
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.aws2.firehose.KinesisFirehose2ComponentConfigurer
diff --git a/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-firehose-endpoint b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-firehose-endpoint
new file mode 100644
index 0000000..1865874
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/META-INF/services/org/apache/camel/configurer/aws2-kinesis-firehose-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.aws2.firehose.KinesisFirehose2EndpointConfigurer
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
new file mode 100644
index 0000000..1fc4f82
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/firehose/aws2-kinesis-firehose.json
@@ -0,0 +1,42 @@
+{
+ "component": {
+ "kind": "component",
+ "scheme": "aws2-kinesis-firehose",
+ "extendsScheme": "",
+ "syntax": "aws2-kinesis-firehose:streamName",
+ "title": "AWS 2 Kinesis Firehose",
+ "description": "The aws-kinesis-firehose component is used for producing Amazon's Kinesis Firehose streams.",
+ "label": "cloud,messaging",
+ "deprecated": false,
+ "async": false,
+ "consumerOnly": false,
+ "producerOnly": true,
+ "lenientProperties": false,
+ "javaType": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Component",
+ "firstVersion": "3.2.0",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-aws2-kinesis",
+ "version": "3.2.0-SNAPSHOT"
+ },
+ "componentProperties": {
+ "accessKey": { "kind": "property", "displayName": "Access Key", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Amazon AWS Access Key" },
+ "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 [...]
+ "region": { "kind": "property", "displayName": "Region", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Amazon AWS Region" },
+ "secretKey": { "kind": "property", "displayName": "Secret Key", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Amazon AWS Secret Key" },
+ "basicPropertyBinding": { "kind": "property", "displayName": "Basic Property Binding", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities" },
+ "configuration": { "kind": "property", "displayName": "Configuration", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "deprecated": false, "secret": false, "description": "The AWS Kinesis Firehose default configuration" }
+ },
+ "properties": {
+ "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 [...]
+ "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 [...]
+ "region": { "kind": "parameter", "displayName": "Region", "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": "The region in which Kinesis client needs to work. When using this parameter, the configuration will expect the capitalized name of the reg [...]
+ "basicPropertyBinding": { "kind": "parameter", "displayName": "Basic Property Binding", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities" },
+ "synchronous": { "kind": "parameter", "displayName": "Synchronous", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "description": "Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported)." },
+ "accessKey": { "kind": "parameter", "displayName": "Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "Amazon AWS Access Key" },
+ "secretKey": { "kind": "parameter", "displayName": "Secret Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration", "configurationField": "configuration", "description": "Amazon AWS Secret Key" }
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json b/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
new file mode 100644
index 0000000..8da7d1a
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/generated/resources/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json
@@ -0,0 +1,67 @@
+{
+ "component": {
+ "kind": "component",
+ "scheme": "aws2-kinesis",
+ "extendsScheme": "",
+ "syntax": "aws2-kinesis:streamName",
+ "title": "AWS 2 Kinesis",
+ "description": "The aws-kinesis component is for consuming and producing records from Amazon Kinesis Streams.",
+ "label": "cloud,messaging",
+ "deprecated": false,
+ "async": false,
+ "consumerOnly": false,
+ "producerOnly": false,
+ "lenientProperties": false,
+ "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2Component",
+ "firstVersion": "3.2.0",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-aws2-kinesis",
+ "version": "3.2.0-SNAPSHOT"
+ },
+ "componentProperties": {
+ "accessKey": { "kind": "property", "displayName": "Access Key", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Amazon AWS Access Key" },
+ "region": { "kind": "property", "displayName": "Region", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Amazon AWS Region" },
+ "secretKey": { "kind": "property", "displayName": "Secret Key", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "description": "Amazon AWS Secret Key" },
+ "bridgeErrorHandler": { "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by [...]
+ "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 [...]
+ "basicPropertyBinding": { "kind": "property", "displayName": "Basic Property Binding", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities" },
+ "configuration": { "kind": "property", "displayName": "Configuration", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "deprecated": false, "secret": false, "description": "The AWS S3 default configuration" }
+ },
+ "properties": {
+ "streamName": { "kind": "path", "displayName": "Stream Name", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Name of the stream" },
+ "amazonKinesisClient": { "kind": "parameter", "displayName": "Amazon Kinesis Client", "group": "common", "label": "", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.kinesis.KinesisClient", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon Kinesis client to use for all requests for this endpoint" },
+ "proxyHost": { "kind": "parameter", "displayName": "Proxy Host", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "To define a proxy host when instantiating the Kinesis client" },
+ "proxyPort": { "kind": "parameter", "displayName": "Proxy Port", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "To define a proxy port when instantiating the Kinesis client" },
+ "proxyProtocol": { "kind": "parameter", "displayName": "Proxy Protocol", "group": "common", "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.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "To define a proxy protocol when instantiating the Kinesis cl [...]
+ "region": { "kind": "parameter", "displayName": "Region", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "The region in which Kinesis client needs to work. When using this parameter, the configuration will expect the capitalized name of the region (for ex [...]
+ "bridgeErrorHandler": { "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled b [...]
+ "iteratorType": { "kind": "parameter", "displayName": "Iterator Type", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.kinesis.model.ShardIteratorType", "enum": [ "AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "TRIM_HORIZON", "LATEST", "AT_TIMESTAMP", "null" ], "deprecated": false, "secret": false, "defaultValue": "TRIM_HORIZON", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configura [...]
+ "maxResultsPerRequest": { "kind": "parameter", "displayName": "Max Results Per Request", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "defaultValue": "1", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Maximum number of records that will be fetched in each poll" },
+ "sendEmptyMessageWhenIdle": { "kind": "parameter", "displayName": "Send Empty Message When Idle", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead." },
+ "sequenceNumber": { "kind": "parameter", "displayName": "Sequence Number", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "The sequence number to start polling from. Required if iteratorType is set to AFTER_SEQUENCE_NUMBER or AT_SEQUENCE_NUMBER" },
+ "shardClosed": { "kind": "parameter", "displayName": "Shard Closed", "group": "consumer", "label": "consumer", "required": false, "type": "object", "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum", "enum": [ "ignore", "fail", "silent" ], "deprecated": false, "secret": false, "defaultValue": "ignore", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Define [...]
+ "shardId": { "kind": "parameter", "displayName": "Shard Id", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Defines which shardId in the Kinesis stream to get records from" },
+ "exceptionHandler": { "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with [...]
+ "exchangePattern": { "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut", "InOptionalOut" ], "deprecated": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },
+ "pollStrategy": { "kind": "parameter", "displayName": "Poll Strategy", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false, "secret": false, "description": "A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation before an Exchange h [...]
+ "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 [...]
+ "basicPropertyBinding": { "kind": "parameter", "displayName": "Basic Property Binding", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities" },
+ "synchronous": { "kind": "parameter", "displayName": "Synchronous", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "false", "description": "Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported)." },
+ "backoffErrorThreshold": { "kind": "parameter", "displayName": "Backoff Error Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "description": "The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in." },
+ "backoffIdleThreshold": { "kind": "parameter", "displayName": "Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "description": "The number of subsequent idle polls that should happen before the backoffMultipler should kick-in." },
+ "backoffMultiplier": { "kind": "parameter", "displayName": "Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "secret": false, "description": "To let the scheduled polling consumer backoff if there has been a number of subsequent idles\/errors in a row. The multiplier is then the number of polls that will be skipped before the next actual attempt is happening again. When this option [...]
+ "delay": { "kind": "parameter", "displayName": "Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "500", "description": "Milliseconds before the next poll. You can also specify time values using units, such as 60s (60 seconds), 5m30s (5 minutes and 30 seconds), and 1h (1 hour)." },
+ "greedy": { "kind": "parameter", "displayName": "Greedy", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": false, "description": "If greedy is enabled, then the ScheduledPollConsumer will run immediately again, if the previous run polled 1 or more messages." },
+ "initialDelay": { "kind": "parameter", "displayName": "Initial Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "1000", "description": "Milliseconds before the first poll starts. You can also specify time values using units, such as 60s (60 seconds), 5m30s (5 minutes and 30 seconds), and 1h (1 hour)." },
+ "repeatCount": { "kind": "parameter", "displayName": "Repeat Count", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "secret": false, "defaultValue": "0", "description": "Specifies a maximum limit of number of fires. So if you set it to 1, the scheduler will only fire once. If you set it to 5, it will only fire five times. A value of zero or negative means fire forever." },
+ "runLoggingLevel": { "kind": "parameter", "displayName": "Run Logging Level", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "org.apache.camel.LoggingLevel", "enum": [ "TRACE", "DEBUG", "INFO", "WARN", "ERROR", "OFF" ], "deprecated": false, "secret": false, "defaultValue": "TRACE", "description": "The consumer logs a start\/complete log line when it polls. This option allows you to configure the logging level for that." },
+ "scheduledExecutorService": { "kind": "parameter", "displayName": "Scheduled Executor Service", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.util.concurrent.ScheduledExecutorService", "deprecated": false, "secret": false, "description": "Allows for configuring a custom\/shared thread pool to use for the consumer. By default each consumer has its own single threaded thread pool." },
+ "scheduler": { "kind": "parameter", "displayName": "Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "string", "javaType": "java.lang.String", "enum": [ "none", "spring", "quartz" ], "deprecated": false, "secret": false, "defaultValue": "none", "description": "To use a cron scheduler from either camel-spring or camel-quartz component" },
+ "schedulerProperties": { "kind": "parameter", "displayName": "Scheduler Properties", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.util.Map<java.lang.String, java.lang.Object>", "prefix": "scheduler.", "multiValue": true, "deprecated": false, "secret": false, "description": "To configure additional properties when using a custom scheduler or any of the Quartz, Spring based scheduler." },
+ "startScheduler": { "kind": "parameter", "displayName": "Start Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "description": "Whether the scheduler should be auto started." },
+ "timeUnit": { "kind": "parameter", "displayName": "Time Unit", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.util.concurrent.TimeUnit", "enum": [ "NANOSECONDS", "MICROSECONDS", "MILLISECONDS", "SECONDS", "MINUTES", "HOURS", "DAYS" ], "deprecated": false, "secret": false, "defaultValue": "MILLISECONDS", "description": "Time unit for initialDelay and delay options." },
+ "useFixedDelay": { "kind": "parameter", "displayName": "Use Fixed Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "secret": false, "defaultValue": "true", "description": "Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in JDK for details." },
+ "accessKey": { "kind": "parameter", "displayName": "Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Access Key" },
+ "secretKey": { "kind": "parameter", "displayName": "Secret Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Secret Key" }
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/docs/aws-kinesis-component.adoc b/components/camel-aws2-kinesis/src/main/docs/aws-kinesis-component.adoc
new file mode 100644
index 0000000..b4b246a
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/docs/aws-kinesis-component.adoc
@@ -0,0 +1,240 @@
+[[aws-kinesis-component]]
+= AWS Kinesis Component
+
+*Since Camel 2.17*
+
+// HEADER START
+*Both producer and consumer is supported*
+// HEADER END
+
+The Kinesis component supports receiving messages from and sending messages to Amazon Kinesis
+service.
+
+Prerequisites
+
+You must have a valid Amazon Web Services developer account, and be
+signed up to use Amazon Kinesis. More information are available
+at https://aws.amazon.com/kinesis/[AWS Kinesis]
+
+== URI Format
+
+[source,java]
+-----------------------------------
+aws-kinesis://stream-name[?options]
+-----------------------------------
+
+The stream needs to be created prior to it being used. +
+ You can append query options to the URI in the following format,
+?options=value&option2=value&...
+
+== URI Options
+
+
+// component options: START
+The AWS Kinesis component supports 7 options, which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *accessKey* (common) | Amazon AWS Access Key | | String
+| *region* (common) | Amazon AWS Region | | String
+| *secretKey* (common) | Amazon AWS Secret Key | | String
+| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
+| *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 [...]
+| *basicPropertyBinding* (advanced) | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *configuration* (advanced) | The AWS S3 default configuration | | KinesisConfiguration
+|===
+// component options: END
+
+
+
+
+
+
+
+// endpoint options: START
+The AWS Kinesis endpoint is configured using URI syntax:
+
+----
+aws-kinesis:streamName
+----
+
+with the following path and query parameters:
+
+=== Path Parameters (1 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *streamName* | *Required* Name of the stream | | String
+|===
+
+
+=== Query Parameters (34 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *amazonKinesisClient* (common) | Amazon Kinesis client to use for all requests for this endpoint | | AmazonKinesis
+| *proxyHost* (common) | To define a proxy host when instantiating the Kinesis client | | String
+| *proxyPort* (common) | To define a proxy port when instantiating the Kinesis client | | Integer
+| *proxyProtocol* (common) | To define a proxy protocol when instantiating the Kinesis client. The value can be one of: HTTP, HTTPS | HTTPS | Protocol
+| *region* (common) | The region in which Kinesis client needs to work. When using this parameter, the configuration will expect the capitalized name of the region (for example AP_EAST_1)You'll need to use the name Regions.EU_WEST_1.name() | | String
+| *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean
+| *iteratorType* (consumer) | Defines where in the Kinesis stream to start getting records. The value can be one of: AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER, TRIM_HORIZON, LATEST, AT_TIMESTAMP | TRIM_HORIZON | ShardIteratorType
+| *maxResultsPerRequest* (consumer) | Maximum number of records that will be fetched in each poll | 1 | int
+| *sendEmptyMessageWhenIdle* (consumer) | If the polling consumer did not poll any files, you can enable this option to send an empty message (no body) instead. | false | boolean
+| *sequenceNumber* (consumer) | The sequence number to start polling from. Required if iteratorType is set to AFTER_SEQUENCE_NUMBER or AT_SEQUENCE_NUMBER | | String
+| *shardClosed* (consumer) | Define what will be the behavior in case of shard closed. Possible value are ignore, silent and fail. In case of ignore a message will be logged and the consumer will restart from the beginning,in case of silent there will be no logging and the consumer will start from the beginning,in case of fail a ReachedClosedStateException will be raised. The value can be one of: ignore, fail, silent | ignore | KinesisShardClosedStrategyEnum
+| *shardId* (consumer) | Defines which shardId in the Kinesis stream to get records from | | String
+| *exceptionHandler* (consumer) | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored. | | ExceptionHandler
+| *exchangePattern* (consumer) | Sets the exchange pattern when the consumer creates an exchange. The value can be one of: InOnly, InOut, InOptionalOut | | ExchangePattern
+| *pollStrategy* (consumer) | A pluggable org.apache.camel.PollingConsumerPollingStrategy allowing you to provide your custom implementation to control error handling usually occurred during the poll operation before an Exchange have been created and being routed in Camel. | | PollingConsumerPollStrategy
+| *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 [...]
+| *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+| *backoffErrorThreshold* (scheduler) | The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in. | | int
+| *backoffIdleThreshold* (scheduler) | The number of subsequent idle polls that should happen before the backoffMultipler should kick-in. | | int
+| *backoffMultiplier* (scheduler) | To let the scheduled polling consumer backoff if there has been a number of subsequent idles/errors in a row. The multiplier is then the number of polls that will be skipped before the next actual attempt is happening again. When this option is in use then backoffIdleThreshold and/or backoffErrorThreshold must also be configured. | | int
+| *delay* (scheduler) | Milliseconds before the next poll. You can also specify time values using units, such as 60s (60 seconds), 5m30s (5 minutes and 30 seconds), and 1h (1 hour). | 500 | long
+| *greedy* (scheduler) | If greedy is enabled, then the ScheduledPollConsumer will run immediately again, if the previous run polled 1 or more messages. | false | boolean
+| *initialDelay* (scheduler) | Milliseconds before the first poll starts. You can also specify time values using units, such as 60s (60 seconds), 5m30s (5 minutes and 30 seconds), and 1h (1 hour). | 1000 | long
+| *repeatCount* (scheduler) | Specifies a maximum limit of number of fires. So if you set it to 1, the scheduler will only fire once. If you set it to 5, it will only fire five times. A value of zero or negative means fire forever. | 0 | long
+| *runLoggingLevel* (scheduler) | The consumer logs a start/complete log line when it polls. This option allows you to configure the logging level for that. The value can be one of: TRACE, DEBUG, INFO, WARN, ERROR, OFF | TRACE | LoggingLevel
+| *scheduledExecutorService* (scheduler) | Allows for configuring a custom/shared thread pool to use for the consumer. By default each consumer has its own single threaded thread pool. | | ScheduledExecutorService
+| *scheduler* (scheduler) | To use a cron scheduler from either camel-spring or camel-quartz component. The value can be one of: none, spring, quartz | none | String
+| *schedulerProperties* (scheduler) | To configure additional properties when using a custom scheduler or any of the Quartz, Spring based scheduler. | | Map
+| *startScheduler* (scheduler) | Whether the scheduler should be auto started. | true | boolean
+| *timeUnit* (scheduler) | Time unit for initialDelay and delay options. The value can be one of: NANOSECONDS, MICROSECONDS, MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS | MILLISECONDS | TimeUnit
+| *useFixedDelay* (scheduler) | Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in JDK for details. | true | boolean
+| *accessKey* (security) | Amazon AWS Access Key | | String
+| *secretKey* (security) | Amazon AWS Secret Key | | String
+|===
+// endpoint options: END
+
+
+
+
+
+
+
+Required Kinesis component options
+
+You have to provide the amazonKinesisClient in the
+Registry with proxies and relevant credentials
+configured.
+
+== Batch Consumer
+
+This component implements the Batch Consumer.
+
+This allows you for instance to know how many messages exists in this
+batch and for instance let the Aggregator
+aggregate this number of messages.
+
+== Usage
+
+=== Message headers set by the Kinesis consumer
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|=======================================================================
+|Header |Type |Description
+
+|`CamelAwsKinesisSequenceNumber` |`String` |The sequence number of the record. This is represented as a String as it
+size is not defined by the API. If it is to be used as a numerical type then use
+
+|`CamelAwsKinesisApproximateArrivalTimestamp` |`String` |The time AWS assigned as the arrival time of the record.
+
+|`CamelAwsKinesisPartitionKey` |`String` |Identifies which shard in the stream the data record is assigned to.
+|=======================================================================
+
+=== AmazonKinesis configuration
+
+You will need to create an instance of AmazonKinesisClient and
+bind it to the registry
+
+[source,java]
+--------------------------------------------------------------------------------------------------------------------
+ClientConfiguration clientConfiguration = new ClientConfiguration();
+clientConfiguration.setProxyHost("http://myProxyHost");
+clientConfiguration.setProxyPort(8080);
+
+Region region = Region.getRegion(Regions.fromName(region));
+region.createClient(AmazonKinesisClient.class, null, clientConfiguration);
+// the 'null' here is the AWSCredentialsProvider which defaults to an instance of DefaultAWSCredentialsProviderChain
+
+registry.bind("kinesisClient", client);
+--------------------------------------------------------------------------------------------------------------------
+
+You then have to reference the AmazonKinesisClient in the `amazonKinesisClient` URI option.
+
+[source,java]
+--------------------------------------------------------------------------------------------------------------------
+from("aws-kinesis://mykinesisstream?amazonKinesisClient=#kinesisClient")
+ .to("log:out?showAll=true");
+--------------------------------------------------------------------------------------------------------------------
+
+=== Providing AWS Credentials
+
+It is recommended that the credentials are obtained by using the
+http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html[DefaultAWSCredentialsProviderChain]
+that is the default when creating a new ClientConfiguration instance,
+however, a
+different http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html[AWSCredentialsProvider]
+can be specified when calling createClient(...).
+
+=== Message headers used by the Kinesis producer to write to Kinesis. The producer expects that the message body is a `byte[]`.
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|=======================================================================
+|Header |Type |Description
+
+|`CamelAwsKinesisPartitionKey` |`String` |The PartitionKey to pass to Kinesis to store this record.
+
+|`CamelAwsKinesisSequenceNumber` |`String` |Optional paramter to indicate the sequence number of this record.
+
+|=======================================================================
+
+=== Message headers set by the Kinesis producer on successful storage of a Record
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|=======================================================================
+|Header |Type |Description
+
+|`CamelAwsKinesisSequenceNumber` |`String` |The sequence number of the record, as defined in
+http://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_ResponseSyntax[Response Syntax]
+
+|`CamelAwsKinesisShardId` |`String` |The shard ID of where the Record was stored
+
+
+|=======================================================================
+
+== Automatic detection of AmazonKinesis client in registry
+
+The component is capable of detecting the presence of an AmazonKinesis bean into the registry.
+If it's the only instance of that type it will be used as client and you won't have to define it as uri parameter.
+This may be really useful for smarter configuration of the endpoint.
+
+== Dependencies
+
+Maven users will need to add the following dependency to their pom.xml.
+
+*pom.xml*
+
+[source,xml]
+---------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-aws-kinesis</artifactId>
+ <version>${camel-version}</version>
+</dependency>
+---------------------------------------
+
+where `$\{camel-version\}` must be replaced by the actual version of Camel.
+
+
+include::camel-spring-boot::page$aws-kinesis-starter.adoc[]
diff --git a/components/camel-aws2-kinesis/src/main/docs/aws-kinesis-firehose-component.adoc b/components/camel-aws2-kinesis/src/main/docs/aws-kinesis-firehose-component.adoc
new file mode 100644
index 0000000..1e48f96
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/docs/aws-kinesis-firehose-component.adoc
@@ -0,0 +1,179 @@
+[[aws-kinesis-firehose-component]]
+= AWS Kinesis Firehose Component
+
+*Since Camel 2.19*
+
+// HEADER START
+*Only producer is supported*
+// HEADER END
+
+The Kinesis Firehose component supports sending messages to Amazon Kinesis Firehose service.
+
+Prerequisites
+
+You must have a valid Amazon Web Services developer account, and be
+signed up to use Amazon Kinesis Firehose. More information are available
+at https://aws.amazon.com/kinesis/firehose/[AWS Kinesis Firehose]
+
+== URI Format
+
+[source,java]
+-----------------------------------
+aws-kinesis-firehose://delivery-stream-name[?options]
+-----------------------------------
+
+The stream needs to be created prior to it being used. +
+ You can append query options to the URI in the following format,
+?options=value&option2=value&...
+
+== URI Options
+
+
+// component options: START
+The AWS Kinesis Firehose component supports 6 options, which are listed below.
+
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *accessKey* (producer) | Amazon AWS Access Key | | String
+| *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 [...]
+| *region* (producer) | Amazon AWS Region | | String
+| *secretKey* (producer) | Amazon AWS Secret Key | | String
+| *basicPropertyBinding* (advanced) | Whether the component should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *configuration* (advanced) | The AWS Kinesis Firehose default configuration | | KinesisFirehoseConfiguration
+|===
+// component options: END
+
+
+
+
+
+
+
+// endpoint options: START
+The AWS Kinesis Firehose endpoint is configured using URI syntax:
+
+----
+aws-kinesis-firehose:streamName
+----
+
+with the following path and query parameters:
+
+=== Path Parameters (1 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *streamName* | *Required* Name of the stream | | String
+|===
+
+
+=== Query Parameters (10 parameters):
+
+
+[width="100%",cols="2,5,^1,2",options="header"]
+|===
+| Name | Description | Default | Type
+| *amazonKinesisFirehoseClient* (producer) | Amazon Kinesis Firehose client to use for all requests for this endpoint | | AmazonKinesisFirehose
+| *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 [...]
+| *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
+| *region* (producer) | The region in which Kinesis client needs to work. When using this parameter, the configuration will expect the capitalized name of the region (for example AP_EAST_1)You'll need to use the name Regions.EU_WEST_1.name() | | String
+| *basicPropertyBinding* (advanced) | Whether the endpoint should use basic property binding (Camel 2.x) or the newer property binding with additional capabilities | false | boolean
+| *synchronous* (advanced) | Sets whether synchronous processing should be strictly used, or Camel is allowed to use asynchronous processing (if supported). | false | boolean
+| *accessKey* (security) | Amazon AWS Access Key | | String
+| *secretKey* (security) | Amazon AWS Secret Key | | String
+|===
+// endpoint options: END
+
+
+
+
+
+
+
+Required Kinesis Firehose component options
+
+You have to provide the amazonKinesisClient in the
+Registry with proxies and relevant credentials
+configured.
+
+
+== Usage
+
+=== Amazon Kinesis Firehose configuration
+
+You will need to create an instance of AmazonKinesisClient and
+bind it to the registry
+
+[source,java]
+--------------------------------------------------------------------------------------------------------------------
+ClientConfiguration clientConfiguration = new ClientConfiguration();
+clientConfiguration.setProxyHost("http://myProxyHost");
+clientConfiguration.setProxyPort(8080);
+
+Region region = Region.getRegion(Regions.fromName(region));
+region.createClient(AmazonKinesisClient.class, null, clientConfiguration);
+// the 'null' here is the AWSCredentialsProvider which defaults to an instance of DefaultAWSCredentialsProviderChain
+
+registry.bind("kinesisFirehoseClient", client);
+--------------------------------------------------------------------------------------------------------------------
+
+You then have to reference the AmazonKinesisFirehoseClient in the `amazonKinesisFirehoseClient` URI option.
+
+[source,java]
+--------------------------------------------------------------------------------------------------------------------
+from("aws-kinesis-firehose://mykinesisdeliverystream?amazonKinesisFirehoseClient=#kinesisClient")
+ .to("log:out?showAll=true");
+--------------------------------------------------------------------------------------------------------------------
+
+=== Providing AWS Credentials
+
+It is recommended that the credentials are obtained by using the
+http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html[DefaultAWSCredentialsProviderChain]
+that is the default when creating a new ClientConfiguration instance,
+however, a
+different http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html[AWSCredentialsProvider]
+can be specified when calling createClient(...).
+
+=== Message headers set by the Kinesis producer on successful storage of a Record
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|=======================================================================
+|Header |Type |Description
+
+|`CamelAwsKinesisFirehoseRecordId` |`String` |The record ID, as defined in
+http://docs.aws.amazon.com/firehose/latest/APIReference/API_PutRecord.html#API_PutRecord_ResponseSyntax[Response Syntax]
+
+
+|=======================================================================
+
+== Automatic detection of AmazonKinesisFirehose client in registry
+
+The component is capable of detecting the presence of an AmazonKinesisFirehose bean into the registry.
+If it's the only instance of that type it will be used as client and you won't have to define it as uri parameter.
+This may be really useful for smarter configuration of the endpoint.
+
+== Dependencies
+
+Maven users will need to add the following dependency to their pom.xml.
+
+*pom.xml*
+
+[source,xml]
+---------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-aws-kinesis</artifactId>
+ <version>${camel-version}</version>
+</dependency>
+---------------------------------------
+
+where `$\{camel-version\}` must be replaced by the actual version of Camel.
+
+
+include::camel-spring-boot::page$aws-kinesis-starter.adoc[]
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Component.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Component.java
new file mode 100644
index 0000000..a9a25b9
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Component.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Configuration;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.util.ObjectHelper;
+
+import software.amazon.awssdk.services.firehose.FirehoseClient;
+
+@Component("aws2-kinesis-firehose")
+public class KinesisFirehose2Component extends DefaultComponent {
+
+ @Metadata
+ private String accessKey;
+ @Metadata
+ private String secretKey;
+ @Metadata
+ private String region;
+ @Metadata(label = "advanced")
+ private KinesisFirehose2Configuration configuration;
+
+ public KinesisFirehose2Component() {
+ this(null);
+ }
+
+ public KinesisFirehose2Component(CamelContext context) {
+ super(context);
+
+ registerExtension(new KinesisFirehose2ComponentVerifierExtension());
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ KinesisFirehose2Configuration configuration = this.configuration != null ? this.configuration.copy() : new KinesisFirehose2Configuration();
+ configuration.setStreamName(remaining);
+ KinesisFirehose2Endpoint endpoint = new KinesisFirehose2Endpoint(uri, configuration, this);
+ endpoint.getConfiguration().setAccessKey(accessKey);
+ endpoint.getConfiguration().setSecretKey(secretKey);
+ endpoint.getConfiguration().setRegion(region);
+ setProperties(endpoint, parameters);
+ checkAndSetRegistryClient(configuration);
+ if (configuration.getAmazonKinesisFirehoseClient() == null && (configuration.getAccessKey() == null || configuration.getSecretKey() == null)) {
+ throw new IllegalArgumentException("AmazonKinesisFirehoseClient or accessKey and secretKey must be specified");
+ }
+ return endpoint;
+ }
+
+ public KinesisFirehose2Configuration getConfiguration() {
+ return configuration;
+ }
+
+ /**
+ * The AWS Kinesis Firehose default configuration
+ */
+ public void setConfiguration(KinesisFirehose2Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ /**
+ * Amazon AWS Access Key
+ */
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ /**
+ * Amazon AWS Secret Key
+ */
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getRegion() {
+ return region;
+ }
+
+ /**
+ * Amazon AWS Region
+ */
+ public void setRegion(String region) {
+ this.region = region;
+ }
+
+ private void checkAndSetRegistryClient(KinesisFirehose2Configuration configuration) {
+ Set<FirehoseClient> clients = getCamelContext().getRegistry().findByType(FirehoseClient.class);
+ if (clients.size() == 1) {
+ configuration.setAmazonKinesisFirehoseClient(clients.stream().findFirst().get());
+ }
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentVerifierExtension.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentVerifierExtension.java
new file mode 100644
index 0000000..e362b5f
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2ComponentVerifierExtension.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import java.util.Map;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.firehose.FirehoseClient;
+import software.amazon.awssdk.services.firehose.FirehoseClientBuilder;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.KinesisClientBuilder;
+
+import org.apache.camel.component.extension.verifier.DefaultComponentVerifierExtension;
+import org.apache.camel.component.extension.verifier.ResultBuilder;
+import org.apache.camel.component.extension.verifier.ResultErrorBuilder;
+import org.apache.camel.component.extension.verifier.ResultErrorHelper;
+
+public class KinesisFirehose2ComponentVerifierExtension extends DefaultComponentVerifierExtension {
+
+ public KinesisFirehose2ComponentVerifierExtension() {
+ this("aws2-kinesis-firehose");
+ }
+
+ public KinesisFirehose2ComponentVerifierExtension(String scheme) {
+ super(scheme);
+ }
+
+ // *********************************
+ // Parameters validation
+ // *********************************
+
+ @Override
+ protected Result verifyParameters(Map<String, Object> parameters) {
+
+ ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.PARAMETERS).error(ResultErrorHelper.requiresOption("accessKey", parameters))
+ .error(ResultErrorHelper.requiresOption("secretKey", parameters)).error(ResultErrorHelper.requiresOption("region", parameters))
+ .error(ResultErrorHelper.requiresOption("streamName", parameters));
+
+ // Validate using the catalog
+
+ super.verifyParametersAgainstCatalog(builder, parameters);
+
+ return builder.build();
+ }
+
+ // *********************************
+ // Connectivity validation
+ // *********************************
+
+ @Override
+ protected Result verifyConnectivity(Map<String, Object> parameters) {
+ ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.CONNECTIVITY);
+
+ try {
+ KinesisFirehose2Configuration configuration = setProperties(new KinesisFirehose2Configuration(), parameters);
+ AwsBasicCredentials cred = AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey());
+ FirehoseClientBuilder clientBuilder = FirehoseClient.builder();
+ FirehoseClient client = clientBuilder.credentialsProvider(StaticCredentialsProvider.create(cred)).region(Region.of(configuration.getRegion())).build();
+ client.listDeliveryStreams();
+ } catch (SdkClientException e) {
+ ResultErrorBuilder errorBuilder = ResultErrorBuilder.withCodeAndDescription(VerificationError.StandardCode.AUTHENTICATION, e.getMessage())
+ .detail("aws_kinesis_firehose_exception_message", e.getMessage()).detail(VerificationError.ExceptionAttribute.EXCEPTION_CLASS, e.getClass().getName())
+ .detail(VerificationError.ExceptionAttribute.EXCEPTION_INSTANCE, e);
+
+ builder.error(errorBuilder.build());
+ } catch (Exception e) {
+ builder.error(ResultErrorBuilder.withException(e).build());
+ }
+ return builder.build();
+ }
+}
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
new file mode 100644
index 0000000..82cab1f
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Configuration.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.spi.UriPath;
+
+import software.amazon.awssdk.core.Protocol;
+import software.amazon.awssdk.services.firehose.FirehoseClient;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+
+@UriParams
+public class KinesisFirehose2Configuration implements Cloneable {
+
+ @UriPath(description = "Name of the stream")
+ @Metadata(required = true)
+ private String streamName;
+ @UriParam(label = "security", secret = true, description = "Amazon AWS Access Key")
+ private String accessKey;
+ @UriParam(label = "security", secret = true, description = "Amazon AWS Secret Key")
+ private String secretKey;
+ @UriParam(description = "The region in which Kinesis client needs to work. When using this parameter, the configuration will expect the capitalized name of the region (for example AP_EAST_1)"
+ + "You'll need to use the name Regions.EU_WEST_1.name()")
+ private String region;
+ @UriParam(description = "Amazon Kinesis Firehose client to use for all requests for this endpoint")
+ private FirehoseClient amazonKinesisFirehoseClient;
+ @UriParam(enums = "HTTP,HTTPS", defaultValue = "HTTPS", description = "To define a proxy protocol when instantiating the Kinesis Firehose client")
+ private Protocol proxyProtocol = Protocol.HTTPS;
+ @UriParam(description = "To define a proxy host when instantiating the Kinesis Firehose client")
+ private String proxyHost;
+ @UriParam(description = "To define a proxy port when instantiating the Kinesis Firehose client")
+ private Integer proxyPort;
+
+ public void setAmazonKinesisFirehoseClient(FirehoseClient client) {
+ this.amazonKinesisFirehoseClient = client;
+ }
+
+ public FirehoseClient getAmazonKinesisFirehoseClient() {
+ return amazonKinesisFirehoseClient;
+ }
+
+ public void setStreamName(String streamName) {
+ this.streamName = streamName;
+ }
+
+ public String getStreamName() {
+ return streamName;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getRegion() {
+ return region;
+ }
+
+ public void setRegion(String region) {
+ this.region = region;
+ }
+
+ public Protocol getProxyProtocol() {
+ return proxyProtocol;
+ }
+
+ public void setProxyProtocol(Protocol proxyProtocol) {
+ this.proxyProtocol = proxyProtocol;
+ }
+
+ public String getProxyHost() {
+ return proxyHost;
+ }
+
+ public void setProxyHost(String proxyHost) {
+ this.proxyHost = proxyHost;
+ }
+
+ public Integer getProxyPort() {
+ return proxyPort;
+ }
+
+ public void setProxyPort(Integer proxyPort) {
+ this.proxyPort = proxyPort;
+ }
+
+ // *************************************************
+ //
+ // *************************************************
+
+ public KinesisFirehose2Configuration copy() {
+ try {
+ return (KinesisFirehose2Configuration)super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(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
new file mode 100644
index 0000000..72ed001
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Constants.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+public interface KinesisFirehose2Constants {
+
+ String RECORD_ID = "CamelAwsKinesisFirehoseRecordId";
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
new file mode 100644
index 0000000..8a85366
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Endpoint.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import java.net.URI;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.util.ObjectHelper;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.http.apache.ProxyConfiguration;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.firehose.FirehoseClient;
+import software.amazon.awssdk.services.firehose.FirehoseClientBuilder;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.KinesisClientBuilder;
+
+/**
+ * The aws-kinesis-firehose component is used for producing Amazon's Kinesis Firehose streams.
+ */
+@UriEndpoint(firstVersion = "3.2.0", scheme = "aws2-kinesis-firehose", title = "AWS 2 Kinesis Firehose", syntax = "aws2-kinesis-firehose:streamName",
+ producerOnly = true, label = "cloud,messaging")
+public class KinesisFirehose2Endpoint extends DefaultEndpoint {
+
+ @UriParam
+ private KinesisFirehose2Configuration configuration;
+
+ private FirehoseClient kinesisFirehoseClient;
+
+ public KinesisFirehose2Endpoint(String uri, KinesisFirehose2Configuration configuration, KinesisFirehose2Component component) {
+ super(uri, component);
+ this.configuration = configuration;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new KinesisFirehose2Producer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ throw new UnsupportedOperationException("You cannot consume messages from this endpoint");
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ kinesisFirehoseClient = configuration.getAmazonKinesisFirehoseClient() != null ? configuration.getAmazonKinesisFirehoseClient()
+ : createKinesisFirehoseClient();
+
+ }
+
+ @Override
+ public void doStop() throws Exception {
+ if (ObjectHelper.isEmpty(configuration.getAmazonKinesisFirehoseClient())) {
+ if (kinesisFirehoseClient != null) {
+ kinesisFirehoseClient.close();
+ }
+ }
+ super.doStop();
+ }
+
+ FirehoseClient createKinesisFirehoseClient() {
+ FirehoseClient client = null;
+ FirehoseClientBuilder clientBuilder = FirehoseClient.builder();
+ ProxyConfiguration.Builder proxyConfig = null;
+ ApacheHttpClient.Builder httpClientBuilder = null;
+ boolean isClientConfigFound = false;
+ if (ObjectHelper.isNotEmpty(configuration.getProxyHost()) && ObjectHelper.isNotEmpty(configuration.getProxyPort())) {
+ proxyConfig = ProxyConfiguration.builder();
+ URI proxyEndpoint = URI.create(configuration.getProxyProtocol() + configuration.getProxyHost() + configuration.getProxyPort());
+ proxyConfig.endpoint(proxyEndpoint);
+ httpClientBuilder = ApacheHttpClient.builder().proxyConfiguration(proxyConfig.build());
+ isClientConfigFound = true;
+ }
+ if (configuration.getAccessKey() != null && configuration.getSecretKey() != null) {
+ AwsBasicCredentials cred = AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey());
+ if (isClientConfigFound) {
+ clientBuilder = clientBuilder.httpClientBuilder(httpClientBuilder).credentialsProvider(StaticCredentialsProvider.create(cred));
+ } else {
+ clientBuilder = clientBuilder.credentialsProvider(StaticCredentialsProvider.create(cred));
+ }
+ } else {
+ if (!isClientConfigFound) {
+ clientBuilder = clientBuilder.httpClientBuilder(httpClientBuilder);
+ }
+ }
+ if (ObjectHelper.isNotEmpty(configuration.getRegion())) {
+ clientBuilder = clientBuilder.region(Region.of(configuration.getRegion()));
+ }
+ client = clientBuilder.build();
+ return client;
+ }
+
+ public FirehoseClient getClient() {
+ return kinesisFirehoseClient;
+ }
+
+ public KinesisFirehose2Configuration getConfiguration() {
+ return configuration;
+ }
+}
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
new file mode 100644
index 0000000..7f26ec8
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/firehose/KinesisFirehose2Producer.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import java.nio.ByteBuffer;
+
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.support.DefaultProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.firehose.model.PutRecordRequest;
+import software.amazon.awssdk.services.firehose.model.PutRecordResponse;
+import software.amazon.awssdk.services.firehose.model.Record;
+
+public class KinesisFirehose2Producer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisFirehose2Producer.class);
+
+ public KinesisFirehose2Producer(KinesisFirehose2Endpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public KinesisFirehose2Endpoint getEndpoint() {
+ return (KinesisFirehose2Endpoint) super.getEndpoint();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ PutRecordRequest request = createRequest(exchange);
+ LOG.trace("Sending request [{}] from exchange [{}]...", request, exchange);
+ PutRecordResponse putRecordResult = getEndpoint().getClient().putRecord(request);
+ LOG.trace("Received result [{}]", putRecordResult);
+ Message message = getMessageForResponse(exchange);
+ message.setHeader(KinesisFirehose2Constants.RECORD_ID, putRecordResult.recordId());
+ }
+
+ private PutRecordRequest createRequest(Exchange exchange) {
+ ByteBuffer body = exchange.getIn().getBody(ByteBuffer.class);
+ Record.Builder record = Record.builder();
+ record.data(SdkBytes.fromByteBuffer(body));
+
+ PutRecordRequest.Builder putRecordRequest = PutRecordRequest.builder();
+ putRecordRequest.deliveryStreamName(getEndpoint().getConfiguration().getStreamName());
+ putRecordRequest.record(record.build());
+ return putRecordRequest.build();
+ }
+
+ public static Message getMessageForResponse(final Exchange exchange) {
+ return exchange.getMessage();
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Component.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Component.java
new file mode 100644
index 0000000..36dee4f
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Component.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.DefaultComponent;
+
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+
+@Component("aws2-kinesis")
+public class Kinesis2Component extends DefaultComponent {
+
+ @Metadata
+ private String accessKey;
+ @Metadata
+ private String secretKey;
+ @Metadata
+ private String region;
+ @Metadata(label = "advanced")
+ private Kinesis2Configuration configuration;
+
+ public Kinesis2Component() {
+ this(null);
+ }
+
+ public Kinesis2Component(CamelContext context) {
+ super(context);
+
+ registerExtension(new Kinesis2ComponentVerifierExtension());
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ Kinesis2Configuration configuration = this.configuration != null ? this.configuration.copy() : new Kinesis2Configuration();
+ configuration.setStreamName(remaining);
+ Kinesis2Endpoint endpoint = new Kinesis2Endpoint(uri, configuration, this);
+ endpoint.getConfiguration().setAccessKey(accessKey);
+ endpoint.getConfiguration().setSecretKey(secretKey);
+ endpoint.getConfiguration().setRegion(region);
+ setProperties(endpoint, parameters);
+ checkAndSetRegistryClient(configuration);
+ if (configuration.getAmazonKinesisClient() == null && (configuration.getAccessKey() == null || configuration.getSecretKey() == null)) {
+ throw new IllegalArgumentException("amazonKinesisClient or accessKey and secretKey must be specified");
+ }
+ return endpoint;
+ }
+
+ public Kinesis2Configuration getConfiguration() {
+ return configuration;
+ }
+
+ /**
+ * The AWS S3 default configuration
+ */
+ public void setConfiguration(Kinesis2Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ /**
+ * Amazon AWS Access Key
+ */
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ /**
+ * Amazon AWS Secret Key
+ */
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getRegion() {
+ return region;
+ }
+
+ /**
+ * Amazon AWS Region
+ */
+ public void setRegion(String region) {
+ this.region = region;
+ }
+
+ private void checkAndSetRegistryClient(Kinesis2Configuration configuration) {
+ Set<KinesisClient> clients = getCamelContext().getRegistry().findByType(KinesisClient.class);
+ if (clients.size() == 1) {
+ configuration.setAmazonKinesisClient(clients.stream().findFirst().get());
+ }
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentVerifierExtension.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentVerifierExtension.java
new file mode 100644
index 0000000..1554969
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentVerifierExtension.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.util.Map;
+
+import org.apache.camel.component.extension.verifier.DefaultComponentVerifierExtension;
+import org.apache.camel.component.extension.verifier.ResultBuilder;
+import org.apache.camel.component.extension.verifier.ResultErrorBuilder;
+import org.apache.camel.component.extension.verifier.ResultErrorHelper;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.KinesisClientBuilder;
+
+public class Kinesis2ComponentVerifierExtension extends DefaultComponentVerifierExtension {
+
+ public Kinesis2ComponentVerifierExtension() {
+ this("aws2-kinesis");
+ }
+
+ public Kinesis2ComponentVerifierExtension(String scheme) {
+ super(scheme);
+ }
+
+ // *********************************
+ // Parameters validation
+ // *********************************
+
+ @Override
+ protected Result verifyParameters(Map<String, Object> parameters) {
+
+ ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.PARAMETERS).error(ResultErrorHelper.requiresOption("accessKey", parameters))
+ .error(ResultErrorHelper.requiresOption("secretKey", parameters)).error(ResultErrorHelper.requiresOption("region", parameters))
+ .error(ResultErrorHelper.requiresOption("streamName", parameters));
+
+ // Validate using the catalog
+
+ super.verifyParametersAgainstCatalog(builder, parameters);
+
+ return builder.build();
+ }
+
+ // *********************************
+ // Connectivity validation
+ // *********************************
+
+ @Override
+ protected Result verifyConnectivity(Map<String, Object> parameters) {
+ ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.CONNECTIVITY);
+
+ try {
+ Kinesis2Configuration configuration = setProperties(new Kinesis2Configuration(), parameters);
+ AwsBasicCredentials cred = AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey());
+ KinesisClientBuilder clientBuilder = KinesisClient.builder();
+ KinesisClient client = clientBuilder.credentialsProvider(StaticCredentialsProvider.create(cred)).region(Region.of(configuration.getRegion())).build();
+ client.listStreams();
+ } catch (SdkClientException e) {
+ ResultErrorBuilder errorBuilder = ResultErrorBuilder.withCodeAndDescription(VerificationError.StandardCode.AUTHENTICATION, e.getMessage())
+ .detail("aws_kinesis_exception_message", e.getMessage()).detail(VerificationError.ExceptionAttribute.EXCEPTION_CLASS, e.getClass().getName())
+ .detail(VerificationError.ExceptionAttribute.EXCEPTION_INSTANCE, e);
+
+ builder.error(errorBuilder.build());
+ } catch (Exception e) {
+ builder.error(ResultErrorBuilder.withException(e).build());
+ }
+ return builder.build();
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java
new file mode 100644
index 0000000..575f057
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.spi.UriPath;
+
+import software.amazon.awssdk.core.Protocol;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+
+@UriParams
+public class Kinesis2Configuration implements Cloneable {
+
+ @UriPath(description = "Name of the stream")
+ @Metadata(required = true)
+ private String streamName;
+ @UriParam(label = "security", secret = true, description = "Amazon AWS Access Key")
+ private String accessKey;
+ @UriParam(label = "security", secret = true, description = "Amazon AWS Secret Key")
+ private String secretKey;
+ @UriParam(description = "The region in which Kinesis client needs to work. When using this parameter, the configuration will expect the capitalized name of the region (for example AP_EAST_1)"
+ + "You'll need to use the name Regions.EU_WEST_1.name()")
+ private String region;
+ @UriParam(description = "Amazon Kinesis client to use for all requests for this endpoint")
+ private KinesisClient amazonKinesisClient;
+ @UriParam(label = "consumer", description = "Maximum number of records that will be fetched in each poll", defaultValue = "1")
+ private int maxResultsPerRequest = 1;
+ @UriParam(label = "consumer", description = "Defines where in the Kinesis stream to start getting records", defaultValue = "TRIM_HORIZON")
+ private ShardIteratorType iteratorType = ShardIteratorType.TRIM_HORIZON;
+ @UriParam(label = "consumer", description = "Defines which shardId in the Kinesis stream to get records from")
+ private String shardId = "";
+ @UriParam(label = "consumer", description = "The sequence number to start polling from. Required if iteratorType is set to AFTER_SEQUENCE_NUMBER or AT_SEQUENCE_NUMBER")
+ private String sequenceNumber = "";
+ @UriParam(label = "consumer", defaultValue = "ignore", description = "Define what will be the behavior in case of shard closed. Possible value are ignore, silent and fail."
+ + " In case of ignore a message will be logged and the consumer will restart from the beginning,"
+ + "in case of silent there will be no logging and the consumer will start from the beginning,"
+ + "in case of fail a ReachedClosedStateException will be raised")
+ private Kinesis2ShardClosedStrategyEnum shardClosed;
+ @UriParam(enums = "HTTP,HTTPS", defaultValue = "HTTPS", description = "To define a proxy protocol when instantiating the Kinesis client")
+ private Protocol proxyProtocol = Protocol.HTTPS;
+ @UriParam(description = "To define a proxy host when instantiating the Kinesis client")
+ private String proxyHost;
+ @UriParam(description = "To define a proxy port when instantiating the Kinesis client")
+ private Integer proxyPort;
+
+ public KinesisClient getAmazonKinesisClient() {
+ return amazonKinesisClient;
+ }
+
+ public void setAmazonKinesisClient(KinesisClient amazonKinesisClient) {
+ this.amazonKinesisClient = amazonKinesisClient;
+ }
+
+ public int getMaxResultsPerRequest() {
+ return maxResultsPerRequest;
+ }
+
+ public void setMaxResultsPerRequest(int maxResultsPerRequest) {
+ this.maxResultsPerRequest = maxResultsPerRequest;
+ }
+
+ public String getStreamName() {
+ return streamName;
+ }
+
+ public void setStreamName(String streamName) {
+ this.streamName = streamName;
+ }
+
+ public ShardIteratorType getIteratorType() {
+ return iteratorType;
+ }
+
+ public void setIteratorType(ShardIteratorType iteratorType) {
+ this.iteratorType = iteratorType;
+ }
+
+ public String getShardId() {
+ return shardId;
+ }
+
+ public void setShardId(String shardId) {
+ this.shardId = shardId;
+ }
+
+ public String getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ public void setSequenceNumber(String sequenceNumber) {
+ this.sequenceNumber = sequenceNumber;
+ }
+
+ public Kinesis2ShardClosedStrategyEnum getShardClosed() {
+ return shardClosed;
+ }
+
+ public void setShardClosed(Kinesis2ShardClosedStrategyEnum shardClosed) {
+ this.shardClosed = shardClosed;
+ }
+
+ public String getAccessKey() {
+ return accessKey;
+ }
+
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ public String getSecretKey() {
+ return secretKey;
+ }
+
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
+ public String getRegion() {
+ return region;
+ }
+
+ public void setRegion(String region) {
+ this.region = region;
+ }
+
+ public Protocol getProxyProtocol() {
+ return proxyProtocol;
+ }
+
+ public void setProxyProtocol(Protocol proxyProtocol) {
+ this.proxyProtocol = proxyProtocol;
+ }
+
+ public String getProxyHost() {
+ return proxyHost;
+ }
+
+ public void setProxyHost(String proxyHost) {
+ this.proxyHost = proxyHost;
+ }
+
+ public Integer getProxyPort() {
+ return proxyPort;
+ }
+
+ public void setProxyPort(Integer proxyPort) {
+ this.proxyPort = proxyPort;
+ }
+
+ // *************************************************
+ //
+ // *************************************************
+
+ public Kinesis2Configuration copy() {
+ try {
+ return (Kinesis2Configuration)super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeCamelException(e);
+ }
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Constants.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Constants.java
new file mode 100644
index 0000000..b09aac9
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Constants.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+public interface Kinesis2Constants {
+
+ String SEQUENCE_NUMBER = "CamelAwsKinesisSequenceNumber";
+ String APPROX_ARRIVAL_TIME = "CamelAwsKinesisApproximateArrivalTimestamp";
+ String PARTITION_KEY = "CamelAwsKinesisPartitionKey";
+
+ /**
+ * in a Kinesis Record object, the shard ID is used on writes to indicate where the data was stored
+ */
+ String SHARD_ID = "CamelAwsKinesisShardId";
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
new file mode 100644
index 0000000..e080083
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.util.ArrayDeque;
+import java.util.List;
+import java.util.Queue;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.support.ScheduledBatchPollingConsumer;
+import org.apache.camel.util.CastUtils;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
+import software.amazon.awssdk.services.kinesis.model.Record;
+import software.amazon.awssdk.services.kinesis.model.Shard;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+
+public class Kinesis2Consumer extends ScheduledBatchPollingConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(Kinesis2Consumer.class);
+
+ private String currentShardIterator;
+ private boolean isShardClosed;
+
+ public Kinesis2Consumer(Kinesis2Endpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ }
+
+ @Override
+ protected int poll() throws Exception {
+ GetRecordsRequest req = GetRecordsRequest.builder().shardIterator(getShardItertor()).limit(getEndpoint().getConfiguration().getMaxResultsPerRequest()).build();
+ GetRecordsResponse result = getClient().getRecords(req);
+
+ Queue<Exchange> exchanges = createExchanges(result.records());
+ int processedExchangeCount = processBatch(CastUtils.cast(exchanges));
+
+ // May cache the last successful sequence number, and pass it to the
+ // getRecords request. That way, on the next poll, we start from where
+ // we left off, however, I don't know what happens to subsequent
+ // exchanges when an earlier echangee fails.
+
+ currentShardIterator = result.nextShardIterator();
+ if (isShardClosed) {
+ switch (getEndpoint().getConfiguration().getShardClosed()) {
+ case ignore:
+ LOG.warn("The shard {} is in closed state", currentShardIterator);
+ break;
+ case silent:
+ break;
+ case fail:
+ LOG.info("Shard Iterator reaches CLOSE status:{} {}", getEndpoint().getConfiguration().getStreamName(), getEndpoint().getConfiguration().getShardId());
+ throw new ReachedClosedStatusException(getEndpoint().getConfiguration().getStreamName(), getEndpoint().getConfiguration().getShardId());
+ default:
+ throw new IllegalArgumentException("Unsupported shard closed strategy");
+ }
+ }
+
+ return processedExchangeCount;
+ }
+
+ @Override
+ public int processBatch(Queue<Object> exchanges) throws Exception {
+ int processedExchanges = 0;
+ while (!exchanges.isEmpty()) {
+ final Exchange exchange = ObjectHelper.cast(Exchange.class, exchanges.poll());
+
+ LOG.trace("Processing exchange [{}] started.", exchange);
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ LOG.trace("Processing exchange [{}] done.", exchange);
+ }
+ });
+ processedExchanges++;
+ }
+ return processedExchanges;
+ }
+
+ private KinesisClient getClient() {
+ return getEndpoint().getClient();
+ }
+
+ @Override
+ public Kinesis2Endpoint getEndpoint() {
+ return (Kinesis2Endpoint)super.getEndpoint();
+ }
+
+ private String getShardItertor() {
+ // either return a cached one or get a new one via a GetShardIterator
+ // request.
+ if (currentShardIterator == null) {
+ String shardId;
+
+ // If ShardId supplied use it, else choose first one
+ if (!getEndpoint().getConfiguration().getShardId().isEmpty()) {
+ shardId = getEndpoint().getConfiguration().getShardId();
+ DescribeStreamRequest req1 = DescribeStreamRequest.builder().streamName(getEndpoint().getConfiguration().getStreamName()).build();
+ DescribeStreamResponse res1 = getClient().describeStream(req1);
+ for (Shard shard : res1.streamDescription().shards()) {
+ if (shard.shardId().equalsIgnoreCase(getEndpoint().getConfiguration().getShardId())) {
+ isShardClosed = shard.sequenceNumberRange().endingSequenceNumber() != null;
+ }
+ }
+
+ } else {
+ DescribeStreamRequest req1 = DescribeStreamRequest.builder().streamName(getEndpoint().getConfiguration().getStreamName()).build();
+ DescribeStreamResponse res1 = getClient().describeStream(req1);
+ shardId = res1.streamDescription().shards().get(0).shardId();
+ isShardClosed = res1.streamDescription().shards().get(0).sequenceNumberRange().endingSequenceNumber() != null;
+ }
+ LOG.debug("ShardId is: {}", shardId);
+
+ GetShardIteratorRequest.Builder req = GetShardIteratorRequest.builder().streamName(getEndpoint().getConfiguration().getStreamName()).shardId(shardId)
+ .shardIteratorType(getEndpoint().getConfiguration().getIteratorType());
+
+ if (hasSequenceNumber()) {
+ req.startingSequenceNumber(getEndpoint().getConfiguration().getSequenceNumber());
+ }
+
+ GetShardIteratorResponse result = getClient().getShardIterator(req.build());
+ currentShardIterator = result.shardIterator();
+ }
+ LOG.debug("Shard Iterator is: {}", currentShardIterator);
+ return currentShardIterator;
+ }
+
+ private Queue<Exchange> createExchanges(List<Record> records) {
+ Queue<Exchange> exchanges = new ArrayDeque<>();
+ for (Record record : records) {
+ exchanges.add(getEndpoint().createExchange(record));
+ }
+ return exchanges;
+ }
+
+ private boolean hasSequenceNumber() {
+ return !getEndpoint().getConfiguration().getSequenceNumber().isEmpty()
+ && (getEndpoint().getConfiguration().getIteratorType().equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER)
+ || getEndpoint().getConfiguration().getIteratorType().equals(ShardIteratorType.AT_SEQUENCE_NUMBER));
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
new file mode 100644
index 0000000..187efbd
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Endpoint.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.http.apache.ProxyConfiguration;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.KinesisClientBuilder;
+import software.amazon.awssdk.services.kinesis.model.Record;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+
+import java.net.URI;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.support.ScheduledPollEndpoint;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * The aws-kinesis component is for consuming and producing records from Amazon
+ * Kinesis Streams.
+ */
+@UriEndpoint(firstVersion = "3.2.0", scheme = "aws2-kinesis", title = "AWS 2 Kinesis", syntax = "aws2-kinesis:streamName", label = "cloud,messaging")
+public class Kinesis2Endpoint extends ScheduledPollEndpoint {
+
+ @UriParam
+ private Kinesis2Configuration configuration;
+
+ private KinesisClient kinesisClient;
+
+ public Kinesis2Endpoint(String uri, Kinesis2Configuration configuration, Kinesis2Component component) {
+ super(uri, component);
+ this.configuration = configuration;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ kinesisClient = configuration.getAmazonKinesisClient() != null ? configuration.getAmazonKinesisClient()
+ : createKinesisClient();
+
+
+ if ((configuration.getIteratorType().equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER) || configuration.getIteratorType().equals(ShardIteratorType.AT_SEQUENCE_NUMBER))
+ && configuration.getSequenceNumber().isEmpty()) {
+ throw new IllegalArgumentException("Sequence Number must be specified with iterator Types AFTER_SEQUENCE_NUMBER or AT_SEQUENCE_NUMBER");
+ }
+ }
+
+ @Override
+ public void doStop() throws Exception {
+ if (ObjectHelper.isEmpty(configuration.getAmazonKinesisClient())) {
+ if (kinesisClient != null) {
+ kinesisClient.close();
+ }
+ }
+ super.doStop();
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new Kinesis2Producer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ final Kinesis2Consumer consumer = new Kinesis2Consumer(this, processor);
+ consumer.setSchedulerProperties(getSchedulerProperties());
+ configureConsumer(consumer);
+ return consumer;
+ }
+
+ public Exchange createExchange(Record record) {
+ Exchange exchange = super.createExchange();
+ exchange.getIn().setBody(record);
+ exchange.getIn().setHeader(Kinesis2Constants.APPROX_ARRIVAL_TIME, record.approximateArrivalTimestamp());
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, record.partitionKey());
+ exchange.getIn().setHeader(Kinesis2Constants.SEQUENCE_NUMBER, record.sequenceNumber());
+ return exchange;
+ }
+
+ public KinesisClient getClient() {
+ return kinesisClient;
+ }
+
+ public Kinesis2Configuration getConfiguration() {
+ return configuration;
+ }
+
+ KinesisClient createKinesisClient() {
+ KinesisClient client = null;
+ KinesisClientBuilder clientBuilder = KinesisClient.builder();
+ ProxyConfiguration.Builder proxyConfig = null;
+ ApacheHttpClient.Builder httpClientBuilder = null;
+ boolean isClientConfigFound = false;
+ if (ObjectHelper.isNotEmpty(configuration.getProxyHost()) && ObjectHelper.isNotEmpty(configuration.getProxyPort())) {
+ proxyConfig = ProxyConfiguration.builder();
+ URI proxyEndpoint = URI.create(configuration.getProxyProtocol() + configuration.getProxyHost() + configuration.getProxyPort());
+ proxyConfig.endpoint(proxyEndpoint);
+ httpClientBuilder = ApacheHttpClient.builder().proxyConfiguration(proxyConfig.build());
+ isClientConfigFound = true;
+ }
+ if (configuration.getAccessKey() != null && configuration.getSecretKey() != null) {
+ AwsBasicCredentials cred = AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey());
+ if (isClientConfigFound) {
+ clientBuilder = clientBuilder.httpClientBuilder(httpClientBuilder).credentialsProvider(StaticCredentialsProvider.create(cred));
+ } else {
+ clientBuilder = clientBuilder.credentialsProvider(StaticCredentialsProvider.create(cred));
+ }
+ } else {
+ if (!isClientConfigFound) {
+ clientBuilder = clientBuilder.httpClientBuilder(httpClientBuilder);
+ }
+ }
+ if (ObjectHelper.isNotEmpty(configuration.getRegion())) {
+ clientBuilder = clientBuilder.region(Region.of(configuration.getRegion()));
+ }
+ client = clientBuilder.build();
+ return client;
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
new file mode 100644
index 0000000..56a3d96
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.nio.ByteBuffer;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.support.DefaultProducer;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
+import software.amazon.awssdk.services.kinesis.model.PutRecordResponse;
+
+public class Kinesis2Producer extends DefaultProducer {
+
+ public Kinesis2Producer(Kinesis2Endpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public Kinesis2Endpoint getEndpoint() {
+ return (Kinesis2Endpoint) super.getEndpoint();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ PutRecordRequest request = createRequest(exchange);
+ PutRecordResponse putRecordResult = getEndpoint().getClient().putRecord(request);
+ Message message = getMessageForResponse(exchange);
+ message.setHeader(Kinesis2Constants.SEQUENCE_NUMBER, putRecordResult.sequenceNumber());
+ message.setHeader(Kinesis2Constants.SHARD_ID, putRecordResult.shardId());
+ }
+
+ private PutRecordRequest createRequest(Exchange exchange) {
+ ByteBuffer body = exchange.getIn().getBody(ByteBuffer.class);
+ Object partitionKey = exchange.getIn().getHeader(Kinesis2Constants.PARTITION_KEY);
+ Object sequenceNumber = exchange.getIn().getHeader(Kinesis2Constants.SEQUENCE_NUMBER);
+
+ PutRecordRequest.Builder putRecordRequest = PutRecordRequest.builder();
+ putRecordRequest.data(SdkBytes.fromByteBuffer(body));
+ putRecordRequest.streamName(getEndpoint().getConfiguration().getStreamName());
+ putRecordRequest.partitionKey(partitionKey.toString());
+ if (sequenceNumber != null) {
+ putRecordRequest.sequenceNumberForOrdering(sequenceNumber.toString());
+ }
+ return putRecordRequest.build();
+ }
+
+ public static Message getMessageForResponse(final Exchange exchange) {
+ return exchange.getMessage();
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2ShardClosedStrategyEnum.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2ShardClosedStrategyEnum.java
new file mode 100644
index 0000000..166334e
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2ShardClosedStrategyEnum.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+public enum Kinesis2ShardClosedStrategyEnum {
+
+ ignore,
+ fail,
+ silent
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/ReachedClosedStatusException.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/ReachedClosedStatusException.java
new file mode 100644
index 0000000..8747263
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/ReachedClosedStatusException.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+public class ReachedClosedStatusException extends Exception {
+
+ private static final long serialVersionUID = -2701697822726751407L;
+
+ private final String streamName;
+ private final String shardId;
+
+ public ReachedClosedStatusException(String streamName, String shardId) {
+ this.streamName = streamName;
+ this.shardId = shardId;
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/RecordStringConverter.java b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/RecordStringConverter.java
new file mode 100644
index 0000000..9e83008
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/RecordStringConverter.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+
+import org.apache.camel.Converter;
+
+import software.amazon.awssdk.services.kinesis.model.Record;
+
+// Allow to ignore this type converter if the kinesis JARs are not present on the classpath
+@Converter(generateLoader = true, ignoreOnLoadError = true)
+public final class RecordStringConverter {
+
+ private RecordStringConverter() {
+ }
+
+ @Converter
+ public static String toString(Record record) {
+ Charset charset = Charset.forName("UTF-8");
+
+ ByteBuffer buffer = record.data().asByteBuffer();
+ if (buffer.hasArray()) {
+ byte[] bytes = record.data().asByteArray();
+ return new String(bytes, charset);
+ } else {
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ return new String(bytes, charset);
+ }
+ }
+
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseComponentConfigurationTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseComponentConfigurationTest.java
new file mode 100644
index 0000000..7d4b2df
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseComponentConfigurationTest.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import org.apache.camel.component.aws2.firehose.KinesisFirehose2Component;
+import org.apache.camel.component.aws2.firehose.KinesisFirehose2Endpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import software.amazon.awssdk.core.Protocol;
+import software.amazon.awssdk.regions.Region;
+
+public class KinesisFirehoseComponentConfigurationTest extends CamelTestSupport {
+
+ @Test
+ public void createEndpointWithAccessAndSecretKey() throws Exception {
+ KinesisFirehose2Component component = context.getComponent("aws2-kinesis-firehose", KinesisFirehose2Component.class);
+ KinesisFirehose2Endpoint endpoint = (KinesisFirehose2Endpoint)component.createEndpoint("aws2-kinesis-firehose://some_stream_name?accessKey=xxxxx&secretKey=yyyyy");
+
+ assertEquals("some_stream_name", endpoint.getConfiguration().getStreamName());
+ assertEquals("xxxxx", endpoint.getConfiguration().getAccessKey());
+ assertEquals("yyyyy", endpoint.getConfiguration().getSecretKey());
+ }
+
+ @Test
+ public void createEndpointWithComponentElements() throws Exception {
+ KinesisFirehose2Component component = context.getComponent("aws2-kinesis-firehose", KinesisFirehose2Component.class);
+ component.setAccessKey("XXX");
+ component.setSecretKey("YYY");
+ KinesisFirehose2Endpoint endpoint = (KinesisFirehose2Endpoint)component.createEndpoint("aws2-kinesis-firehose://some_stream_name");
+
+ assertEquals("some_stream_name", endpoint.getConfiguration().getStreamName());
+ assertEquals("XXX", endpoint.getConfiguration().getAccessKey());
+ assertEquals("YYY", endpoint.getConfiguration().getSecretKey());
+ }
+
+ @Test
+ public void createEndpointWithComponentAndEndpointElements() throws Exception {
+ KinesisFirehose2Component component = context.getComponent("aws2-kinesis-firehose", KinesisFirehose2Component.class);
+ component.setAccessKey("XXX");
+ component.setSecretKey("YYY");
+ component.setRegion(Region.US_WEST_1.toString());
+ KinesisFirehose2Endpoint endpoint = (KinesisFirehose2Endpoint)component.createEndpoint("aws2-kinesis-firehose://some_stream_name?accessKey=xxxxxx&secretKey=yyyyy®ion=US_EAST_1");
+
+ assertEquals("some_stream_name", endpoint.getConfiguration().getStreamName());
+ assertEquals("xxxxxx", endpoint.getConfiguration().getAccessKey());
+ assertEquals("yyyyy", endpoint.getConfiguration().getSecretKey());
+ assertEquals("US_EAST_1", endpoint.getConfiguration().getRegion());
+ }
+
+ @Test
+ public void createEndpointWithComponentEndpointElementsAndProxy() throws Exception {
+ KinesisFirehose2Component component = context.getComponent("aws2-kinesis-firehose", KinesisFirehose2Component.class);
+ component.setAccessKey("XXX");
+ component.setSecretKey("YYY");
+ component.setRegion(Region.US_WEST_1.toString());
+ KinesisFirehose2Endpoint endpoint = (KinesisFirehose2Endpoint)component.createEndpoint("aws2-kinesis-firehose://label?accessKey=xxxxxx&secretKey=yyyyy®ion=US_EAST_1&proxyHost=localhost&proxyPort=9000&proxyProtocol=HTTP");
+
+ assertEquals("xxxxxx", endpoint.getConfiguration().getAccessKey());
+ assertEquals("yyyyy", endpoint.getConfiguration().getSecretKey());
+ assertEquals("US_EAST_1", endpoint.getConfiguration().getRegion());
+ assertEquals(Protocol.HTTP, endpoint.getConfiguration().getProxyProtocol());
+ assertEquals("localhost", endpoint.getConfiguration().getProxyHost());
+ assertEquals(Integer.valueOf(9000), endpoint.getConfiguration().getProxyPort());
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseComponentVerifierExtensionTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseComponentVerifierExtensionTest.java
new file mode 100644
index 0000000..5b216c2
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseComponentVerifierExtensionTest.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.Component;
+import org.apache.camel.component.extension.ComponentVerifierExtension;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class KinesisFirehoseComponentVerifierExtensionTest extends CamelTestSupport {
+
+ // *************************************************
+ // Tests (parameters)
+ // *************************************************
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ public void testParameters() throws Exception {
+ Component component = context().getComponent("aws2-kinesis-firehose");
+
+ ComponentVerifierExtension verifier = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new);
+
+ Map<String, Object> parameters = new HashMap<>();
+ parameters.put("secretKey", "l");
+ parameters.put("accessKey", "k");
+ parameters.put("region", "l");
+ parameters.put("streamName", "test");
+
+ ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.PARAMETERS, parameters);
+
+ Assert.assertEquals(ComponentVerifierExtension.Result.Status.OK, result.getStatus());
+ }
+
+ @Test
+ public void testConnectivity() throws Exception {
+ Component component = context().getComponent("aws2-kinesis-firehose");
+ ComponentVerifierExtension verifier = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new);
+
+ Map<String, Object> parameters = new HashMap<>();
+ parameters.put("secretKey", "l");
+ parameters.put("accessKey", "k");
+ parameters.put("region", "US_EAST_1");
+ parameters.put("streamName", "test");
+
+ ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.CONNECTIVITY, parameters);
+
+ Assert.assertEquals(ComponentVerifierExtension.Result.Status.ERROR, result.getStatus());
+ }
+
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseEndpointTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseEndpointTest.java
new file mode 100644
index 0000000..c3afb12
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/KinesisFirehoseEndpointTest.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.aws2.firehose.KinesisFirehose2Endpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.SimpleRegistry;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.firehose.FirehoseClient;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisFirehoseEndpointTest {
+
+ @Mock
+ private FirehoseClient amazonKinesisFirehoseClient;
+
+ private CamelContext camelContext;
+
+ @Before
+ public void setup() throws Exception {
+ SimpleRegistry registry = new SimpleRegistry();
+ registry.bind("firehoseClient", amazonKinesisFirehoseClient);
+ camelContext = new DefaultCamelContext(registry);
+ }
+
+ @Test
+ public void allEndpointParams() throws Exception {
+ KinesisFirehose2Endpoint endpoint = (KinesisFirehose2Endpoint) camelContext.getEndpoint("aws2-kinesis-firehose://some_stream_name"
+ + "?amazonKinesisFirehoseClient=#firehoseClient"
+ );
+ endpoint.start();
+
+ assertThat(endpoint.getClient(), is(amazonKinesisFirehoseClient));
+ assertThat(endpoint.getConfiguration().getStreamName(), is("some_stream_name"));
+ }
+
+ @Test
+ public void allClientCreationParams() throws Exception {
+ KinesisFirehose2Endpoint endpoint = (KinesisFirehose2Endpoint) camelContext.getEndpoint("aws2-kinesis-firehose://some_stream_name"
+ + "?accessKey=xxx&secretKey=yyy®ion=us-east-1"
+ );
+
+ assertThat(endpoint.getConfiguration().getRegion(), is(Region.US_EAST_1.id()));
+ assertThat(endpoint.getConfiguration().getAccessKey(), is("xxx"));
+ assertThat(endpoint.getConfiguration().getSecretKey(), is("yyy"));
+ assertThat(endpoint.getConfiguration().getStreamName(), is("some_stream_name"));
+ }
+}
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
new file mode 100644
index 0000000..1240902
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/firehose/integration/KinesisFirehoseComponentIntegrationTest.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.firehose.integration;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.aws2.firehose.KinesisFirehose2Constants;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import software.amazon.awssdk.services.firehose.FirehoseClient;
+
+@Ignore("Must be manually tested.")
+public class KinesisFirehoseComponentIntegrationTest extends CamelTestSupport {
+
+ @BindToRegistry("FirehoseClient")
+ FirehoseClient client = FirehoseClient.builder().build();
+
+ @Test
+ public void testFirehoseRouting() throws Exception {
+ Exchange exchange = template.send("direct:start", ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setBody("my message text");
+ }
+ });
+ assertNotNull(exchange.getIn().getHeader(KinesisFirehose2Constants.RECORD_ID));
+ }
+
+ @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");
+ }
+ };
+ }
+}
+
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisComponentConfigurationTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisComponentConfigurationTest.java
new file mode 100644
index 0000000..25dc454
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisComponentConfigurationTest.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.component.aws2.kinesis.Kinesis2Component;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Endpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+import software.amazon.awssdk.core.Protocol;
+import software.amazon.awssdk.regions.Region;
+
+public class KinesisComponentConfigurationTest extends CamelTestSupport {
+
+ @Test
+ public void createEndpointWithAccessAndSecretKey() throws Exception {
+ Kinesis2Component component = context.getComponent("aws2-kinesis", Kinesis2Component.class);
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint)component.createEndpoint("aws2-kinesis://some_stream_name?accessKey=xxxxx&secretKey=yyyyy");
+
+ assertEquals("some_stream_name", endpoint.getConfiguration().getStreamName());
+ assertEquals("xxxxx", endpoint.getConfiguration().getAccessKey());
+ assertEquals("yyyyy", endpoint.getConfiguration().getSecretKey());
+ }
+
+ @Test
+ public void createEndpointWithComponentElements() throws Exception {
+ Kinesis2Component component = context.getComponent("aws2-kinesis", Kinesis2Component.class);
+ component.setAccessKey("XXX");
+ component.setSecretKey("YYY");
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint)component.createEndpoint("aws2-kinesis://some_stream_name");
+
+ assertEquals("some_stream_name", endpoint.getConfiguration().getStreamName());
+ assertEquals("XXX", endpoint.getConfiguration().getAccessKey());
+ assertEquals("YYY", endpoint.getConfiguration().getSecretKey());
+ }
+
+ @Test
+ public void createEndpointWithComponentAndEndpointElements() throws Exception {
+ Kinesis2Component component = context.getComponent("aws2-kinesis", Kinesis2Component.class);
+ component.setAccessKey("XXX");
+ component.setSecretKey("YYY");
+ component.setRegion(Region.US_WEST_1.toString());
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint)component.createEndpoint("aws2-kinesis://some_stream_name?accessKey=xxxxxx&secretKey=yyyyy®ion=US_EAST_1");
+
+ assertEquals("some_stream_name", endpoint.getConfiguration().getStreamName());
+ assertEquals("xxxxxx", endpoint.getConfiguration().getAccessKey());
+ assertEquals("yyyyy", endpoint.getConfiguration().getSecretKey());
+ assertEquals("US_EAST_1", endpoint.getConfiguration().getRegion());
+ }
+
+ @Test
+ public void createEndpointWithComponentEndpointElementsAndProxy() throws Exception {
+ Kinesis2Component component = context.getComponent("aws2-kinesis", Kinesis2Component.class);
+ component.setAccessKey("XXX");
+ component.setSecretKey("YYY");
+ component.setRegion(Region.US_WEST_1.toString());
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint)component.createEndpoint("aws2-kinesis://label?accessKey=xxxxxx&secretKey=yyyyy®ion=US_EAST_1&proxyHost=localhost&proxyPort=9000&proxyProtocol=HTTP");
+
+ assertEquals("xxxxxx", endpoint.getConfiguration().getAccessKey());
+ assertEquals("yyyyy", endpoint.getConfiguration().getSecretKey());
+ assertEquals("US_EAST_1", endpoint.getConfiguration().getRegion());
+ assertEquals(Protocol.HTTP, endpoint.getConfiguration().getProxyProtocol());
+ assertEquals("localhost", endpoint.getConfiguration().getProxyHost());
+ assertEquals(Integer.valueOf(9000), endpoint.getConfiguration().getProxyPort());
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisComponentVerifierExtensionTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisComponentVerifierExtensionTest.java
new file mode 100644
index 0000000..c109cdf
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisComponentVerifierExtensionTest.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.Component;
+import org.apache.camel.component.extension.ComponentVerifierExtension;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class KinesisComponentVerifierExtensionTest extends CamelTestSupport {
+
+ // *************************************************
+ // Tests (parameters)
+ // *************************************************
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ public void testParameters() throws Exception {
+ Component component = context().getComponent("aws2-kinesis");
+
+ ComponentVerifierExtension verifier = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new);
+
+ Map<String, Object> parameters = new HashMap<>();
+ parameters.put("secretKey", "l");
+ parameters.put("accessKey", "k");
+ parameters.put("region", "l");
+ parameters.put("streamName", "test");
+
+ ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.PARAMETERS, parameters);
+
+ Assert.assertEquals(ComponentVerifierExtension.Result.Status.OK, result.getStatus());
+ }
+
+ @Test
+ public void testConnectivity() throws Exception {
+ Component component = context().getComponent("aws2-kinesis");
+ ComponentVerifierExtension verifier = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new);
+
+ Map<String, Object> parameters = new HashMap<>();
+ parameters.put("secretKey", "l");
+ parameters.put("accessKey", "k");
+ parameters.put("region", "US_EAST_1");
+ parameters.put("streamName", "test");
+
+ ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.CONNECTIVITY, parameters);
+
+ Assert.assertEquals(ComponentVerifierExtension.Result.Status.ERROR, result.getStatus());
+ }
+
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithFailTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithFailTest.java
new file mode 100644
index 0000000..38e6450
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithFailTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.util.ArrayList;
+
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Component;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Configuration;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Consumer;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Endpoint;
+import org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum;
+import org.apache.camel.component.aws2.kinesis.ReachedClosedStatusException;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
+import software.amazon.awssdk.services.kinesis.model.SequenceNumberRange;
+import software.amazon.awssdk.services.kinesis.model.Shard;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+import software.amazon.awssdk.services.kinesis.model.StreamDescription;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisConsumerClosedShardWithFailTest {
+
+ @Mock
+ private KinesisClient kinesisClient;
+ @Mock
+ private AsyncProcessor processor;
+
+ private final CamelContext context = new DefaultCamelContext();
+ private final Kinesis2Component component = new Kinesis2Component(context);
+
+ private Kinesis2Consumer undertest;
+
+ @Before
+ public void setup() throws Exception {
+ Kinesis2Configuration configuration = new Kinesis2Configuration();
+ configuration.setAmazonKinesisClient(kinesisClient);
+ configuration.setIteratorType(ShardIteratorType.LATEST);
+ configuration.setShardClosed(Kinesis2ShardClosedStrategyEnum.fail);
+ configuration.setStreamName("streamName");
+ Kinesis2Endpoint endpoint = new Kinesis2Endpoint(null, configuration, component);
+ endpoint.start();
+ undertest = new Kinesis2Consumer(endpoint, processor);
+
+
+ SequenceNumberRange range = SequenceNumberRange.builder().endingSequenceNumber("20").build();
+ Shard shard = Shard.builder().shardId("shardId").sequenceNumberRange(range).build();
+ ArrayList<Shard> shardList = new ArrayList<>();
+ shardList.add(shard);
+
+ when(kinesisClient.getRecords(any(GetRecordsRequest.class))).thenReturn(GetRecordsResponse.builder().nextShardIterator("nextShardIterator").build());
+ when(kinesisClient.describeStream(any(DescribeStreamRequest.class)))
+ .thenReturn(DescribeStreamResponse.builder().streamDescription(StreamDescription.builder().shards(shardList).build()).build());
+ when(kinesisClient.getShardIterator(any(GetShardIteratorRequest.class))).thenReturn(GetShardIteratorResponse.builder().shardIterator("shardIterator").build());
+ }
+
+ @Test(expected = ReachedClosedStatusException.class)
+ public void itObtainsAShardIteratorOnFirstPoll() throws Exception {
+ undertest.poll();
+
+ final ArgumentCaptor<DescribeStreamRequest> describeStreamReqCap = ArgumentCaptor.forClass(DescribeStreamRequest.class);
+ final ArgumentCaptor<GetShardIteratorRequest> getShardIteratorReqCap = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
+
+ verify(kinesisClient).describeStream(describeStreamReqCap.capture());
+ assertThat(describeStreamReqCap.getValue().streamName(), is("streamName"));
+
+ verify(kinesisClient).getShardIterator(getShardIteratorReqCap.capture());
+ assertThat(getShardIteratorReqCap.getValue().streamName(), is("streamName"));
+ assertThat(getShardIteratorReqCap.getValue().shardId(), is("shardId"));
+ assertThat(getShardIteratorReqCap.getValue().shardIteratorType(), is("LATEST"));
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
new file mode 100644
index 0000000..bccdd37
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Date;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Component;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Configuration;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Constants;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Consumer;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Endpoint;
+import org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
+import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
+import software.amazon.awssdk.services.kinesis.model.Record;
+import software.amazon.awssdk.services.kinesis.model.SequenceNumberRange;
+import software.amazon.awssdk.services.kinesis.model.Shard;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+import software.amazon.awssdk.services.kinesis.model.StreamDescription;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisConsumerClosedShardWithSilentTest {
+
+ @Mock
+ private KinesisClient kinesisClient;
+ @Mock
+ private AsyncProcessor processor;
+
+ private final CamelContext context = new DefaultCamelContext();
+ private final Kinesis2Component component = new Kinesis2Component(context);
+
+ private Kinesis2Consumer undertest;
+
+ @Before
+ public void setup() throws Exception {
+ Kinesis2Configuration configuration = new Kinesis2Configuration();
+ configuration.setAmazonKinesisClient(kinesisClient);
+ configuration.setIteratorType(ShardIteratorType.LATEST);
+ configuration.setShardClosed(Kinesis2ShardClosedStrategyEnum.silent);
+ configuration.setStreamName("streamName");
+ Kinesis2Endpoint endpoint = new Kinesis2Endpoint(null, configuration, component);
+ endpoint.start();
+ undertest = new Kinesis2Consumer(endpoint, processor);
+
+ SequenceNumberRange range = SequenceNumberRange.builder().endingSequenceNumber("20").build();
+ Shard shard = Shard.builder().shardId("shardId").sequenceNumberRange(range).build();
+ ArrayList<Shard> shardList = new ArrayList<>();
+ shardList.add(shard);
+
+
+ when(kinesisClient.getRecords(any(GetRecordsRequest.class)))
+ .thenReturn(GetRecordsResponse.builder()
+ .nextShardIterator("nextShardIterator").build()
+ );
+ when(kinesisClient.describeStream(any(DescribeStreamRequest.class)))
+ .thenReturn(DescribeStreamResponse.builder()
+ .streamDescription(StreamDescription.builder()
+ .shards(shardList).build()
+ ).build()
+ );
+ when(kinesisClient.getShardIterator(any(GetShardIteratorRequest.class)))
+ .thenReturn(GetShardIteratorResponse.builder()
+ .shardIterator("shardIterator").build()
+ );
+ }
+
+ @Test
+ public void itObtainsAShardIteratorOnFirstPoll() throws Exception {
+ undertest.poll();
+
+ final ArgumentCaptor<DescribeStreamRequest> describeStreamReqCap = ArgumentCaptor.forClass(DescribeStreamRequest.class);
+ final ArgumentCaptor<GetShardIteratorRequest> getShardIteratorReqCap = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
+
+ verify(kinesisClient).describeStream(describeStreamReqCap.capture());
+ assertThat(describeStreamReqCap.getValue().streamName(), is("streamName"));
+
+ verify(kinesisClient).getShardIterator(getShardIteratorReqCap.capture());
+ assertThat(getShardIteratorReqCap.getValue().streamName(), is("streamName"));
+ assertThat(getShardIteratorReqCap.getValue().shardId(), is("shardId"));
+ assertThat(getShardIteratorReqCap.getValue().shardIteratorType(), is(ShardIteratorType.LATEST));
+ }
+
+ @Test
+ public void itDoesNotMakeADescribeStreamRequestIfShardIdIsSet() throws Exception {
+ undertest.getEndpoint().getConfiguration().setShardId("shardIdPassedAsUrlParam");
+
+ undertest.poll();
+
+ final ArgumentCaptor<GetShardIteratorRequest> getShardIteratorReqCap = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
+
+ verify(kinesisClient).getShardIterator(getShardIteratorReqCap.capture());
+ assertThat(getShardIteratorReqCap.getValue().streamName(), is("streamName"));
+ assertThat(getShardIteratorReqCap.getValue().shardId(), is("shardIdPassedAsUrlParam"));
+ assertThat(getShardIteratorReqCap.getValue().shardIteratorType(), is(ShardIteratorType.LATEST));
+ }
+
+ @Test
+ public void itObtainsAShardIteratorOnFirstPollForSequenceNumber() throws Exception {
+ undertest.getEndpoint().getConfiguration().setSequenceNumber("12345");
+ undertest.getEndpoint().getConfiguration().setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
+
+ undertest.poll();
+
+ final ArgumentCaptor<DescribeStreamRequest> describeStreamReqCap = ArgumentCaptor.forClass(DescribeStreamRequest.class);
+ final ArgumentCaptor<GetShardIteratorRequest> getShardIteratorReqCap = ArgumentCaptor.forClass(GetShardIteratorRequest.class);
+
+ verify(kinesisClient).describeStream(describeStreamReqCap.capture());
+ assertThat(describeStreamReqCap.getValue().streamName(), is("streamName"));
+
+ verify(kinesisClient).getShardIterator(getShardIteratorReqCap.capture());
+ assertThat(getShardIteratorReqCap.getValue().streamName(), is("streamName"));
+ assertThat(getShardIteratorReqCap.getValue().shardId(), is("shardId"));
+ assertThat(getShardIteratorReqCap.getValue().shardIteratorType(), is(ShardIteratorType.AFTER_SEQUENCE_NUMBER));
+ assertThat(getShardIteratorReqCap.getValue().startingSequenceNumber(), is("12345"));
+
+ }
+
+ @Test
+ public void itUsesTheShardIteratorOnPolls() throws Exception {
+ undertest.poll();
+
+ final ArgumentCaptor<GetRecordsRequest> getRecordsReqCap = ArgumentCaptor.forClass(GetRecordsRequest.class);
+ verify(kinesisClient).getRecords(getRecordsReqCap.capture());
+
+ assertThat(getRecordsReqCap.getValue().shardIterator(), is("shardIterator"));
+ }
+
+ @Test
+ public void itUsesTheShardIteratorOnSubsiquentPolls() throws Exception {
+ undertest.poll();
+ undertest.poll();
+
+ final ArgumentCaptor<GetRecordsRequest> getRecordsReqCap = ArgumentCaptor.forClass(GetRecordsRequest.class);
+
+ verify(kinesisClient, times(1)).describeStream(any(DescribeStreamRequest.class));
+ verify(kinesisClient, times(1)).getShardIterator(any(GetShardIteratorRequest.class));
+ verify(kinesisClient, times(2)).getRecords(getRecordsReqCap.capture());
+ assertThat(getRecordsReqCap.getAllValues().get(0).shardIterator(), is("shardIterator"));
+ assertThat(getRecordsReqCap.getAllValues().get(1).shardIterator(), is("nextShardIterator"));
+ }
+
+ @Test
+ public void recordsAreSentToTheProcessor() throws Exception {
+ when(kinesisClient.getRecords(any(GetRecordsRequest.class)))
+ .thenReturn(GetRecordsResponse.builder()
+ .nextShardIterator("nextShardIterator")
+ .records(Record.builder().sequenceNumber("1").build(), Record.builder().sequenceNumber("2").build()).build()
+ );
+
+ int messageCount = undertest.poll();
+
+ assertThat(messageCount, is(2));
+ final ArgumentCaptor<Exchange> exchangeCaptor = ArgumentCaptor.forClass(Exchange.class);
+
+ verify(processor, times(2)).process(exchangeCaptor.capture(), any(AsyncCallback.class));
+ assertThat(exchangeCaptor.getAllValues().get(0).getIn().getBody(Record.class).sequenceNumber(), is("1"));
+ assertThat(exchangeCaptor.getAllValues().get(1).getIn().getBody(Record.class).sequenceNumber(), is("2"));
+ }
+
+ @Test
+ public void exchangePropertiesAreSet() throws Exception {
+ String partitionKey = "partitionKey";
+ String sequenceNumber = "1";
+ when(kinesisClient.getRecords(any(GetRecordsRequest.class)))
+ .thenReturn(GetRecordsResponse.builder()
+ .nextShardIterator("nextShardIterator")
+ .records(Record.builder()
+ .sequenceNumber(sequenceNumber)
+ .approximateArrivalTimestamp(Instant.now())
+ .partitionKey(partitionKey).build()
+ ).build()
+ );
+
+ undertest.poll();
+
+ final ArgumentCaptor<Exchange> exchangeCaptor = ArgumentCaptor.forClass(Exchange.class);
+
+ verify(processor).process(exchangeCaptor.capture(), any(AsyncCallback.class));
+ assertThat(exchangeCaptor.getValue().getIn().getHeader(Kinesis2Constants.PARTITION_KEY, String.class), is(partitionKey));
+ assertThat(exchangeCaptor.getValue().getIn().getHeader(Kinesis2Constants.SEQUENCE_NUMBER, String.class), is(sequenceNumber));
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointTest.java
new file mode 100644
index 0000000..413c107
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointTest.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Endpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.SimpleRegistry;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.ShardIteratorType;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KinesisEndpointTest {
+
+ @Mock
+ private KinesisClient amazonKinesisClient;
+
+ private CamelContext camelContext;
+
+ @Before
+ public void setup() throws Exception {
+ SimpleRegistry registry = new SimpleRegistry();
+ registry.bind("kinesisClient", amazonKinesisClient);
+ camelContext = new DefaultCamelContext(registry);
+ }
+
+ @Test
+ public void allTheEndpointParams() throws Exception {
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint) camelContext.getEndpoint("aws2-kinesis://some_stream_name"
+ + "?amazonKinesisClient=#kinesisClient"
+ + "&maxResultsPerRequest=101"
+ + "&iteratorType=latest"
+ + "&shardId=abc"
+ + "&sequenceNumber=123"
+ );
+
+ assertThat(endpoint.getConfiguration().getAmazonKinesisClient(), is(amazonKinesisClient));
+ assertThat(endpoint.getConfiguration().getStreamName(), is("some_stream_name"));
+ assertThat(endpoint.getConfiguration().getIteratorType(), is(ShardIteratorType.LATEST));
+ assertThat(endpoint.getConfiguration().getMaxResultsPerRequest(), is(101));
+ assertThat(endpoint.getConfiguration().getSequenceNumber(), is("123"));
+ assertThat(endpoint.getConfiguration().getShardId(), is("abc"));
+ }
+
+ @Test
+ public void onlyRequiredEndpointParams() throws Exception {
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint) camelContext.getEndpoint("aws2-kinesis://some_stream_name"
+ + "?amazonKinesisClient=#kinesisClient"
+ );
+
+ assertThat(endpoint.getConfiguration().getAmazonKinesisClient(), is(amazonKinesisClient));
+ assertThat(endpoint.getConfiguration().getStreamName(), is("some_stream_name"));
+ assertThat(endpoint.getConfiguration().getIteratorType(), is(ShardIteratorType.TRIM_HORIZON));
+ assertThat(endpoint.getConfiguration().getMaxResultsPerRequest(), is(1));
+ }
+
+ @Test
+ public void afterSequenceNumberRequiresSequenceNumber() throws Exception {
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint) camelContext.getEndpoint("aws2-kinesis://some_stream_name"
+ + "?amazonKinesisClient=#kinesisClient"
+ + "&iteratorType=AFTER_SEQUENCE_NUMBER"
+ + "&shardId=abc"
+ + "&sequenceNumber=123"
+ );
+
+ assertThat(endpoint.getConfiguration().getAmazonKinesisClient(), is(amazonKinesisClient));
+ assertThat(endpoint.getConfiguration().getStreamName(), is("some_stream_name"));
+ assertThat(endpoint.getConfiguration().getIteratorType(), is(ShardIteratorType.AFTER_SEQUENCE_NUMBER));
+ assertThat(endpoint.getConfiguration().getShardId(), is("abc"));
+ assertThat(endpoint.getConfiguration().getSequenceNumber(), is("123"));
+ }
+
+ @Test
+ public void atSequenceNumberRequiresSequenceNumber() throws Exception {
+ Kinesis2Endpoint endpoint = (Kinesis2Endpoint) camelContext.getEndpoint("aws2-kinesis://some_stream_name"
+ + "?amazonKinesisClient=#kinesisClient"
+ + "&iteratorType=AT_SEQUENCE_NUMBER"
+ + "&shardId=abc"
+ + "&sequenceNumber=123"
+ );
+
+ assertThat(endpoint.getConfiguration().getAmazonKinesisClient(), is(amazonKinesisClient));
+ assertThat(endpoint.getConfiguration().getStreamName(), is("some_stream_name"));
+ assertThat(endpoint.getConfiguration().getIteratorType(), is(ShardIteratorType.AT_SEQUENCE_NUMBER));
+ assertThat(endpoint.getConfiguration().getShardId(), is("abc"));
+ assertThat(endpoint.getConfiguration().getSequenceNumber(), is("123"));
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/RecordStringConverterTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/RecordStringConverterTest.java
new file mode 100644
index 0000000..e368f87
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/RecordStringConverterTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+
+import org.apache.camel.component.aws2.kinesis.RecordStringConverter;
+import org.junit.Test;
+
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.kinesis.model.Record;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+public class RecordStringConverterTest {
+
+ @Test
+ public void convertRecordToString() throws Exception {
+ Record record = Record.builder()
+ .sequenceNumber("1")
+ .data(SdkBytes.fromByteBuffer(ByteBuffer.wrap("this is a String".getBytes(Charset.forName("UTF-8"))))).build();
+
+ String result = RecordStringConverter.toString(record);
+ assertThat(result, is("this is a String"));
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisComponentIntegrationTest.java b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisComponentIntegrationTest.java
new file mode 100644
index 0000000..22062c1
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisComponentIntegrationTest.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis.integration;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Processor;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Constants;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.Record;
+
+@Ignore("Must be manually tested.")
+public class KinesisComponentIntegrationTest extends CamelTestSupport {
+
+ @BindToRegistry("amazonKinesisClient")
+ KinesisClient client = KinesisClient.builder().build();
+
+ @EndpointInject("direct:start")
+ private ProducerTemplate template;
+
+ @EndpointInject("mock:result")
+ private MockEndpoint result;
+
+ @Test
+ public void send() throws Exception {
+ result.expectedMessageCount(2);
+
+ template.send("direct:start", ExchangePattern.InOnly, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "partition-1");
+ exchange.getIn().setBody("Kinesis Event 1.");
+ }
+ });
+
+ template.send("direct:start", ExchangePattern.InOut, new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(Kinesis2Constants.PARTITION_KEY, "partition-1");
+ exchange.getIn().setBody("Kinesis Event 2.");
+ }
+ });
+
+ assertMockEndpointsSatisfied();
+
+ assertResultExchange(result.getExchanges().get(0), "Kinesis Event 1.", "partition-1");
+ assertResultExchange(result.getExchanges().get(1), "Kinesis Event 2.", "partition-1");
+ }
+
+ private void assertResultExchange(Exchange resultExchange, String data, String partition) {
+ assertIsInstanceOf(Record.class, resultExchange.getIn().getBody());
+ Record record = resultExchange.getIn().getBody(Record.class);
+ assertEquals(data, new String(record.data().asByteArray()));
+ assertEquals(partition, resultExchange.getIn().getHeader(Kinesis2Constants.PARTITION_KEY));
+ assertNotNull(resultExchange.getIn().getHeader(Kinesis2Constants.APPROX_ARRIVAL_TIME));
+ assertNotNull(resultExchange.getIn().getHeader(Kinesis2Constants.SEQUENCE_NUMBER));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ String kinesisEndpointUri = "aws2-kinesis://kinesis1?amazonKinesisClient=#amazonKinesisClient";
+
+ from("direct:start").to(kinesisEndpointUri);
+ }
+ };
+ }
+}
diff --git a/components/camel-aws2-kinesis/src/test/resources/log4j2.properties b/components/camel-aws2-kinesis/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..a87d93f
--- /dev/null
+++ b/components/camel-aws2-kinesis/src/test/resources/log4j2.properties
@@ -0,0 +1,28 @@
+## ---------------------------------------------------------------------------
+## Licensed to the Apache Software Foundation (ASF) under one or more
+## contributor license agreements. See the NOTICE file distributed with
+## this work for additional information regarding copyright ownership.
+## The ASF licenses this file to You under the Apache License, Version 2.0
+## (the "License"); you may not use this file except in compliance with
+## the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+## ---------------------------------------------------------------------------
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-aws-kinesis-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.out.type = Console
+appender.out.name = out
+appender.out.layout.type = PatternLayout
+appender.out.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
diff --git a/core/camel-allcomponents/pom.xml b/core/camel-allcomponents/pom.xml
index 4d24103..327d574 100644
--- a/core/camel-allcomponents/pom.xml
+++ b/core/camel-allcomponents/pom.xml
@@ -188,6 +188,10 @@
</dependency>
<dependency>
<groupId>org.apache.camel</groupId>
+ <artifactId>camel-aws2-kinesis</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
<artifactId>camel-aws2-kms</artifactId>
</dependency>
<dependency>
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
index b681fbe..35ac9c2 100644
--- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/ComponentsBuilderFactory.java
@@ -541,6 +541,30 @@ public interface ComponentsBuilderFactory {
return org.apache.camel.builder.component.dsl.Aws2IamComponentBuilderFactory.aws2Iam();
}
/**
+ * AWS 2 Kinesis (camel-aws2-kinesis)
+ * The aws-kinesis component is for consuming and producing records from
+ * Amazon Kinesis Streams.
+ *
+ * Category: cloud,messaging
+ * Since: 3.2
+ * Maven coordinates: org.apache.camel:camel-aws2-kinesis
+ */
+ static org.apache.camel.builder.component.dsl.Aws2KinesisComponentBuilderFactory.Aws2KinesisComponentBuilder aws2Kinesis() {
+ return org.apache.camel.builder.component.dsl.Aws2KinesisComponentBuilderFactory.aws2Kinesis();
+ }
+ /**
+ * AWS 2 Kinesis Firehose (camel-aws2-kinesis)
+ * The aws-kinesis-firehose component is used for producing Amazon's Kinesis
+ * Firehose streams.
+ *
+ * Category: cloud,messaging
+ * Since: 3.2
+ * Maven coordinates: org.apache.camel:camel-aws2-kinesis
+ */
+ static org.apache.camel.builder.component.dsl.Aws2KinesisFirehoseComponentBuilderFactory.Aws2KinesisFirehoseComponentBuilder aws2KinesisFirehose() {
+ return org.apache.camel.builder.component.dsl.Aws2KinesisFirehoseComponentBuilderFactory.aws2KinesisFirehose();
+ }
+ /**
* AWS 2 KMS (camel-aws2-kms)
* The aws2-kms is used for managing Amazon KMS
*
diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisComponentBuilderFactory.java
new file mode 100644
index 0000000..3391c01
--- /dev/null
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisComponentBuilderFactory.java
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.builder.component.dsl;
+
+import javax.annotation.Generated;
+import org.apache.camel.Component;
+import org.apache.camel.builder.component.AbstractComponentBuilder;
+import org.apache.camel.builder.component.ComponentBuilder;
+import org.apache.camel.component.aws2.kinesis.Kinesis2Component;
+
+/**
+ * The aws-kinesis component is for consuming and producing records from Amazon
+ * Kinesis Streams.
+ *
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.ComponentDslMojo")
+public interface Aws2KinesisComponentBuilderFactory {
+
+ /**
+ * AWS 2 Kinesis (camel-aws2-kinesis)
+ * The aws-kinesis component is for consuming and producing records from
+ * Amazon Kinesis Streams.
+ *
+ * Category: cloud,messaging
+ * Since: 3.2
+ * Maven coordinates: org.apache.camel:camel-aws2-kinesis
+ */
+ static Aws2KinesisComponentBuilder aws2Kinesis() {
+ return new Aws2KinesisComponentBuilderImpl();
+ }
+
+ /**
+ * Builder for the AWS 2 Kinesis component.
+ */
+ interface Aws2KinesisComponentBuilder
+ extends
+ ComponentBuilder<Kinesis2Component> {
+ /**
+ * Amazon AWS Access Key.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default Aws2KinesisComponentBuilder accessKey(java.lang.String accessKey) {
+ doSetProperty("accessKey", accessKey);
+ return this;
+ }
+ /**
+ * Amazon AWS Region.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default Aws2KinesisComponentBuilder region(java.lang.String region) {
+ doSetProperty("region", region);
+ return this;
+ }
+ /**
+ * Amazon AWS Secret Key.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: common
+ */
+ default Aws2KinesisComponentBuilder secretKey(java.lang.String secretKey) {
+ doSetProperty("secretKey", secretKey);
+ return this;
+ }
+ /**
+ * Allows for bridging the consumer to the Camel routing Error Handler,
+ * which mean any exceptions occurred while the consumer is trying to
+ * pickup incoming messages, or the likes, will now be processed as a
+ * message and handled by the routing Error Handler. By default the
+ * consumer will use the org.apache.camel.spi.ExceptionHandler to deal
+ * with exceptions, that will be logged at WARN or ERROR level and
+ * ignored.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: consumer
+ */
+ default Aws2KinesisComponentBuilder bridgeErrorHandler(
+ boolean bridgeErrorHandler) {
+ doSetProperty("bridgeErrorHandler", bridgeErrorHandler);
+ return this;
+ }
+ /**
+ * 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 starting the
+ * producer may take a little time and prolong the total processing time
+ * of the processing.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer
+ */
+ default Aws2KinesisComponentBuilder lazyStartProducer(
+ boolean lazyStartProducer) {
+ doSetProperty("lazyStartProducer", lazyStartProducer);
+ return this;
+ }
+ /**
+ * Whether the component should use basic property binding (Camel 2.x)
+ * or the newer property binding with additional capabilities.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default Aws2KinesisComponentBuilder basicPropertyBinding(
+ boolean basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * The AWS S3 default configuration.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.aws2.kinesis.Kinesis2Configuration</code> type.
+ *
+ * Group: advanced
+ */
+ default Aws2KinesisComponentBuilder configuration(
+ org.apache.camel.component.aws2.kinesis.Kinesis2Configuration configuration) {
+ doSetProperty("configuration", configuration);
+ return this;
+ }
+ }
+
+ class Aws2KinesisComponentBuilderImpl
+ extends
+ AbstractComponentBuilder<Kinesis2Component>
+ implements
+ Aws2KinesisComponentBuilder {
+ @Override
+ protected Kinesis2Component buildConcreteComponent() {
+ return new Kinesis2Component();
+ }
+ @Override
+ protected boolean setPropertyOnComponent(
+ Component component,
+ String name,
+ Object value) {
+ switch (name) {
+ case "accessKey": ((Kinesis2Component) component).setAccessKey((java.lang.String) value); return true;
+ case "region": ((Kinesis2Component) component).setRegion((java.lang.String) value); return true;
+ case "secretKey": ((Kinesis2Component) component).setSecretKey((java.lang.String) value); return true;
+ case "bridgeErrorHandler": ((Kinesis2Component) component).setBridgeErrorHandler((boolean) value); return true;
+ case "lazyStartProducer": ((Kinesis2Component) component).setLazyStartProducer((boolean) value); return true;
+ case "basicPropertyBinding": ((Kinesis2Component) component).setBasicPropertyBinding((boolean) value); return true;
+ case "configuration": ((Kinesis2Component) component).setConfiguration((org.apache.camel.component.aws2.kinesis.Kinesis2Configuration) value); return true;
+ default: return false;
+ }
+ }
+ }
+}
\ No newline at end of file
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
new file mode 100644
index 0000000..739e646
--- /dev/null
+++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisFirehoseComponentBuilderFactory.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.builder.component.dsl;
+
+import javax.annotation.Generated;
+import org.apache.camel.Component;
+import org.apache.camel.builder.component.AbstractComponentBuilder;
+import org.apache.camel.builder.component.ComponentBuilder;
+import org.apache.camel.component.aws2.firehose.KinesisFirehose2Component;
+
+/**
+ * The aws-kinesis-firehose component is used for producing Amazon's Kinesis
+ * Firehose streams.
+ *
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.ComponentDslMojo")
+public interface Aws2KinesisFirehoseComponentBuilderFactory {
+
+ /**
+ * AWS 2 Kinesis Firehose (camel-aws2-kinesis)
+ * The aws-kinesis-firehose component is used for producing Amazon's Kinesis
+ * Firehose streams.
+ *
+ * Category: cloud,messaging
+ * Since: 3.2
+ * Maven coordinates: org.apache.camel:camel-aws2-kinesis
+ */
+ static Aws2KinesisFirehoseComponentBuilder aws2KinesisFirehose() {
+ return new Aws2KinesisFirehoseComponentBuilderImpl();
+ }
+
+ /**
+ * Builder for the AWS 2 Kinesis Firehose component.
+ */
+ interface Aws2KinesisFirehoseComponentBuilder
+ extends
+ ComponentBuilder<KinesisFirehose2Component> {
+ /**
+ * Amazon AWS Access Key.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ */
+ default Aws2KinesisFirehoseComponentBuilder accessKey(
+ java.lang.String accessKey) {
+ doSetProperty("accessKey", accessKey);
+ return this;
+ }
+ /**
+ * 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 starting the
+ * producer may take a little time and prolong the total processing time
+ * of the processing.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: producer
+ */
+ default Aws2KinesisFirehoseComponentBuilder lazyStartProducer(
+ boolean lazyStartProducer) {
+ doSetProperty("lazyStartProducer", lazyStartProducer);
+ return this;
+ }
+ /**
+ * Amazon AWS Region.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ */
+ default Aws2KinesisFirehoseComponentBuilder region(
+ java.lang.String region) {
+ doSetProperty("region", region);
+ return this;
+ }
+ /**
+ * Amazon AWS Secret Key.
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: producer
+ */
+ default Aws2KinesisFirehoseComponentBuilder secretKey(
+ java.lang.String secretKey) {
+ doSetProperty("secretKey", secretKey);
+ return this;
+ }
+ /**
+ * Whether the component should use basic property binding (Camel 2.x)
+ * or the newer property binding with additional capabilities.
+ *
+ * The option is a: <code>boolean</code> type.
+ *
+ * Default: false
+ * Group: advanced
+ */
+ default Aws2KinesisFirehoseComponentBuilder basicPropertyBinding(
+ boolean basicPropertyBinding) {
+ doSetProperty("basicPropertyBinding", basicPropertyBinding);
+ return this;
+ }
+ /**
+ * The AWS Kinesis Firehose default configuration.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration</code> type.
+ *
+ * Group: advanced
+ */
+ default Aws2KinesisFirehoseComponentBuilder configuration(
+ org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration configuration) {
+ doSetProperty("configuration", configuration);
+ return this;
+ }
+ }
+
+ class Aws2KinesisFirehoseComponentBuilderImpl
+ extends
+ AbstractComponentBuilder<KinesisFirehose2Component>
+ implements
+ Aws2KinesisFirehoseComponentBuilder {
+ @Override
+ protected KinesisFirehose2Component buildConcreteComponent() {
+ return new KinesisFirehose2Component();
+ }
+ @Override
+ protected boolean setPropertyOnComponent(
+ Component component,
+ String name,
+ Object value) {
+ switch (name) {
+ case "accessKey": ((KinesisFirehose2Component) component).setAccessKey((java.lang.String) value); return true;
+ case "lazyStartProducer": ((KinesisFirehose2Component) component).setLazyStartProducer((boolean) value); return true;
+ case "region": ((KinesisFirehose2Component) component).setRegion((java.lang.String) value); return true;
+ case "secretKey": ((KinesisFirehose2Component) component).setSecretKey((java.lang.String) value); return true;
+ case "basicPropertyBinding": ((KinesisFirehose2Component) component).setBasicPropertyBinding((boolean) value); return true;
+ case "configuration": ((KinesisFirehose2Component) component).setConfiguration((org.apache.camel.component.aws2.firehose.KinesisFirehose2Configuration) value); return true;
+ default: return false;
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/core/camel-componentdsl/src/generated/resources/metadata.json b/core/camel-componentdsl/src/generated/resources/metadata.json
index b80f1b0..48cc4e3 100644
--- a/core/camel-componentdsl/src/generated/resources/metadata.json
+++ b/core/camel-componentdsl/src/generated/resources/metadata.json
@@ -501,6 +501,46 @@
"artifactId": "camel-aws2-iam",
"version": "3.2.0-SNAPSHOT"
},
+ "Aws2KinesisComponentBuilderFactory": {
+ "kind": "component",
+ "scheme": "aws2-kinesis",
+ "extendsScheme": "",
+ "syntax": "aws2-kinesis:streamName",
+ "title": "AWS 2 Kinesis",
+ "description": "The aws-kinesis component is for consuming and producing records from Amazon Kinesis Streams.",
+ "label": "cloud,messaging",
+ "deprecated": false,
+ "deprecationNote": "cloud,messaging",
+ "async": false,
+ "consumerOnly": false,
+ "producerOnly": false,
+ "lenientProperties": false,
+ "javaType": "org.apache.camel.component.aws2.kinesis.Kinesis2Component",
+ "firstVersion": "3.2.0",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-aws2-kinesis",
+ "version": "3.2.0-SNAPSHOT"
+ },
+ "Aws2KinesisFirehoseComponentBuilderFactory": {
+ "kind": "component",
+ "scheme": "aws2-kinesis-firehose",
+ "extendsScheme": "",
+ "syntax": "aws2-kinesis-firehose:streamName",
+ "title": "AWS 2 Kinesis Firehose",
+ "description": "The aws-kinesis-firehose component is used for producing Amazon's Kinesis Firehose streams.",
+ "label": "cloud,messaging",
+ "deprecated": false,
+ "deprecationNote": "cloud,messaging",
+ "async": false,
+ "consumerOnly": false,
+ "producerOnly": true,
+ "lenientProperties": false,
+ "javaType": "org.apache.camel.component.aws2.firehose.KinesisFirehose2Component",
+ "firstVersion": "3.2.0",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-aws2-kinesis",
+ "version": "3.2.0-SNAPSHOT"
+ },
"Aws2KmsComponentBuilderFactory": {
"kind": "component",
"scheme": "aws2-kms",