AWS Kinesis

At the component level, you set general and shared configurations that are, then, inherited by the endpoints. It is the highest configuration level.

For example, a component may have security settings, credentials for authentication, urls for network connection and so forth.

Some components only have a few options, and others may have many. Because components typically have pre-configured defaults that are commonly used, then you may often only need to configure a few options on a component; or none at all.

You can configure components using:

You usually spend more time setting up endpoints because they have many options. These options help you customize what you want the endpoint to do. The options are also categorized into whether the endpoint is used as a consumer (from), as a producer (to), or both.

Configuring endpoints is most often done directly in the endpoint URI as path and query parameters. You can also use the Endpoint DSL and DataFormat DSL as a type safe way of configuring endpoints and data formats in Java.

A good practice when configuring options is to use Property Placeholders.

Property placeholders provide a few benefits:

The following two sections list all the options, firstly for the component followed by the endpoint.

This component implements the Batch Consumer.

This allows you, for instance, to know how many messages exist in this batch and for instance, let the Aggregator aggregate this number of messages.

The consumer is able to consume either from a single specific shard or all available shards (multiple shards consumption) of Amazon Kinesis, therefore, if you leave the 'shardId' 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.

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:

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:

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 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.

The consumer supports a custom KinesisResumeAction for controlling where each shard starts reading (e.g., resuming from a persisted sequence number). Register a subclass in the Camel registry under the key CamelKinesisDbResumeAction.

The consumer creates a separate instance per shard via reflection to avoid concurrent mutation when multiple shards are processed in parallel. Because of this, custom subclasses must provide a public no-arg constructor. Per-shard state (builder, shardId, streamName) is injected via setters after construction.

When a shard needs a new shard iterator, evalEntry is called for every entry of the resume cache, with the shard id as key and the sequence number (a String, or the KinesisOffset stored by KinesisResumeStrategy) as value. The default action starts the shard after the sequence number of its own entry and ignores the entries of the other shards; a subclass should compare the key with getShardId() the same way.

This component implements the Batch Producer.

This allows you to send multiple messages in a single request to Amazon Kinesis. A batch can have any number of records: the producer splits it into PutRecords requests of at most 500 records and 10 MiB, the limits of a request (the size of a record is its data plus its partition key). A record above 10 MiB, the largest record that Kinesis accepts, fails the exchange with an IllegalArgumentException before any request is sent. A stream can have a lower limit for the size of its records (1 to 10 MiB), which Kinesis checks.

The batch type needs to implement the Iterable interface. For example, it can be a List, Set or any other collection type. The message type can be one or more of types byte[], ByteBuffer, UTF-8 String, or InputStream. Other types are not supported. A java.nio.file.Path body is not a batch: like a java.io.File, it is sent as a single record with the content of the file.

A PutRecords request can be partly written: Kinesis writes some of its records and returns an error code for the others (for example ProvisionedThroughputExceededException). The producer then sends only the records that failed again, up to 3 times, waiting 100, 200 and 400 milliseconds plus a random part of up to a quarter of the delay (so that producers throttled at the same time do not all send again at the same time), and then goes on with the next requests of the batch. When every record fails, for example when the whole stream is throttled, each request waits 700 to 875 milliseconds in total before its records are counted as not written, so a batch of 10,000 records (20 requests of 500 records) can block the caller for 14 to 17.5 seconds, plus the time of the 80 calls and of the retries of the AWS SDK client.

A record that is sent again is written after the next records of its request: if record A fails and record B, with the same partition key, is written, the stream has B before A. PutRecords does not guarantee the order of the records either. When the order of the records of a partition key matters, send them one by one, with a body that is not an Iterable.

After the last request, the CamelAwsKinesisRecordCount header has the number of records of the batch, and the CamelAwsKinesisFailedRecordCount header the number of records that are still not written. When that number is not 0, the exchange fails with an exception that gives both numbers and the first error code. When a PutRecords call throws, for example an InvalidArgumentException, a KMS error, or an error of the AWS SDK client after its own retries, the exchange fails at once with that exception, and the next requests of the batch are not sent. Both headers are set in that case too: the records of that call and of the requests that were not sent are counted as not written, even if the call may have written some of them (for example after a timeout).

A failure does not undo the records that are already written. A redelivery of the exchange, by the error handler of the route or by the source of the message (for example a transacted JMS or an SQS consumer), sends the whole batch again, so the records that were written before the failure are written twice. PutRecords has no idempotency token: a request whose outcome is unknown, for example after a timeout, can also write a record twice when it is sent again, for example by the retries of the AWS SDK client. A consumer of the stream that must not process a record twice needs an id in the data of the record to detect a duplicate.

A body that is not an Iterable is sent as a single record, converted to a byte[]. A body that does not convert to a byte[] fails with an InvalidPayloadException.

You have the possibility of avoiding the usage of explicit static credentials by specifying the useDefaultCredentialsProvider option and set it to true.

The order of evaluation for Default Credentials Provider is the following:

You have also the possibility of using Profile Credentials Provider, by specifying the useProfileCredentialsProvider option to true and profileCredentialsName to the profile name.

Only one of static, default and profile credentials could be used at the same time.

For more information about this you can look at AWS credentials documentation

You then have to reference the KinesisClient in the amazonKinesisClient URI option.

It is recommended that the credentials are obtained by using the DefaultAWSCredentialsProviderChain that is the default when creating a new ClientConfiguration instance, however, a different AWSCredentialsProvider can be specified when calling createClient(…​).

The component supports also the KCL (Kinesis Client Library) for consuming from a Kinesis Data Stream.

To enable this feature you’ll need to set three different parameter in your endpoint and set the region:

This feature will make possible to automatically checkpointing the Shard Iterations by combining the usage of KCL, DynamoDB Table and CloudWatch alarms.

Everything will work out of the box, by simply using your AWS Credentials.

By default, a record whose exchange fails (the route ends with an exception) is reported to the exception handler of the consumer and checkpointed with its batch, so it is not consumed again. Use a dead letter channel or redelivery in the route when a failed record must not be lost.

With kclStopCheckpointOnFailure=true, the checkpoint of a shard stays before the first record of it that failed. KCL does not deliver records of a shard again while the consumer holds the lease of the shard, so the consumer goes on routing the next records, but it checkpoints only the records before the failed one, also when the route stops. The next holder of the lease (the route started again, or another worker taking over the lease) consumes the records from the failed one on again. This has a cost: