This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 0856951b23b CAMEL-20776: Kinesis Consumer uses a separate thread to
monitor shard changes (#14406)
0856951b23b is described below
commit 0856951b23b682e324b83b4731b48fa664bfe95a
Author: Fan Yang <fan...@hotmail.com>
AuthorDate: Mon Jun 10 12:56:22 2024 +0800
CAMEL-20776: Kinesis Consumer uses a separate thread to monitor shard
changes (#14406)
* Kinesis Consumer uses a separate thread to monitor shard changes
* Use millisecond and ExecutorServiceManager
* Commit generated files
* Commit generated files
* Adjust code format
* fix tests
* format
---
.../camel/catalog/components/aws2-kinesis.json | 92 +++++++++++-----------
.../aws2/kinesis/Kinesis2ComponentConfigurer.java | 6 ++
.../aws2/kinesis/Kinesis2EndpointConfigurer.java | 6 ++
.../aws2/kinesis/Kinesis2EndpointUriFactory.java | 3 +-
.../camel/component/aws2/kinesis/aws2-kinesis.json | 92 +++++++++++-----------
.../aws2/kinesis/Kinesis2Configuration.java | 11 +++
.../component/aws2/kinesis/Kinesis2Consumer.java | 92 ++++++++++++++++------
.../KinesisConsumerClosedShardWithFailTest.java | 36 +++++----
.../KinesisConsumerClosedShardWithSilentTest.java | 33 ++++----
.../kinesis/integration/KinesisConsumerIT.java | 4 +-
.../KinesisConsumerResumeAfterRestartIT.java | 4 +-
.../integration/KinesisConsumerResumeIT.java | 4 +-
.../dsl/Aws2KinesisComponentBuilderFactory.java | 18 +++++
.../dsl/Kinesis2EndpointBuilderFactory.java | 30 +++++++
.../camel/kotlin/components/Aws2KinesisUriDsl.kt | 14 ++++
15 files changed, 289 insertions(+), 156 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
index 54b40aa1820..8f1a0065ed7 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws2-kinesis.json
@@ -34,23 +34,24 @@
"sequenceNumber": { "index": 8, "kind": "property", "displayName":
"Sequence Number", "group": "consumer", "label": "consumer", "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": "The sequence number to
start polling from. Required if iteratorType is set to AFTER_SEQUENC [...]
"shardClosed": { "index": 9, "kind": "property", "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, "autowired":
false, "secret": false, "defaultValue": "ignore", "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configur [...]
"shardId": { "index": 10, "kind": "property", "displayName": "Shard Id",
"group": "consumer", "label": "consumer", "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": "Defines which shardId in
the Kinesis stream to get records from" },
- "lazyStartProducer": { "index": 11, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "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 producer may otherwise fai [...]
- "amazonKinesisClient": { "index": 12, "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": 13, "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": 14, "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 [...]
- "healthCheckConsumerEnabled": { "index": 15, "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": 16, "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": 17, "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": 18, "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": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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 [...]
+ "shardMonitorInterval": { "index": 11, "kind": "property", "displayName":
"Shard Monitor Interval", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "integer", "javaType": "long",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
10000, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "The interval in
milliseconds to wait between [...]
+ "lazyStartProducer": { "index": 12, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "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 producer may otherwise fai [...]
+ "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 [...]
+ "healthCheckConsumerEnabled": { "index": 16, "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": 17, "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": 18, "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": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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": 28, "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 [...]
@@ -75,33 +76,34 @@
"exceptionHandler": { "index": 12, "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, "autowired": 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 de [...]
"exchangePattern": { "index": 13, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
"pollStrategy": { "index": 14, "kind": "parameter", "displayName": "Poll
Strategy", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false,
"autowired": 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 po
[...]
- "lazyStartProducer": { "index": 15, "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": 16, "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": 17, "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" },
- "proxyHost": { "index": 18, "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": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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 [...]
+ "shardMonitorInterval": { "index": 15, "kind": "parameter", "displayName":
"Shard Monitor Interval", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "integer", "javaType": "long",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
10000, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "The interval in
milliseconds to wait between [...]
+ "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" },
+ "proxyHost": { "index": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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": 43, "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/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 53f6ff4fac6..7232eeee2de 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
@@ -74,6 +74,8 @@ public class Kinesis2ComponentConfigurer extends
PropertyConfigurerSupport imple
case "shardClosed":
getOrCreateConfiguration(target).setShardClosed(property(camelContext,
org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum.class,
value)); return true;
case "shardid":
case "shardId":
getOrCreateConfiguration(target).setShardId(property(camelContext,
java.lang.String.class, value)); return true;
+ case "shardmonitorinterval":
+ case "shardMonitorInterval":
getOrCreateConfiguration(target).setShardMonitorInterval(property(camelContext,
long.class, value)); return true;
case "trustallcertificates":
case "trustAllCertificates":
getOrCreateConfiguration(target).setTrustAllCertificates(property(camelContext,
boolean.class, value)); return true;
case "uriendpointoverride":
@@ -140,6 +142,8 @@ public class Kinesis2ComponentConfigurer extends
PropertyConfigurerSupport imple
case "shardClosed": return
org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum.class;
case "shardid":
case "shardId": return java.lang.String.class;
+ case "shardmonitorinterval":
+ case "shardMonitorInterval": return long.class;
case "trustallcertificates":
case "trustAllCertificates": return boolean.class;
case "uriendpointoverride":
@@ -202,6 +206,8 @@ public class Kinesis2ComponentConfigurer extends
PropertyConfigurerSupport imple
case "shardClosed": return
getOrCreateConfiguration(target).getShardClosed();
case "shardid":
case "shardId": return getOrCreateConfiguration(target).getShardId();
+ case "shardmonitorinterval":
+ case "shardMonitorInterval": return
getOrCreateConfiguration(target).getShardMonitorInterval();
case "trustallcertificates":
case "trustAllCertificates": return
getOrCreateConfiguration(target).isTrustAllCertificates();
case "uriendpointoverride":
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 7712291641f..f883fa53bfe 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
@@ -87,6 +87,8 @@ public class Kinesis2EndpointConfigurer extends
PropertyConfigurerSupport implem
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 "shardmonitorinterval":
+ case "shardMonitorInterval":
target.getConfiguration().setShardMonitorInterval(property(camelContext,
long.class, value)); return true;
case "startscheduler":
case "startScheduler": target.setStartScheduler(property(camelContext,
boolean.class, value)); return true;
case "timeunit":
@@ -179,6 +181,8 @@ public class Kinesis2EndpointConfigurer extends
PropertyConfigurerSupport implem
case "shardClosed": return
org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum.class;
case "shardid":
case "shardId": return java.lang.String.class;
+ case "shardmonitorinterval":
+ case "shardMonitorInterval": return long.class;
case "startscheduler":
case "startScheduler": return boolean.class;
case "timeunit":
@@ -267,6 +271,8 @@ public class Kinesis2EndpointConfigurer extends
PropertyConfigurerSupport implem
case "shardClosed": return target.getConfiguration().getShardClosed();
case "shardid":
case "shardId": return target.getConfiguration().getShardId();
+ case "shardmonitorinterval":
+ case "shardMonitorInterval": return
target.getConfiguration().getShardMonitorInterval();
case "startscheduler":
case "startScheduler": return target.isStartScheduler();
case "timeunit":
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 4c53d9c3d06..20d258a740d 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<>(43);
+ Set<String> props = new HashSet<>(44);
props.add("accessKey");
props.add("amazonKinesisClient");
props.add("asyncClient");
@@ -58,6 +58,7 @@ public class Kinesis2EndpointUriFactory extends
org.apache.camel.support.compone
props.add("sessionToken");
props.add("shardClosed");
props.add("shardId");
+ props.add("shardMonitorInterval");
props.add("startScheduler");
props.add("streamName");
props.add("timeUnit");
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 54b40aa1820..8f1a0065ed7 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
@@ -34,23 +34,24 @@
"sequenceNumber": { "index": 8, "kind": "property", "displayName":
"Sequence Number", "group": "consumer", "label": "consumer", "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": "The sequence number to
start polling from. Required if iteratorType is set to AFTER_SEQUENC [...]
"shardClosed": { "index": 9, "kind": "property", "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, "autowired":
false, "secret": false, "defaultValue": "ignore", "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configur [...]
"shardId": { "index": 10, "kind": "property", "displayName": "Shard Id",
"group": "consumer", "label": "consumer", "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": "Defines which shardId in
the Kinesis stream to get records from" },
- "lazyStartProducer": { "index": 11, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "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 producer may otherwise fai [...]
- "amazonKinesisClient": { "index": 12, "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": 13, "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": 14, "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 [...]
- "healthCheckConsumerEnabled": { "index": 15, "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": 16, "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": 17, "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": 18, "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": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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 [...]
+ "shardMonitorInterval": { "index": 11, "kind": "property", "displayName":
"Shard Monitor Interval", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "integer", "javaType": "long",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
10000, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "The interval in
milliseconds to wait between [...]
+ "lazyStartProducer": { "index": 12, "kind": "property", "displayName":
"Lazy Start Producer", "group": "producer", "label": "producer", "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 producer may otherwise fai [...]
+ "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 [...]
+ "healthCheckConsumerEnabled": { "index": 16, "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": 17, "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": 18, "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": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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": 28, "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 [...]
@@ -75,33 +76,34 @@
"exceptionHandler": { "index": 12, "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, "autowired": 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 de [...]
"exchangePattern": { "index": 13, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
"pollStrategy": { "index": 14, "kind": "parameter", "displayName": "Poll
Strategy", "group": "consumer (advanced)", "label": "consumer,advanced",
"required": false, "type": "object", "javaType":
"org.apache.camel.spi.PollingConsumerPollStrategy", "deprecated": false,
"autowired": 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 po
[...]
- "lazyStartProducer": { "index": 15, "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": 16, "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": 17, "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" },
- "proxyHost": { "index": 18, "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": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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 [...]
+ "shardMonitorInterval": { "index": 15, "kind": "parameter", "displayName":
"Shard Monitor Interval", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "integer", "javaType": "long",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
10000, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "The interval in
milliseconds to wait between [...]
+ "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" },
+ "proxyHost": { "index": 19, "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": 20, "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": 21, "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": 22, "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": 23, "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": 24, "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": 25, "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": 26, "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": 27, "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": 28, "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": 29, "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": 30, "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": 31, "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": 32, "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": 33, "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": 34, "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": 35, "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": 36, "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": 37, "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": 38, "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": 39, "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": 40, "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": 41, "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": 42, "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": 43, "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/Kinesis2Configuration.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java
index b086bd567f6..f3e1f88da65 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
@@ -100,6 +100,9 @@ public class Kinesis2Configuration implements Cloneable {
@UriParam(label = "security",
description = "If using a profile credentials provider this
parameter will set the profile name.")
private String profileCredentialsName;
+ @UriParam(label = "consumer,advanced", description = "The interval in
milliseconds to wait between shard polling",
+ defaultValue = "10000")
+ private long shardMonitorInterval = 10000;
public KinesisClient getAmazonKinesisClient() {
return amazonKinesisClient;
@@ -285,6 +288,14 @@ public class Kinesis2Configuration implements Cloneable {
this.asyncClient = asyncClient;
}
+ public long getShardMonitorInterval() {
+ return shardMonitorInterval;
+ }
+
+ public void setShardMonitorInterval(long shardMonitorInterval) {
+ this.shardMonitorInterval = shardMonitorInterval;
+ }
+
// *************************************************
//
// *************************************************
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
index fd9b1ee098d..9c5c86eb0d6 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java
@@ -17,10 +17,13 @@
package org.apache.camel.component.aws2.kinesis;
import java.util.ArrayDeque;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.AsyncCallback;
@@ -53,6 +56,10 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
private Map<String, String> currentShardIterators = new
java.util.HashMap<>();
+ private List<Shard> currentShardList = new ArrayList<>();
+ private static final String SHARD_MONITOR_EXECUTOR_NAME =
"Kinesis_shard_monitor";
+ private ScheduledExecutorService shardMonitorExecutor;
+
public Kinesis2Consumer(Kinesis2Endpoint endpoint,
Processor processor) {
super(endpoint, processor);
@@ -109,11 +116,9 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
fetchAndPrepareRecordsForCamel(shard, connection,
processedExchangeCount);
} else {
- getShardList(connection)
+ getCurrentShardList()
.parallelStream()
- .forEach(shard -> {
- fetchAndPrepareRecordsForCamel(shard, connection,
processedExchangeCount);
- });
+ .forEach(shard -> fetchAndPrepareRecordsForCamel(shard,
connection, processedExchangeCount));
}
// okay we have some response from aws so lets mark the consumer as
ready
@@ -336,44 +341,79 @@ public class Kinesis2Consumer extends
ScheduledBatchPollingConsumer implements R
super.doStart();
ObjectHelper.notNull(connection, "connection", this);
+ this.shardMonitorExecutor =
getEndpoint().getCamelContext().getExecutorServiceManager()
+ .newSingleThreadScheduledExecutor(this,
SHARD_MONITOR_EXECUTOR_NAME);
+ this.shardMonitorExecutor.scheduleAtFixedRate(new ShardMonitor(),
+ 0, getConfiguration().getShardMonitorInterval(),
TimeUnit.MILLISECONDS);
if (resumeStrategy != null) {
resumeStrategy.loadCache();
}
}
+ @Override
+ protected void doStop() throws Exception {
+ if (this.shardMonitorExecutor != null) {
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(this.shardMonitorExecutor);
+ this.shardMonitorExecutor = null;
+ }
+
+ super.doStop();
+ }
+
protected Kinesis2Configuration getConfiguration() {
return getEndpoint().getConfiguration();
}
- private List<Shard> getShardList(final KinesisConnection
kinesisConnection) {
- var request = ListShardsRequest
- .builder()
- .streamName(getEndpoint().getConfiguration().getStreamName())
- .build();
+ protected synchronized List<Shard> getCurrentShardList() {
+ return this.currentShardList;
+ }
- List<Shard> shardList;
- if (getEndpoint().getConfiguration().isAsyncClient()) {
+ private synchronized void setCurrentShardList(List<Shard> latestShardList)
{
+ this.currentShardList = latestShardList;
+ }
+
+ private class ShardMonitor implements Runnable {
+ @Override
+ public void run() {
try {
+ List<Shard> latestShardList = getShardList(connection);
+ if (latestShardList != null) {
+ setCurrentShardList(latestShardList);
+ }
+ } catch (Exception e) {
+ LOG.warn("Exception getting latest shard list", e);
+ }
+ }
+
+ private List<Shard> getShardList(final KinesisConnection
kinesisConnection) {
+ var request = ListShardsRequest
+ .builder()
+
.streamName(getEndpoint().getConfiguration().getStreamName())
+ .build();
+
+ List<Shard> shardList;
+ if (getEndpoint().getConfiguration().isAsyncClient()) {
+ try {
+ shardList = kinesisConnection
+ .getAsyncClient(getEndpoint())
+ .listShards(request)
+ .get()
+ .shards();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
shardList = kinesisConnection
- .getAsyncClient(getEndpoint())
+ .getClient(getEndpoint())
.listShards(request)
- .get()
.shards();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new RuntimeException(e);
- } catch (ExecutionException e) {
- throw new RuntimeException(e);
}
- } else {
- shardList = kinesisConnection
- .getClient(getEndpoint())
- .listShards(request)
- .shards();
- }
- return shardList;
+ return shardList;
+ }
}
-
}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithFailTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithFailTest.java
index 9f5156ca5a7..4f6f1da4312 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithFailTest.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithFailTest.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.aws2.kinesis;
import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.CamelContext;
@@ -43,9 +44,10 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
@ExtendWith(MockitoExtension.class)
public class KinesisConsumerClosedShardWithFailTest {
@@ -62,20 +64,6 @@ public class KinesisConsumerClosedShardWithFailTest {
@BeforeEach
public void setup() {
- component.start();
-
- Kinesis2Configuration configuration = new Kinesis2Configuration();
- configuration.setAmazonKinesisClient(kinesisClient);
- configuration.setIteratorType(ShardIteratorType.LATEST);
- configuration.setShardClosed(Kinesis2ShardClosedStrategyEnum.fail);
- configuration.setStreamName("streamName");
-
- Kinesis2Endpoint endpoint = new Kinesis2Endpoint("aws2-kinesis:foo",
configuration, component);
- endpoint.start();
- underTest = new Kinesis2Consumer(endpoint, processor);
- underTest.setConnection(component.getConnection());
- underTest.start();
-
SequenceNumberRange range =
SequenceNumberRange.builder().endingSequenceNumber("20").build();
Shard shard =
Shard.builder().shardId("shardId").sequenceNumberRange(range).build();
ArrayList<Shard> shardList = new ArrayList<>();
@@ -90,6 +78,22 @@ public class KinesisConsumerClosedShardWithFailTest {
when(kinesisClient
.listShards(any(ListShardsRequest.class)))
.thenReturn(ListShardsResponse.builder().shards(shardList).build());
+
+ component.start();
+
+ Kinesis2Configuration configuration = new Kinesis2Configuration();
+ configuration.setAmazonKinesisClient(kinesisClient);
+ configuration.setIteratorType(ShardIteratorType.LATEST);
+ configuration.setShardClosed(Kinesis2ShardClosedStrategyEnum.fail);
+ configuration.setStreamName("streamName");
+
+ Kinesis2Endpoint endpoint = new Kinesis2Endpoint("aws2-kinesis:foo",
configuration, component);
+ endpoint.start();
+ underTest = new Kinesis2Consumer(endpoint, processor);
+ underTest.setConnection(component.getConnection());
+ underTest.start();
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> !(underTest.getCurrentShardList().isEmpty()));
}
@Test
@@ -113,7 +117,7 @@ public class KinesisConsumerClosedShardWithFailTest {
assertThat(getShardIteratorReqCap.getValue().shardId(), is("shardId"));
assertThat(getShardIteratorReqCap.getValue().shardIteratorType(),
is(ShardIteratorType.LATEST));
- verify(kinesisClient, times(2)).listShards(getListShardsCap.capture());
+ verify(kinesisClient,
atLeastOnce()).listShards(getListShardsCap.capture());
assertThat(getListShardsCap.getValue().streamName(), is("streamName"));
}
}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
index 034af13ed1b..643bb199ad9 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerClosedShardWithSilentTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.aws2.kinesis;
import java.nio.charset.Charset;
import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.CamelContext;
@@ -52,6 +53,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
@@ -68,20 +70,6 @@ public class KinesisConsumerClosedShardWithSilentTest {
@BeforeEach
public void setup() {
- component.start();
-
- Kinesis2Configuration configuration = new Kinesis2Configuration();
- configuration.setAmazonKinesisClient(kinesisClient);
- configuration.setIteratorType(ShardIteratorType.LATEST);
- configuration.setShardClosed(Kinesis2ShardClosedStrategyEnum.silent);
- configuration.setStreamName("streamName");
-
- Kinesis2Endpoint endpoint = new Kinesis2Endpoint("aws2-kinesis:foo",
configuration, component);
- endpoint.start();
- underTest = new Kinesis2Consumer(endpoint, processor);
- underTest.setConnection(component.getConnection());
- underTest.start();
-
SequenceNumberRange range =
SequenceNumberRange.builder().endingSequenceNumber("20").build();
Shard shard =
Shard.builder().shardId("shardId").sequenceNumberRange(range).build();
ArrayList<Shard> shardList = new ArrayList<>();
@@ -98,7 +86,6 @@ public class KinesisConsumerClosedShardWithSilentTest {
.data(SdkBytes.fromString("Hello",
Charset.defaultCharset()))
.build())
.build());
-
when(kinesisClient
.getShardIterator(any(GetShardIteratorRequest.class)))
.thenReturn(GetShardIteratorResponse.builder().shardIterator("shardIterator").build());
@@ -106,9 +93,21 @@ public class KinesisConsumerClosedShardWithSilentTest {
.listShards(any(ListShardsRequest.class)))
.thenReturn(ListShardsResponse.builder().shards(shardList).build());
- context.start();
- underTest.start();
+ component.start();
+
+ Kinesis2Configuration configuration = new Kinesis2Configuration();
+ configuration.setAmazonKinesisClient(kinesisClient);
+ configuration.setIteratorType(ShardIteratorType.LATEST);
+ configuration.setShardClosed(Kinesis2ShardClosedStrategyEnum.silent);
+ configuration.setStreamName("streamName");
+ Kinesis2Endpoint endpoint = new Kinesis2Endpoint("aws2-kinesis:foo",
configuration, component);
+ endpoint.start();
+ underTest = new Kinesis2Consumer(endpoint, processor);
+ underTest.setConnection(component.getConnection());
+ underTest.start();
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> !(underTest.getCurrentShardList().isEmpty()));
}
@Test
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerIT.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerIT.java
index 3ce4a00f8bf..a645957cf21 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerIT.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerIT.java
@@ -17,8 +17,8 @@
package org.apache.camel.component.aws2.kinesis.integration;
-import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import org.apache.camel.EndpointInject;
@@ -76,7 +76,7 @@ public class KinesisConsumerIT extends CamelTestSupport {
private KinesisClient client;
private String streamName = AWSCommon.KINESIS_STREAM_BASE_NAME + "-cons-"
+ TestUtils.randomWithRange(0, 100);
private final int messageCount = 20;
- private List<KinesisData> receivedMessages = new ArrayList<>();
+ private List<KinesisData> receivedMessages = new CopyOnWriteArrayList<>();
@Override
protected RouteBuilder createRouteBuilder() {
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeAfterRestartIT.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeAfterRestartIT.java
index 662383c4858..41a9b10efbf 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeAfterRestartIT.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeAfterRestartIT.java
@@ -16,8 +16,8 @@
*/
package org.apache.camel.component.aws2.kinesis.integration;
-import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.camel.builder.RouteBuilder;
import
org.apache.camel.component.aws2.kinesis.consumer.KinesisConsumerOffsetProcessor;
@@ -47,7 +47,7 @@ public class KinesisConsumerResumeAfterRestartIT extends
CamelTestSupport {
String streamName = "my-stream";
- List<String> receivedMessages = new ArrayList<>();
+ List<String> receivedMessages = new CopyOnWriteArrayList<>();
ResumeCache resumeCache = TransientResumeStrategy.createSimpleCache();
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeIT.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeIT.java
index 4d8cfbc2483..1868e7ad251 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeIT.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/integration/KinesisConsumerResumeIT.java
@@ -17,8 +17,8 @@
package org.apache.camel.component.aws2.kinesis.integration;
-import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import org.apache.camel.EndpointInject;
@@ -124,7 +124,7 @@ public class KinesisConsumerResumeIT extends
CamelTestSupport {
private String streamName = AWSCommon.KINESIS_STREAM_BASE_NAME + "-cons-"
+ TestUtils.randomWithRange(0, 100);
private final int messageCount = 20;
private final int expectedCount = messageCount / 2;
- private List<KinesisData> receivedMessages = new ArrayList<>();
+ private List<KinesisData> receivedMessages = new CopyOnWriteArrayList<>();
private List<PutRecordsResponse> previousRecords;
private TestKinesisResumeAdapter adapter = new
TestKinesisResumeAdapter(expectedCount);
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisComponentBuilderFactory.java
index f6e950e56aa..180c7b5d0c6 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/Aws2KinesisComponentBuilderFactory.java
@@ -253,6 +253,23 @@ public interface Aws2KinesisComponentBuilderFactory {
}
+ /**
+ * The interval in milliseconds to wait between shard polling.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Default: 10000
+ * Group: consumer (advanced)
+ *
+ * @param shardMonitorInterval the value to set
+ * @return the dsl builder
+ */
+ default Aws2KinesisComponentBuilder shardMonitorInterval(long
shardMonitorInterval) {
+ doSetProperty("shardMonitorInterval", shardMonitorInterval);
+ 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
@@ -585,6 +602,7 @@ public interface Aws2KinesisComponentBuilderFactory {
case "sequenceNumber":
getOrCreateConfiguration((Kinesis2Component)
component).setSequenceNumber((java.lang.String) value); return true;
case "shardClosed": getOrCreateConfiguration((Kinesis2Component)
component).setShardClosed((org.apache.camel.component.aws2.kinesis.Kinesis2ShardClosedStrategyEnum)
value); return true;
case "shardId": getOrCreateConfiguration((Kinesis2Component)
component).setShardId((java.lang.String) value); return true;
+ case "shardMonitorInterval":
getOrCreateConfiguration((Kinesis2Component)
component).setShardMonitorInterval((long) value); return true;
case "lazyStartProducer": ((Kinesis2Component)
component).setLazyStartProducer((boolean) value); return true;
case "amazonKinesisClient":
getOrCreateConfiguration((Kinesis2Component)
component).setAmazonKinesisClient((software.amazon.awssdk.services.kinesis.KinesisClient)
value); return true;
case "asyncClient": getOrCreateConfiguration((Kinesis2Component)
component).setAsyncClient((boolean) value); return true;
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java
index 6339e4c8d44..5188a698bd9 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/Kinesis2EndpointBuilderFactory.java
@@ -1186,6 +1186,36 @@ public interface Kinesis2EndpointBuilderFactory {
doSetProperty("pollStrategy", pollStrategy);
return this;
}
+ /**
+ * The interval in milliseconds to wait between shard polling.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Default: 10000
+ * Group: consumer (advanced)
+ *
+ * @param shardMonitorInterval the value to set
+ * @return the dsl builder
+ */
+ default AdvancedKinesis2EndpointConsumerBuilder
shardMonitorInterval(long shardMonitorInterval) {
+ doSetProperty("shardMonitorInterval", shardMonitorInterval);
+ return this;
+ }
+ /**
+ * The interval in milliseconds to wait between shard polling.
+ *
+ * The option will be converted to a <code>long</code> type.
+ *
+ * Default: 10000
+ * Group: consumer (advanced)
+ *
+ * @param shardMonitorInterval the value to set
+ * @return the dsl builder
+ */
+ default AdvancedKinesis2EndpointConsumerBuilder
shardMonitorInterval(String shardMonitorInterval) {
+ doSetProperty("shardMonitorInterval", shardMonitorInterval);
+ return this;
+ }
/**
* Amazon Kinesis client to use for all requests for this endpoint.
*
diff --git
a/dsl/camel-kotlin-api/src/generated/kotlin/org/apache/camel/kotlin/components/Aws2KinesisUriDsl.kt
b/dsl/camel-kotlin-api/src/generated/kotlin/org/apache/camel/kotlin/components/Aws2KinesisUriDsl.kt
index a1cb996ebba..80d48c03684 100644
---
a/dsl/camel-kotlin-api/src/generated/kotlin/org/apache/camel/kotlin/components/Aws2KinesisUriDsl.kt
+++
b/dsl/camel-kotlin-api/src/generated/kotlin/org/apache/camel/kotlin/components/Aws2KinesisUriDsl.kt
@@ -215,6 +215,20 @@ public class Aws2KinesisUriDsl(
it.property("pollStrategy", pollStrategy)
}
+ /**
+ * The interval in milliseconds to wait between shard polling
+ */
+ public fun shardMonitorInterval(shardMonitorInterval: String) {
+ it.property("shardMonitorInterval", shardMonitorInterval)
+ }
+
+ /**
+ * The interval in milliseconds to wait between shard polling
+ */
+ public fun shardMonitorInterval(shardMonitorInterval: Int) {
+ it.property("shardMonitorInterval", shardMonitorInterval.toString())
+ }
+
/**
* 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
