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 14d840974f9c CAMEL-26650: camel-aws2-kinesis - the consumer fetches up
to 1000 records per shard in each poll by default (#29272)
14d840974f9c is described below
commit 14d840974f9c55f5c716ed0fe3d6f87a7d6659c4
Author: Federico Mariani <[email protected]>
AuthorDate: Fri Nov 6 08:21:17 2026 +0100
CAMEL-26650: camel-aws2-kinesis - the consumer fetches up to 1000 records
per shard in each poll by default (#29272)
The Kinesis consumer now fetches up to 1000 records per shard in each poll
by default. Docs and upgrade guide explain the effect on graceful shutdown.
AI-generated with Claude Code on behalf of Federico Mariani.
Co-Authored-By: Claude <[email protected]>
---
.../camel/catalog/components/aws2-kinesis.json | 4 +-
.../camel/catalog/docs/aws2-kinesis-component.adoc | 37 ++++++
.../camel/component/aws2/kinesis/aws2-kinesis.json | 4 +-
.../src/main/docs/aws2-kinesis-component.adoc | 37 ++++++
.../aws2/kinesis/Kinesis2Configuration.java | 8 +-
.../KinesisConsumerGetRecordsLimitTest.java | 143 +++++++++++++++++++++
.../aws2/kinesis/KinesisEndpointTest.java | 2 +-
.../integration/KinesisConsumerResumeIT.java | 5 +
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 18 +++
.../dsl/Aws2KinesisComponentBuilderFactory.java | 6 +-
.../dsl/Kinesis2EndpointBuilderFactory.java | 12 +-
11 files changed, 262 insertions(+), 14 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 7bc42d67463f..27f0d5a3bd4b 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
@@ -31,7 +31,7 @@
"uriEndpointOverride": { "index": 4, "kind": "property", "displayName":
"Uri Endpoint Override", "group": "common", "label": "common", "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": "Set the overriding uri
endpoint. This option needs to be used in combination with ov [...]
"bridgeErrorHandler": { "index": 5, "kind": "property", "displayName":
"Bridge Error Handler", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Allows for bridging the consumer to the Camel routing Error Handler, which
mean any exceptions (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
"iteratorType": { "index": 6, "kind": "property", "displayName": "Iterator
Type", "group": "consumer", "label": "consumer", "required": false, "type":
"enum", "javaType":
"software.amazon.awssdk.services.kinesis.model.ShardIteratorType", "enum": [
"AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "TRIM_HORIZON", "LATEST",
"AT_TIMESTAMP", "null" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "TRIM_HORIZON", "configurationClass":
"org.apache.camel.component.a [...]
- "maxResultsPerRequest": { "index": 7, "kind": "property", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResultsPerRequest": { "index": 7, "kind": "property", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000,
"configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records fetched from each shard in each poll: the [...]
"messageTimestamp": { "index": 8, "kind": "property", "displayName":
"Message Timestamp", "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 message timestamp to
start polling from. Required if iteratorType is set to AT_TIME [...]
"sequenceNumber": { "index": 9, "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": 10, "kind": "property", "displayName": "Shard
Closed", "group": "consumer", "label": "consumer", "required": false, "type":
"enum", "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": "configura [...]
@@ -84,7 +84,7 @@
"region": { "index": 3, "kind": "parameter", "displayName": "Region",
"group": "common", "label": "", "required": false, "type": "enum", "javaType":
"java.lang.String", "enum": [ "ap-south-2", "ap-south-1", "eu-south-1",
"eu-south-2", "us-gov-east-1", "me-central-1", "il-central-1", "ca-central-1",
"eu-central-1", "us-iso-west-1", "eu-central-2", "eu-isoe-west-1", "us-west-1",
"us-west-2", "af-south-1", "eu-north-1", "eu-west-3", "eu-west-2", "eu-west-1",
"ap-northeast-3", "ap-northe [...]
"uriEndpointOverride": { "index": 4, "kind": "parameter", "displayName":
"Uri Endpoint Override", "group": "common", "label": "common", "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": "Set the overriding uri
endpoint. This option needs to be used in combination with o [...]
"iteratorType": { "index": 5, "kind": "parameter", "displayName":
"Iterator Type", "group": "consumer", "label": "consumer", "required": false,
"type": "enum", "javaType":
"software.amazon.awssdk.services.kinesis.model.ShardIteratorType", "enum": [
"AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "TRIM_HORIZON", "LATEST",
"AT_TIMESTAMP", "null" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "TRIM_HORIZON", "configurationClass":
"org.apache.camel.component. [...]
- "maxResultsPerRequest": { "index": 6, "kind": "parameter", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResultsPerRequest": { "index": 6, "kind": "parameter", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000,
"configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records fetched from each shard in each poll: th [...]
"messageTimestamp": { "index": 7, "kind": "parameter", "displayName":
"Message Timestamp", "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 message timestamp to
start polling from. Required if iteratorType is set to AT_TIM [...]
"sendEmptyMessageWhenIdle": { "index": 8, "kind": "parameter",
"displayName": "Send Empty Message When Idle", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "description": "If the polling consumer did not poll any files, you can
enable this option to send an empty message (no body) instead." },
"sequenceNumber": { "index": 9, "kind": "parameter", "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_SEQUEN [...]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
index 2c81d405ee68..00c03c331a4b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws2-kinesis-component.adoc
@@ -66,6 +66,43 @@ all available shards (multiple shards consumption) of Amazon
Kinesis, therefore,
property in the DSL configuration empty, then it'll consume all available
shards
otherwise only the specified shard corresponding to the shardId will be
consumed.
+=== Consumer throughput and the Kinesis read quotas
+
+In each poll the consumer sends one `GetRecords` request to each shard it
reads. The request asks for up to
+`maxResultsPerRequest` records (1000 by default; Kinesis accepts 1 to 10,000),
so a poll routes up to
+`maxResultsPerRequest` records per shard.
+
+Kinesis limits the reads of each shard:
+
+* 5 `GetRecords` requests per second;
+* 2 MiB of data per second;
+* at most 10,000 records or 10 MiB in one response.
+
+A request over these quotas fails with a
`ProvisionedThroughputExceededException`. After a response of 10 MiB, the
+requests of the next 5 seconds to that shard fail too.
+
+The scheduler options decide how often a shard is read:
+
+* With `greedy=false` (the default), the next poll starts `delay` milliseconds
(500 by default) after the previous one
+ ended. A shard then gets at most about 2 requests per second, so up to about
twice `maxResultsPerRequest` records
+ per second.
+* With `greedy=true`, the next poll starts at once after a poll that routed
records, and `delay` applies only after a
+ poll that routed none. With a backlog and a fast route, the consumer can
then send more than 5 requests per second to
+ a shard.
+
+To keep up with a shard, choose `maxResultsPerRequest` and `delay` so that the
records of a poll times the polls per
+second stay within 2 MiB per second. For example, 1000 records of 1 KiB every
500 ms is about 2 MiB per second. For
+larger records, lower `maxResultsPerRequest` or raise `delay`. While the
records of a response are routed, the
+consumer holds that response in memory, one for each shard of the poll.
+
+A xref:manual::graceful-shutdown.adoc[graceful shutdown] waits until the
consumer has routed all the records of the
+poll that is running. Choose `maxResultsPerRequest` so that the route handles
the records of a poll within the shutdown
+timeout (45 seconds by default). Otherwise the shutdown is forced, and the
records of the poll that are not routed yet
+fail with a `RouteStoppingException`.
+
+These options apply to the polling consumer only. The KCL consumer
(`useKclConsumers=true`) fetches the records with
+the settings of the Kinesis Client Library and does not use
`maxResultsPerRequest`, `delay` or `greedy`.
+
=== Custom Resume Action
The consumer supports a custom `KinesisResumeAction` for controlling where
each shard starts reading
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 7bc42d67463f..27f0d5a3bd4b 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
@@ -31,7 +31,7 @@
"uriEndpointOverride": { "index": 4, "kind": "property", "displayName":
"Uri Endpoint Override", "group": "common", "label": "common", "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": "Set the overriding uri
endpoint. This option needs to be used in combination with ov [...]
"bridgeErrorHandler": { "index": 5, "kind": "property", "displayName":
"Bridge Error Handler", "group": "consumer", "label": "consumer", "required":
false, "type": "boolean", "javaType": "boolean", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": false, "description":
"Allows for bridging the consumer to the Camel routing Error Handler, which
mean any exceptions (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
"iteratorType": { "index": 6, "kind": "property", "displayName": "Iterator
Type", "group": "consumer", "label": "consumer", "required": false, "type":
"enum", "javaType":
"software.amazon.awssdk.services.kinesis.model.ShardIteratorType", "enum": [
"AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "TRIM_HORIZON", "LATEST",
"AT_TIMESTAMP", "null" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "TRIM_HORIZON", "configurationClass":
"org.apache.camel.component.a [...]
- "maxResultsPerRequest": { "index": 7, "kind": "property", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResultsPerRequest": { "index": 7, "kind": "property", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000,
"configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records fetched from each shard in each poll: the [...]
"messageTimestamp": { "index": 8, "kind": "property", "displayName":
"Message Timestamp", "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 message timestamp to
start polling from. Required if iteratorType is set to AT_TIME [...]
"sequenceNumber": { "index": 9, "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": 10, "kind": "property", "displayName": "Shard
Closed", "group": "consumer", "label": "consumer", "required": false, "type":
"enum", "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": "configura [...]
@@ -84,7 +84,7 @@
"region": { "index": 3, "kind": "parameter", "displayName": "Region",
"group": "common", "label": "", "required": false, "type": "enum", "javaType":
"java.lang.String", "enum": [ "ap-south-2", "ap-south-1", "eu-south-1",
"eu-south-2", "us-gov-east-1", "me-central-1", "il-central-1", "ca-central-1",
"eu-central-1", "us-iso-west-1", "eu-central-2", "eu-isoe-west-1", "us-west-1",
"us-west-2", "af-south-1", "eu-north-1", "eu-west-3", "eu-west-2", "eu-west-1",
"ap-northeast-3", "ap-northe [...]
"uriEndpointOverride": { "index": 4, "kind": "parameter", "displayName":
"Uri Endpoint Override", "group": "common", "label": "common", "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": "Set the overriding uri
endpoint. This option needs to be used in combination with o [...]
"iteratorType": { "index": 5, "kind": "parameter", "displayName":
"Iterator Type", "group": "consumer", "label": "consumer", "required": false,
"type": "enum", "javaType":
"software.amazon.awssdk.services.kinesis.model.ShardIteratorType", "enum": [
"AT_SEQUENCE_NUMBER", "AFTER_SEQUENCE_NUMBER", "TRIM_HORIZON", "LATEST",
"AT_TIMESTAMP", "null" ], "deprecated": false, "autowired": false, "secret":
false, "defaultValue": "TRIM_HORIZON", "configurationClass":
"org.apache.camel.component. [...]
- "maxResultsPerRequest": { "index": 6, "kind": "parameter", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1, "configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records that will be fetched in each poll" },
+ "maxResultsPerRequest": { "index": 6, "kind": "parameter", "displayName":
"Max Results Per Request", "group": "consumer", "label": "consumer",
"required": false, "type": "integer", "javaType": "int", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": 1000,
"configurationClass":
"org.apache.camel.component.aws2.kinesis.Kinesis2Configuration",
"configurationField": "configuration", "description": "Maximum number of
records fetched from each shard in each poll: th [...]
"messageTimestamp": { "index": 7, "kind": "parameter", "displayName":
"Message Timestamp", "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 message timestamp to
start polling from. Required if iteratorType is set to AT_TIM [...]
"sendEmptyMessageWhenIdle": { "index": 8, "kind": "parameter",
"displayName": "Send Empty Message When Idle", "group": "consumer", "label":
"consumer", "required": false, "type": "boolean", "javaType": "boolean",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
false, "description": "If the polling consumer did not poll any files, you can
enable this option to send an empty message (no body) instead." },
"sequenceNumber": { "index": 9, "kind": "parameter", "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_SEQUEN [...]
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
b/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
index 2c81d405ee68..00c03c331a4b 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
+++
b/components/camel-aws/camel-aws2-kinesis/src/main/docs/aws2-kinesis-component.adoc
@@ -66,6 +66,43 @@ all available shards (multiple shards consumption) of Amazon
Kinesis, therefore,
property in the DSL configuration empty, then it'll consume all available
shards
otherwise only the specified shard corresponding to the shardId will be
consumed.
+=== Consumer throughput and the Kinesis read quotas
+
+In each poll the consumer sends one `GetRecords` request to each shard it
reads. The request asks for up to
+`maxResultsPerRequest` records (1000 by default; Kinesis accepts 1 to 10,000),
so a poll routes up to
+`maxResultsPerRequest` records per shard.
+
+Kinesis limits the reads of each shard:
+
+* 5 `GetRecords` requests per second;
+* 2 MiB of data per second;
+* at most 10,000 records or 10 MiB in one response.
+
+A request over these quotas fails with a
`ProvisionedThroughputExceededException`. After a response of 10 MiB, the
+requests of the next 5 seconds to that shard fail too.
+
+The scheduler options decide how often a shard is read:
+
+* With `greedy=false` (the default), the next poll starts `delay` milliseconds
(500 by default) after the previous one
+ ended. A shard then gets at most about 2 requests per second, so up to about
twice `maxResultsPerRequest` records
+ per second.
+* With `greedy=true`, the next poll starts at once after a poll that routed
records, and `delay` applies only after a
+ poll that routed none. With a backlog and a fast route, the consumer can
then send more than 5 requests per second to
+ a shard.
+
+To keep up with a shard, choose `maxResultsPerRequest` and `delay` so that the
records of a poll times the polls per
+second stay within 2 MiB per second. For example, 1000 records of 1 KiB every
500 ms is about 2 MiB per second. For
+larger records, lower `maxResultsPerRequest` or raise `delay`. While the
records of a response are routed, the
+consumer holds that response in memory, one for each shard of the poll.
+
+A xref:manual::graceful-shutdown.adoc[graceful shutdown] waits until the
consumer has routed all the records of the
+poll that is running. Choose `maxResultsPerRequest` so that the route handles
the records of a poll within the shutdown
+timeout (45 seconds by default). Otherwise the shutdown is forced, and the
records of the poll that are not routed yet
+fail with a `RouteStoppingException`.
+
+These options apply to the polling consumer only. The KCL consumer
(`useKclConsumers=true`) fetches the records with
+the settings of the Kinesis Client Library and does not use
`maxResultsPerRequest`, `delay` or `greedy`.
+
=== Custom Resume Action
The consumer supports a custom `KinesisResumeAction` for controlling where
each shard starts reading
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 fe6942707717..9a15c5de2f8c 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
@@ -49,9 +49,11 @@ public class Kinesis2Configuration implements Cloneable,
AwsCommonConfiguration
@UriParam(description = "Amazon Kinesis client to use for all requests for
this endpoint")
@Metadata(label = "advanced", autowired = true)
private KinesisClient amazonKinesisClient;
- @UriParam(label = "consumer", description = "Maximum number of records
that will be fetched in each poll",
- defaultValue = "1")
- private int maxResultsPerRequest = 1;
+ @UriParam(label = "consumer",
+ description = "Maximum number of records fetched from each shard
in each poll: the Limit of the GetRecords request,"
+ + " from 1 to 10000. Not used by the KCL consumer
(useKclConsumers).",
+ defaultValue = "1000")
+ private int maxResultsPerRequest = 1000;
@UriParam(label = "consumer", description = "Defines where in the Kinesis
stream to start getting records",
defaultValue = "TRIM_HORIZON")
private ShardIteratorType iteratorType = ShardIteratorType.TRIM_HORIZON;
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerGetRecordsLimitTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerGetRecordsLimitTest.java
new file mode 100644
index 000000000000..a8846142eb8d
--- /dev/null
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisConsumerGetRecordsLimitTest.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.kinesis;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.AsyncProcessorSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
+import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
+import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
+import software.amazon.awssdk.services.kinesis.model.ListShardsRequest;
+import software.amazon.awssdk.services.kinesis.model.ListShardsResponse;
+import software.amazon.awssdk.services.kinesis.model.Record;
+import software.amazon.awssdk.services.kinesis.model.SequenceNumberRange;
+import software.amazon.awssdk.services.kinesis.model.Shard;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+/**
+ * The Limit of the GetRecords requests of the polling consumer: {@code
maxResultsPerRequest} records from each shard in
+ * each poll, 1000 by default.
+ */
+@ExtendWith(MockitoExtension.class)
+public class KinesisConsumerGetRecordsLimitTest {
+
+ private static final int BACKLOG = 1500;
+
+ @Mock
+ private KinesisClient kinesisClient;
+
+ private final CamelContext context = new DefaultCamelContext();
+ private final ConcurrentLinkedQueue<Integer> limits = new
ConcurrentLinkedQueue<>();
+ private final AtomicInteger position = new AtomicInteger();
+ private final AtomicInteger routed = new AtomicInteger();
+
+ private Kinesis2Consumer consumer;
+
+ @BeforeEach
+ public void setup() {
+ Shard shard = Shard.builder().shardId("shard-0")
+
.sequenceNumberRange(SequenceNumberRange.builder().startingSequenceNumber("0").build()).build();
+ when(kinesisClient.listShards(any(ListShardsRequest.class)))
+
.thenReturn(ListShardsResponse.builder().shards(shard).build());
+
when(kinesisClient.getShardIterator(any(GetShardIteratorRequest.class)))
+
.thenReturn(GetShardIteratorResponse.builder().shardIterator("iterator").build());
+ // one shard with a backlog of BACKLOG records: a GetRecords returns
as many of them as its Limit asks for
+
when(kinesisClient.getRecords(any(GetRecordsRequest.class))).thenAnswer(invocation
-> {
+ GetRecordsRequest request = invocation.getArgument(0);
+ limits.add(request.limit());
+ List<Record> records = new ArrayList<>();
+ while (records.size() < request.limit() && position.get() <
BACKLOG) {
+ int sequenceNumber = position.getAndIncrement();
+
records.add(Record.builder().sequenceNumber(String.valueOf(sequenceNumber)).partitionKey("key")
+ .data(SdkBytes.fromString("record " + sequenceNumber,
StandardCharsets.UTF_8)).build());
+ }
+ return
GetRecordsResponse.builder().records(records).nextShardIterator("iterator").build();
+ });
+
+ context.getRegistry().bind("kinesisClient", kinesisClient);
+ context.start();
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ if (consumer != null) {
+ consumer.stop();
+ }
+ context.stop();
+ }
+
+ @Test
+ public void defaultLimitIsOneThousandRecords() throws Exception {
+
startConsumer("aws2-kinesis://stream?amazonKinesisClient=#kinesisClient");
+
+ assertEquals(1000, consumer.poll(), "the first poll routes 1000
records of the backlog");
+ assertEquals(500, consumer.poll(), "the second poll routes the rest of
the backlog");
+
+ assertEquals(List.of(1000, 1000), List.copyOf(limits));
+ assertEquals(BACKLOG, routed.get());
+ }
+
+ @Test
+ public void maxResultsPerRequestOneFetchesOneRecordPerPoll() throws
Exception {
+
startConsumer("aws2-kinesis://stream?amazonKinesisClient=#kinesisClient&maxResultsPerRequest=1");
+
+ assertEquals(1, consumer.poll());
+ assertEquals(1, consumer.poll());
+
+ assertEquals(List.of(1, 1), List.copyOf(limits));
+ assertEquals(2, routed.get());
+ }
+
+ private void startConsumer(String uri) throws Exception {
+ Kinesis2Endpoint endpoint = context.getEndpoint(uri,
Kinesis2Endpoint.class);
+ consumer = (Kinesis2Consumer) endpoint.createConsumer(new
AsyncProcessorSupport() {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ routed.incrementAndGet();
+ callback.done(true);
+ return true;
+ }
+ });
+ // the test calls poll()
+ consumer.setStartScheduler(false);
+ consumer.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
consumer.getCurrentShardList().size() == 1);
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointTest.java
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointTest.java
index b77a15ce10d1..d101ca11e6f5 100644
---
a/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointTest.java
+++
b/components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/KinesisEndpointTest.java
@@ -67,7 +67,7 @@ public class KinesisEndpointTest {
assertThat(endpoint.getConfiguration().getAmazonKinesisClient(),
is(amazonKinesisClient));
assertThat(endpoint.getConfiguration().getStreamName(),
is("some_stream_name"));
assertThat(endpoint.getConfiguration().getIteratorType(),
is(ShardIteratorType.TRIM_HORIZON));
- assertThat(endpoint.getConfiguration().getMaxResultsPerRequest(),
is(1));
+ assertThat(endpoint.getConfiguration().getMaxResultsPerRequest(),
is(1000));
}
@Test
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 eb43ec446ec8..10daa2619531 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
@@ -136,6 +136,9 @@ public class KinesisConsumerResumeIT extends
CamelTestSupport {
@Override
public void configure() {
final ResumeCache<Object> simpleCache =
TransientResumeStrategy.createSimpleCache();
+ // the consumer calls the resume action for each entry of the
cache: without an entry for the shard
+ // TestResumeAction is never called and the shard is read from
its first record
+ simpleCache.add("shardId-000000000000", "0");
final
KinesisResumeStrategyConfiguration.KinesisResumeStrategyConfigurationBuilder
resumeConfigurationBuilder
= KinesisResumeStrategyConfiguration.builder()
.withResumeCache(simpleCache);
@@ -204,6 +207,8 @@ public class KinesisConsumerResumeIT extends
CamelTestSupport {
and so on. This is just testing that the code is not mixing
things up.
*/
assertTrue(data.partition.endsWith(data.body), "The data/partition
mismatch for record: " + data);
+ // the consumer resumed at the record number expectedCount
+ assertTrue(Integer.parseInt(data.body) >= expectedCount, "A record
before the resume position: " + data);
}
}
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index cba635c09229..41c97b62f1f8 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -4267,6 +4267,24 @@ A shard created by a split or a merge while the consumer
is running is now read
`iteratorType`. Previously, with `LATEST`, the records stored in the new shard
before the consumer listed it were
skipped. The shards that exist when the consumer starts are read from the
configured `iteratorType`, as before.
+=== camel-aws2-kinesis - the consumer fetches up to 1000 records per shard in
each poll
+
+The default of `maxResultsPerRequest` is now `1000` (it was `1`). In each poll
the consumer asks each shard for up to
+that many records, so with the default `delay` of 500 ms it can read up to
about 2000 records per second from a shard
+(within the Kinesis quota of 2 MiB per second), instead of about 2. A poll now
routes more records, and holds up to
+1000 records per shard in memory while they are routed. The KCL consumer
(`useKclConsumers=true`) does not use this
+option.
+
+A graceful shutdown waits until the consumer has routed all the records of the
poll that is running. With 1000 records
+per shard and a route that takes 50 ms per record, a poll takes about 50
seconds, which is more than the default
+xref:graceful-shutdown.adoc[shutdown timeout] of 45 seconds. The shutdown is
then forced, and the records of the poll
+that are not routed yet fail with a `RouteStoppingException`. Before, a poll
held one record per shard. Lower
+`maxResultsPerRequest`, or raise the shutdown timeout, so that the records of
a poll are routed within the timeout.
+
+To keep the previous behaviour, set `maxResultsPerRequest=1`. See the
+xref:components::aws2-kinesis-component.adoc[AWS Kinesis component] for how
`maxResultsPerRequest`, `delay` and
+`greedy` relate to the Kinesis read quotas of a shard.
+
=== camel-graphql - the credentials are only sent to the authority of httpUri
The `accessToken` (with its `jwtAuthorizationType`) is now added to a request
only when the request goes to the
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 a66781fead76..0da5f5eb938a 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
@@ -183,11 +183,13 @@ public interface Aws2KinesisComponentBuilderFactory {
/**
- * Maximum number of records that will be fetched in each poll.
+ * Maximum number of records fetched from each shard in each poll: the
+ * Limit of the GetRecords request, from 1 to 10000. Not used by the
KCL
+ * consumer (useKclConsumers).
*
* The option is a: <code>int</code> type.
*
- * Default: 1
+ * Default: 1000
* Group: consumer
*
* @param maxResultsPerRequest the value to set
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 1c731d648672..8abe981d2b10 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
@@ -170,11 +170,13 @@ public interface Kinesis2EndpointBuilderFactory {
return this;
}
/**
- * Maximum number of records that will be fetched in each poll.
+ * Maximum number of records fetched from each shard in each poll: the
+ * Limit of the GetRecords request, from 1 to 10000. Not used by the
KCL
+ * consumer (useKclConsumers).
*
* The option is a: <code>int</code> type.
*
- * Default: 1
+ * Default: 1000
* Group: consumer
*
* @param maxResultsPerRequest the value to set
@@ -185,11 +187,13 @@ public interface Kinesis2EndpointBuilderFactory {
return this;
}
/**
- * Maximum number of records that will be fetched in each poll.
+ * Maximum number of records fetched from each shard in each poll: the
+ * Limit of the GetRecords request, from 1 to 10000. Not used by the
KCL
+ * consumer (useKclConsumers).
*
* The option will be converted to a <code>int</code> type.
*
- * Default: 1
+ * Default: 1000
* Group: consumer
*
* @param maxResultsPerRequest the value to set
