This is an automated email from the ASF dual-hosted git repository. acosentino pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 8a2116fb4b91ed040a7134019b9b2d278d3b0919 Author: Andrea Cosentino <anco...@gmail.com> AuthorDate: Tue Jun 25 11:29:31 2024 +0200 CAMEL-20912, CAMEL-20913 - Camel-AWS-Kinesis: KCL Consumers add parameters for passing Cloudwatch and DynamoDB Clients and for profile/default/session credentials Providers Signed-off-by: Andrea Cosentino <anco...@gmail.com> --- .../aws2/kinesis/Kinesis2ComponentConfigurer.java | 12 +++ .../aws2/kinesis/Kinesis2EndpointConfigurer.java | 12 +++ .../aws2/kinesis/Kinesis2EndpointUriFactory.java | 4 +- .../camel/component/aws2/kinesis/aws2-kinesis.json | 84 ++++++++-------- .../aws2/kinesis/KclKinesis2Consumer.java | 112 ++++++++++++++++----- .../aws2/kinesis/Kinesis2Configuration.java | 26 +++++ .../kinesis/integration/KclKinesisConsumerIT.java | 6 +- 7 files changed, 186 insertions(+), 70 deletions(-) diff --git a/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentConfigurer.java b/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentConfigurer.java index 47834f557c6..8f6412022df 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentConfigurer.java +++ b/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2ComponentConfigurer.java @@ -42,7 +42,11 @@ public class Kinesis2ComponentConfigurer extends PropertyConfigurerSupport imple case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true; case "cborenabled": case "cborEnabled": getOrCreateConfiguration(target).setCborEnabled(property(camelContext, boolean.class, value)); return true; + case "cloudwatchasyncclient": + case "cloudWatchAsyncClient": getOrCreateConfiguration(target).setCloudWatchAsyncClient(property(camelContext, software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient.class, value)); return true; case "configuration": target.setConfiguration(property(camelContext, org.apache.camel.component.aws2.kinesis.Kinesis2Configuration.class, value)); return true; + case "dynamodbasyncclient": + case "dynamoDbAsyncClient": getOrCreateConfiguration(target).setDynamoDbAsyncClient(property(camelContext, software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient.class, value)); return true; case "healthcheckconsumerenabled": case "healthCheckConsumerEnabled": target.setHealthCheckConsumerEnabled(property(camelContext, boolean.class, value)); return true; case "healthcheckproducerenabled": @@ -112,7 +116,11 @@ public class Kinesis2ComponentConfigurer extends PropertyConfigurerSupport imple case "bridgeErrorHandler": return boolean.class; case "cborenabled": case "cborEnabled": return boolean.class; + case "cloudwatchasyncclient": + case "cloudWatchAsyncClient": return software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient.class; case "configuration": return org.apache.camel.component.aws2.kinesis.Kinesis2Configuration.class; + case "dynamodbasyncclient": + case "dynamoDbAsyncClient": return software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient.class; case "healthcheckconsumerenabled": case "healthCheckConsumerEnabled": return boolean.class; case "healthcheckproducerenabled": @@ -178,7 +186,11 @@ public class Kinesis2ComponentConfigurer extends PropertyConfigurerSupport imple case "bridgeErrorHandler": return target.isBridgeErrorHandler(); case "cborenabled": case "cborEnabled": return getOrCreateConfiguration(target).isCborEnabled(); + case "cloudwatchasyncclient": + case "cloudWatchAsyncClient": return getOrCreateConfiguration(target).getCloudWatchAsyncClient(); case "configuration": return target.getConfiguration(); + case "dynamodbasyncclient": + case "dynamoDbAsyncClient": return getOrCreateConfiguration(target).getDynamoDbAsyncClient(); case "healthcheckconsumerenabled": case "healthCheckConsumerEnabled": return target.isHealthCheckConsumerEnabled(); case "healthcheckproducerenabled": diff --git a/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointConfigurer.java b/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointConfigurer.java index c656b45554a..43aa3bdfddd 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointConfigurer.java +++ b/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointConfigurer.java @@ -39,7 +39,11 @@ public class Kinesis2EndpointConfigurer extends PropertyConfigurerSupport implem case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true; case "cborenabled": case "cborEnabled": target.getConfiguration().setCborEnabled(property(camelContext, boolean.class, value)); return true; + case "cloudwatchasyncclient": + case "cloudWatchAsyncClient": target.getConfiguration().setCloudWatchAsyncClient(property(camelContext, software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient.class, value)); return true; case "delay": target.setDelay(property(camelContext, long.class, value)); return true; + case "dynamodbasyncclient": + case "dynamoDbAsyncClient": target.getConfiguration().setDynamoDbAsyncClient(property(camelContext, software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient.class, value)); return true; case "exceptionhandler": case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true; case "exchangepattern": @@ -135,7 +139,11 @@ public class Kinesis2EndpointConfigurer extends PropertyConfigurerSupport implem case "bridgeErrorHandler": return boolean.class; case "cborenabled": case "cborEnabled": return boolean.class; + case "cloudwatchasyncclient": + case "cloudWatchAsyncClient": return software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient.class; case "delay": return long.class; + case "dynamodbasyncclient": + case "dynamoDbAsyncClient": return software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient.class; case "exceptionhandler": case "exceptionHandler": return org.apache.camel.spi.ExceptionHandler.class; case "exchangepattern": @@ -227,7 +235,11 @@ public class Kinesis2EndpointConfigurer extends PropertyConfigurerSupport implem case "bridgeErrorHandler": return target.isBridgeErrorHandler(); case "cborenabled": case "cborEnabled": return target.getConfiguration().isCborEnabled(); + case "cloudwatchasyncclient": + case "cloudWatchAsyncClient": return target.getConfiguration().getCloudWatchAsyncClient(); case "delay": return target.getDelay(); + case "dynamodbasyncclient": + case "dynamoDbAsyncClient": return target.getConfiguration().getDynamoDbAsyncClient(); case "exceptionhandler": case "exceptionHandler": return target.getExceptionHandler(); case "exchangepattern": diff --git a/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointUriFactory.java b/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointUriFactory.java index ca416a764b9..40f64207b07 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointUriFactory.java +++ b/components/camel-aws/camel-aws2-kinesis/src/generated/java/org/apache/camel/component/aws2/kinesis/Kinesis2EndpointUriFactory.java @@ -23,7 +23,7 @@ public class Kinesis2EndpointUriFactory extends org.apache.camel.support.compone private static final Set<String> SECRET_PROPERTY_NAMES; private static final Set<String> MULTI_VALUE_PREFIXES; static { - Set<String> props = new HashSet<>(45); + Set<String> props = new HashSet<>(47); props.add("accessKey"); props.add("amazonKinesisClient"); props.add("asyncClient"); @@ -32,7 +32,9 @@ public class Kinesis2EndpointUriFactory extends org.apache.camel.support.compone props.add("backoffMultiplier"); props.add("bridgeErrorHandler"); props.add("cborEnabled"); + props.add("cloudWatchAsyncClient"); props.add("delay"); + props.add("dynamoDbAsyncClient"); props.add("exceptionHandler"); props.add("exchangePattern"); props.add("greedy"); diff --git a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json index 355c628c0c6..ce904d5350e 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json +++ b/components/camel-aws/camel-aws2-kinesis/src/generated/resources/META-INF/org/apache/camel/component/aws2/kinesis/aws2-kinesis.json @@ -39,20 +39,22 @@ "amazonKinesisClient": { "index": 13, "kind": "property", "displayName": "Amazon Kinesis Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.kinesis.KinesisClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon Kinesis clien [...] "asyncClient": { "index": 14, "kind": "property", "displayName": "Async Client", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KinesisAsyncClient instance set it to true" }, "autowiredEnabled": { "index": 15, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching [...] - "useKclConsumers": { "index": 16, "kind": "property", "displayName": "Use Kcl Consumers", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consumer set it to true" }, - "healthCheckConsumerEnabled": { "index": 17, "kind": "property", "displayName": "Health Check Consumer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all consumer based health checks from this component" }, - "healthCheckProducerEnabled": { "index": 18, "kind": "property", "displayName": "Health Check Producer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all producer based health checks from this component. Notice: Camel has by default disabled all producer based health-checks. You can turn on produce [...] - "proxyHost": { "index": 19, "kind": "property", "displayName": "Proxy Host", "group": "proxy", "label": "proxy", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": 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": { "index": 20, "kind": "property", "displayName": "Proxy Port", "group": "proxy", "label": "proxy", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": 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": { "index": 21, "kind": "property", "displayName": "Proxy Protocol", "group": "proxy", "label": "proxy", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "To define a proxy protoc [...] - "accessKey": { "index": 22, "kind": "property", "displayName": "Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Access Key" }, - "profileCredentialsName": { "index": 23, "kind": "property", "displayName": "Profile Credentials Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If using a profile credentials provider this parameter will set the profi [...] - "secretKey": { "index": 24, "kind": "property", "displayName": "Secret Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Secret Key" }, - "sessionToken": { "index": 25, "kind": "property", "displayName": "Session Token", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Session Token used when the user needs to assume a IAM role" }, - "trustAllCertificates": { "index": 26, "kind": "property", "displayName": "Trust All Certificates", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to trust all certificates in case of overriding the [...] - "useDefaultCredentialsProvider": { "index": 27, "kind": "property", "displayName": "Use Default Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expec [...] - "useProfileCredentialsProvider": { "index": 28, "kind": "property", "displayName": "Use Profile Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expec [...] - "useSessionCredentials": { "index": 29, "kind": "property", "displayName": "Use Session Credentials", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expect to use Session [...] + "cloudWatchAsyncClient": { "index": 16, "kind": "property", "displayName": "Cloud Watch Async Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consum [...] + "dynamoDbAsyncClient": { "index": 17, "kind": "property", "displayName": "Dynamo Db Async Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consumer, we c [...] + "useKclConsumers": { "index": 18, "kind": "property", "displayName": "Use Kcl Consumers", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consumer set it to true" }, + "healthCheckConsumerEnabled": { "index": 19, "kind": "property", "displayName": "Health Check Consumer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all consumer based health checks from this component" }, + "healthCheckProducerEnabled": { "index": 20, "kind": "property", "displayName": "Health Check Producer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all producer based health checks from this component. Notice: Camel has by default disabled all producer based health-checks. You can turn on produce [...] + "proxyHost": { "index": 21, "kind": "property", "displayName": "Proxy Host", "group": "proxy", "label": "proxy", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": 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": { "index": 22, "kind": "property", "displayName": "Proxy Port", "group": "proxy", "label": "proxy", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": 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": { "index": 23, "kind": "property", "displayName": "Proxy Protocol", "group": "proxy", "label": "proxy", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "To define a proxy protoc [...] + "accessKey": { "index": 24, "kind": "property", "displayName": "Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Access Key" }, + "profileCredentialsName": { "index": 25, "kind": "property", "displayName": "Profile Credentials Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If using a profile credentials provider this parameter will set the profi [...] + "secretKey": { "index": 26, "kind": "property", "displayName": "Secret Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Secret Key" }, + "sessionToken": { "index": 27, "kind": "property", "displayName": "Session Token", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Session Token used when the user needs to assume a IAM role" }, + "trustAllCertificates": { "index": 28, "kind": "property", "displayName": "Trust All Certificates", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to trust all certificates in case of overriding the [...] + "useDefaultCredentialsProvider": { "index": 29, "kind": "property", "displayName": "Use Default Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expec [...] + "useProfileCredentialsProvider": { "index": 30, "kind": "property", "displayName": "Use Profile Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expec [...] + "useSessionCredentials": { "index": 31, "kind": "property", "displayName": "Use Session Credentials", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expect to use Session [...] }, "headers": { "CamelAwsKinesisSequenceNumber": { "index": 0, "kind": "header", "displayName": "", "group": "common", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The sequence number of the record, as defined in http:\/\/docs.aws.amazon.com\/kinesis\/latest\/APIReference\/API_PutRecord.html#API_PutRecord_ResponseSyntaxResponse Syntax", "constantName": "org.apache.camel.component.aws2.kinesis.Ki [...] @@ -82,31 +84,33 @@ "lazyStartProducer": { "index": 16, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": 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 produ [...] "amazonKinesisClient": { "index": 17, "kind": "parameter", "displayName": "Amazon Kinesis Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.kinesis.KinesisClient", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon Kinesis clie [...] "asyncClient": { "index": 18, "kind": "parameter", "displayName": "Async Client", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KinesisAsyncClient instance set it to true" }, - "useKclConsumers": { "index": 19, "kind": "parameter", "displayName": "Use Kcl Consumers", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consumer set it to true" }, - "proxyHost": { "index": 20, "kind": "parameter", "displayName": "Proxy Host", "group": "proxy", "label": "proxy", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": 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": { "index": 21, "kind": "parameter", "displayName": "Proxy Port", "group": "proxy", "label": "proxy", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": 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": { "index": 22, "kind": "parameter", "displayName": "Proxy Protocol", "group": "proxy", "label": "proxy", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "To define a proxy proto [...] - "backoffErrorThreshold": { "index": 23, "kind": "parameter", "displayName": "Backoff Error Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in." }, - "backoffIdleThreshold": { "index": 24, "kind": "parameter", "displayName": "Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The number of subsequent idle polls that should happen before the backoffMultipler should kick-in." }, - "backoffMultiplier": { "index": 25, "kind": "parameter", "displayName": "Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": 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 h [...] - "delay": { "index": 26, "kind": "parameter", "displayName": "Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 500, "description": "Milliseconds before the next poll." }, - "greedy": { "index": 27, "kind": "parameter", "displayName": "Greedy", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": 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": { "index": 28, "kind": "parameter", "displayName": "Initial Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "description": "Milliseconds before the first poll starts." }, - "repeatCount": { "index": 29, "kind": "parameter", "displayName": "Repeat Count", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": 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": { "index": 30, "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, "autowired": false, "secret": false, "defaultValue": "TRACE", "description": "The consumer logs a start\/complete log line when it polls. This option allows you to configure the l [...] - "scheduledExecutorService": { "index": 31, "kind": "parameter", "displayName": "Scheduled Executor Service", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.util.concurrent.ScheduledExecutorService", "deprecated": false, "autowired": 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": { "index": 32, "kind": "parameter", "displayName": "Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.lang.Object", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "none", "description": "To use a cron scheduler from either camel-spring or camel-quartz component. Use value spring or quartz for built in scheduler" }, - "schedulerProperties": { "index": 33, "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, "autowired": false, "secret": false, "description": "To configure additional properties when using a custom scheduler or any of the Quartz, Spring based scheduler." }, - "startScheduler": { "index": 34, "kind": "parameter", "displayName": "Start Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether the scheduler should be auto started." }, - "timeUnit": { "index": 35, "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, "autowired": false, "secret": false, "defaultValue": "MILLISECONDS", "description": "Time unit for initialDelay and delay options." }, - "useFixedDelay": { "index": 36, "kind": "parameter", "displayName": "Use Fixed Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in JDK for details." }, - "accessKey": { "index": 37, "kind": "parameter", "displayName": "Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Access Key" }, - "profileCredentialsName": { "index": 38, "kind": "parameter", "displayName": "Profile Credentials Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If using a profile credentials provider this parameter will set the prof [...] - "secretKey": { "index": 39, "kind": "parameter", "displayName": "Secret Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Secret Key" }, - "sessionToken": { "index": 40, "kind": "parameter", "displayName": "Session Token", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Session Token used when the user needs to assume a IAM role" }, - "trustAllCertificates": { "index": 41, "kind": "parameter", "displayName": "Trust All Certificates", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to trust all certificates in case of overriding th [...] - "useDefaultCredentialsProvider": { "index": 42, "kind": "parameter", "displayName": "Use Default Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expe [...] - "useProfileCredentialsProvider": { "index": 43, "kind": "parameter", "displayName": "Use Profile Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expe [...] - "useSessionCredentials": { "index": 44, "kind": "parameter", "displayName": "Use Session Credentials", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expect to use Session [...] + "cloudWatchAsyncClient": { "index": 19, "kind": "parameter", "displayName": "Cloud Watch Async Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consu [...] + "dynamoDbAsyncClient": { "index": 20, "kind": "parameter", "displayName": "Dynamo Db Async Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consumer, we [...] + "useKclConsumers": { "index": 21, "kind": "parameter", "displayName": "Use Kcl Consumers", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to a KCL Consumer set it to true" }, + "proxyHost": { "index": 22, "kind": "parameter", "displayName": "Proxy Host", "group": "proxy", "label": "proxy", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": 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": { "index": 23, "kind": "parameter", "displayName": "Proxy Port", "group": "proxy", "label": "proxy", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": 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": { "index": 24, "kind": "parameter", "displayName": "Proxy Protocol", "group": "proxy", "label": "proxy", "required": false, "type": "object", "javaType": "software.amazon.awssdk.core.Protocol", "enum": [ "HTTP", "HTTPS" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "HTTPS", "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "To define a proxy proto [...] + "backoffErrorThreshold": { "index": 25, "kind": "parameter", "displayName": "Backoff Error Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in." }, + "backoffIdleThreshold": { "index": 26, "kind": "parameter", "displayName": "Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The number of subsequent idle polls that should happen before the backoffMultipler should kick-in." }, + "backoffMultiplier": { "index": 27, "kind": "parameter", "displayName": "Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": 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 h [...] + "delay": { "index": 28, "kind": "parameter", "displayName": "Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 500, "description": "Milliseconds before the next poll." }, + "greedy": { "index": 29, "kind": "parameter", "displayName": "Greedy", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": 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": { "index": 30, "kind": "parameter", "displayName": "Initial Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "description": "Milliseconds before the first poll starts." }, + "repeatCount": { "index": 31, "kind": "parameter", "displayName": "Repeat Count", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": 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": { "index": 32, "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, "autowired": false, "secret": false, "defaultValue": "TRACE", "description": "The consumer logs a start\/complete log line when it polls. This option allows you to configure the l [...] + "scheduledExecutorService": { "index": 33, "kind": "parameter", "displayName": "Scheduled Executor Service", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.util.concurrent.ScheduledExecutorService", "deprecated": false, "autowired": 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": { "index": 34, "kind": "parameter", "displayName": "Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.lang.Object", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "none", "description": "To use a cron scheduler from either camel-spring or camel-quartz component. Use value spring or quartz for built in scheduler" }, + "schedulerProperties": { "index": 35, "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, "autowired": false, "secret": false, "description": "To configure additional properties when using a custom scheduler or any of the Quartz, Spring based scheduler." }, + "startScheduler": { "index": 36, "kind": "parameter", "displayName": "Start Scheduler", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether the scheduler should be auto started." }, + "timeUnit": { "index": 37, "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, "autowired": false, "secret": false, "defaultValue": "MILLISECONDS", "description": "Time unit for initialDelay and delay options." }, + "useFixedDelay": { "index": 38, "kind": "parameter", "displayName": "Use Fixed Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Controls if fixed delay or fixed rate is used. See ScheduledExecutorService in JDK for details." }, + "accessKey": { "index": 39, "kind": "parameter", "displayName": "Access Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Access Key" }, + "profileCredentialsName": { "index": 40, "kind": "parameter", "displayName": "Profile Credentials Name", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If using a profile credentials provider this parameter will set the prof [...] + "secretKey": { "index": 41, "kind": "parameter", "displayName": "Secret Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Secret Key" }, + "sessionToken": { "index": 42, "kind": "parameter", "displayName": "Session Token", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Amazon AWS Session Token used when the user needs to assume a IAM role" }, + "trustAllCertificates": { "index": 43, "kind": "parameter", "displayName": "Trust All Certificates", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "If we want to trust all certificates in case of overriding th [...] + "useDefaultCredentialsProvider": { "index": 44, "kind": "parameter", "displayName": "Use Default Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expe [...] + "useProfileCredentialsProvider": { "index": 45, "kind": "parameter", "displayName": "Use Profile Credentials Provider", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expe [...] + "useSessionCredentials": { "index": 46, "kind": "parameter", "displayName": "Use Session Credentials", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.aws2.kinesis.Kinesis2Configuration", "configurationField": "configuration", "description": "Set whether the Kinesis client should expect to use Session [...] } } diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java index e3c05d06682..d8c3549edb1 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java +++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/KclKinesis2Consumer.java @@ -17,16 +17,28 @@ package org.apache.camel.component.aws2.kinesis; import java.util.UUID; -import java.util.concurrent.*; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.support.DefaultConsumer; +import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; +import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClientBuilder; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClientBuilder; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; import software.amazon.kinesis.common.ConfigsBuilder; import software.amazon.kinesis.coordinator.Scheduler; import software.amazon.kinesis.exceptions.InvalidStateException; @@ -69,8 +81,58 @@ public class KclKinesis2Consumer extends DefaultConsumer { protected void doStart() throws Exception { super.doStart(); LOG.debug("Starting KCL Consumer"); + DynamoDbAsyncClient dynamoByAsyncClient = null; + CloudWatchAsyncClient cloudWatchAsyncClient = null; + KinesisAsyncClient kinesisAsyncClient = getEndpoint().getAsyncClient(); + Kinesis2Configuration configuration = getEndpoint().getConfiguration(); + if (ObjectHelper.isEmpty(getEndpoint().getConfiguration().getDynamoDbAsyncClient())) { + DynamoDbAsyncClientBuilder clientBuilder = DynamoDbAsyncClient.builder(); + if (ObjectHelper.isNotEmpty(configuration.getAccessKey()) + && ObjectHelper.isNotEmpty(configuration.getSecretKey())) { + clientBuilder = clientBuilder.credentialsProvider(StaticCredentialsProvider.create( + AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey()))); + } else if (ObjectHelper.isNotEmpty(configuration.getProfileCredentialsName())) { + clientBuilder = clientBuilder + .credentialsProvider( + ProfileCredentialsProvider.create(configuration.getProfileCredentialsName())); + } else if (ObjectHelper.isNotEmpty(configuration.getAccessKey()) + && ObjectHelper.isNotEmpty(configuration.getSecretKey()) + && ObjectHelper.isNotEmpty(configuration.getSessionToken())) { + clientBuilder = clientBuilder.credentialsProvider(StaticCredentialsProvider + .create(AwsSessionCredentials.create(configuration.getAccessKey(), configuration.getSecretKey(), + configuration.getSessionToken()))); + } + if (ObjectHelper.isNotEmpty(configuration.getRegion())) { + clientBuilder = clientBuilder.region(Region.of(configuration.getRegion())); + } + dynamoByAsyncClient + = clientBuilder.build(); + } + if (ObjectHelper.isEmpty(getEndpoint().getConfiguration().getCloudWatchAsyncClient())) { + CloudWatchAsyncClientBuilder clientBuilder = CloudWatchAsyncClient.builder(); + if (ObjectHelper.isNotEmpty(configuration.getAccessKey()) + && ObjectHelper.isNotEmpty(configuration.getSecretKey())) { + clientBuilder = clientBuilder.credentialsProvider(StaticCredentialsProvider.create( + AwsBasicCredentials.create(configuration.getAccessKey(), configuration.getSecretKey()))); + } else if (ObjectHelper.isNotEmpty(configuration.getProfileCredentialsName())) { + clientBuilder = clientBuilder + .credentialsProvider( + ProfileCredentialsProvider.create(configuration.getProfileCredentialsName())); + } else if (ObjectHelper.isNotEmpty(configuration.getAccessKey()) + && ObjectHelper.isNotEmpty(configuration.getSecretKey()) + && ObjectHelper.isNotEmpty(configuration.getSessionToken())) { + clientBuilder = clientBuilder.credentialsProvider(StaticCredentialsProvider + .create(AwsSessionCredentials.create(configuration.getAccessKey(), configuration.getSecretKey(), + configuration.getSessionToken()))); + } + if (ObjectHelper.isNotEmpty(configuration.getRegion())) { + clientBuilder = clientBuilder.region(Region.of(configuration.getRegion())); + } + cloudWatchAsyncClient = clientBuilder.build(); + } this.executor = this.getEndpoint().createExecutor(); - this.executor.submit(new KclKinesis2Consumer.KclKinesisConsumingTask(this.getEndpoint().getConfiguration())); + this.executor.submit(new KclKinesisConsumingTask( + configuration.getStreamName(), kinesisAsyncClient, dynamoByAsyncClient, cloudWatchAsyncClient)); } @Override @@ -103,7 +165,7 @@ public class KclKinesis2Consumer extends DefaultConsumer { class CamelKinesisRecordProcessor implements ShardRecordProcessor { - private static final Logger log = LoggerFactory.getLogger(CamelKinesisRecordProcessor.class); + private static final Logger LOG = LoggerFactory.getLogger(CamelKinesisRecordProcessor.class); private String shardId; private Kinesis2Endpoint endpoint; @@ -115,13 +177,13 @@ public class KclKinesis2Consumer extends DefaultConsumer { @Override public void initialize(InitializationInput initializationInput) { shardId = initializationInput.shardId(); - log.debug("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber()); + LOG.debug("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber()); } @Override public void processRecords(ProcessRecordsInput processRecordsInput) { try { - log.debug("Processing {} record(s)", processRecordsInput.records().size()); + LOG.debug("Processing {} record(s)", processRecordsInput.records().size()); processRecordsInput.records() .forEach(r -> { try { @@ -131,32 +193,32 @@ public class KclKinesis2Consumer extends DefaultConsumer { } }); } catch (Throwable t) { - log.error("Caught throwable while processing records. Aborting."); + LOG.error("Caught throwable while processing records. Aborting."); } } @Override public void leaseLost(LeaseLostInput leaseLostInput) { - log.debug("Lost lease, so terminating."); + LOG.debug("Lost lease, so terminating."); } @Override public void shardEnded(ShardEndedInput shardEndedInput) { try { - log.debug("Reached shard end checkpointing."); + LOG.debug("Reached shard end checkpointing."); shardEndedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { - log.error("Exception while checkpointing at shard end. Giving up.", e); + LOG.error("Exception while checkpointing at shard end. Giving up.", e); } } @Override public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) { try { - log.debug("Scheduler is shutting down, checkpointing."); + LOG.debug("Scheduler is shutting down, checkpointing."); shutdownRequestedInput.checkpointer().checkpoint(); } catch (ShutdownException | InvalidStateException e) { - log.error("Exception while checkpointing at requested shutdown. Giving up.", e); + LOG.error("Exception while checkpointing at requested shutdown. Giving up.", e); } } @@ -177,28 +239,28 @@ public class KclKinesis2Consumer extends DefaultConsumer { class KclKinesisConsumingTask implements Runnable { - private final Kinesis2Configuration configuration; + private final KinesisAsyncClient kinesisAsyncClient; + private final DynamoDbAsyncClient dynamoDbAsyncClient; + private final CloudWatchAsyncClient cloudWatchAsyncClient; + private final String streamName; - KclKinesisConsumingTask(Kinesis2Configuration configuration) { - this.configuration = configuration; + KclKinesisConsumingTask(String streamName, KinesisAsyncClient kinesisAsyncClient, + DynamoDbAsyncClient dynamoDbAsyncClient, CloudWatchAsyncClient cloudWatchAsyncClient) { + this.cloudWatchAsyncClient = cloudWatchAsyncClient; + this.dynamoDbAsyncClient = dynamoDbAsyncClient; + this.kinesisAsyncClient = kinesisAsyncClient; + this.streamName = streamName; } @Override public void run() { try { - DynamoDbAsyncClient dynamoClient - = DynamoDbAsyncClient.builder().region(Region.of(getEndpoint().getConfiguration().getRegion())).build(); - CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder() - .region(Region.of(getEndpoint().getConfiguration().getRegion())).build(); ConfigsBuilder configsBuilder = new ConfigsBuilder( - getEndpoint().getConfiguration().getStreamName(), getEndpoint().getConfiguration().getStreamName(), - getEndpoint().getAsyncClient(), dynamoClient, cloudWatchClient, UUID.randomUUID().toString(), + streamName, streamName, + kinesisAsyncClient, dynamoDbAsyncClient, cloudWatchAsyncClient, + "KclKinesisConsumingTask-" + UUID.randomUUID().toString(), new CamelKinesisRecordProcessorFactory()); - /** - * The Scheduler (also called Worker in earlier versions of the KCL) is the entry point to the KCL. This - * instance is configured with defaults provided by the ConfigsBuilder. - */ Scheduler scheduler = new Scheduler( configsBuilder.checkpointConfig(), configsBuilder.coordinatorConfig(), @@ -207,7 +269,7 @@ public class KclKinesis2Consumer extends DefaultConsumer { configsBuilder.metricsConfig(), configsBuilder.processorConfig(), configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig( - getEndpoint().getConfiguration().getStreamName(), getEndpoint().getAsyncClient()))); + getEndpoint().getConfiguration().getStreamName(), kinesisAsyncClient))); schedulerKcl = scheduler; Thread schedulerThread = new Thread(scheduler); diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java index c00d9771cc6..3e77e170619 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java +++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java @@ -22,6 +22,8 @@ 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.cloudwatch.CloudWatchAsyncClient; +import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; import software.amazon.awssdk.services.kinesis.KinesisClient; import software.amazon.awssdk.services.kinesis.model.ShardIteratorType; @@ -103,9 +105,17 @@ public class Kinesis2Configuration implements Cloneable { @UriParam(label = "consumer,advanced", description = "The interval in milliseconds to wait between shard polling", defaultValue = "10000") private long shardMonitorInterval = 10000; + + // KCL specific parameters @UriParam(label = "advanced", description = "If we want to a KCL Consumer set it to true") private boolean useKclConsumers; + @UriParam(label = "advanced", + description = "If we want to a KCL Consumer, we can pass an instance of DynamoDbAsyncClient") + private DynamoDbAsyncClient dynamoDbAsyncClient; + @UriParam(label = "advanced", + description = "If we want to a KCL Consumer, we can pass an instance of CloudWatchAsyncClient") + private CloudWatchAsyncClient cloudWatchAsyncClient; public KinesisClient getAmazonKinesisClient() { return amazonKinesisClient; @@ -307,6 +317,22 @@ public class Kinesis2Configuration implements Cloneable { this.useKclConsumers = useKclConsumers; } + public DynamoDbAsyncClient getDynamoDbAsyncClient() { + return dynamoDbAsyncClient; + } + + public void setDynamoDbAsyncClient(DynamoDbAsyncClient dynamoDbAsyncClient) { + this.dynamoDbAsyncClient = dynamoDbAsyncClient; + } + + public CloudWatchAsyncClient getCloudWatchAsyncClient() { + return cloudWatchAsyncClient; + } + + public void setCloudWatchAsyncClient(CloudWatchAsyncClient cloudWatchAsyncClient) { + this.cloudWatchAsyncClient = cloudWatchAsyncClient; + } + // ************************************************* // // ************************************************* diff --git a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KclKinesisConsumerIT.java b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KclKinesisConsumerIT.java index a04b5acb900..61816ed5759 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KclKinesisConsumerIT.java +++ b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KclKinesisConsumerIT.java @@ -18,9 +18,7 @@ package org.apache.camel.component.aws2.kinesis.integration; import java.util.concurrent.TimeUnit; -import org.apache.camel.EndpointInject; -import org.apache.camel.ExchangePattern; -import org.apache.camel.ProducerTemplate; +import org.apache.camel.*; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.aws2.kinesis.Kinesis2Constants; import org.apache.camel.component.mock.MockEndpoint; @@ -45,7 +43,7 @@ public class KclKinesisConsumerIT extends CamelTestSupport { @Override public void configure() { - from("direct:start").delay(50000) + from("direct:start").delay(10000) .to("aws2-kinesis://pippo?useDefaultCredentialsProvider=true®ion=eu-west-1").startupOrder(2); from("aws2-kinesis://pippo?useDefaultCredentialsProvider=true&useKclConsumers=true®ion=eu-west-1&asyncClient=true")