Using the Kinesis Client Library

Connects to the data stream

Enumerates the shards within the data stream

Uses leases to coordinates shard associations with its workers

Instantiates a record processor for every shard it manages

Pulls data records from the data stream

Pushes the records to the corresponding record processor

Checkpoints processed records

Balances shard-worker associations (leases) when the worker instance count changes or when the data stream is resharded (shards are split or merged)

KCL 1.x

For more information, see Developing KCL 1.x Consumers

KCL 2.x

For more information, see Developing KCL 2.x Consumers

KCL consumer application – an application that is custom-built using KCL and designed to read and process records from data streams.

Consumer application instance - KCL consumer applications are typically distributed, with one or more application instances running simultaneously in order to coordinate on failures and dynamically load balance data record processing.

Worker – a high level class that a KCL consumer application instance uses to start processing data.

The worker initializes and oversees various tasks, including syncing shard and lease information, tracking shard assignments, and processing data from the shards. A worker provides KCL with the configuration information for the consumer application, such as the name of the data stream whose data records this KCL consumer application is going to process and the AWS credentials that are needed to access this data stream. The worker also kick starts that specific KCL consumer application instance to deliver data records from the data stream to the record processors.

Lease – data that defines the binding between a worker and a shard. Distributed KCL consumer applications use leases to partition data record processing across a fleet of workers. At any given time, each shard of data records is bound to a particular worker by a lease identified by the leaseKey variable.

By default, a worker can hold one or more leases (subject to the value of the maxLeasesForWorker variable) at the same time.

For example, if you have a consumer application instance A with worker A that is processing a data stream with 4 shards, worker A can hold leases to shards 1, 2, 3, and 4 at the same time. But if you have two consumer application instances: A and B with worker A and worker B, and these instances are processing a data stream with 4 shards, worker A and worker B cannot both hold the lease to shard 1 at the same time. One worker holds the lease to a particular shard until it is ready to stop processing this shard’s data records or until it fails. When one worker stops holding the lease, another worker takes up and holds the lease.

For more information, (these are the Java KCL repositories), see https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java for KCL 1.x and https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java for KCL 2.x.

Lease table - a unique Amazon DynamoDB table that is used to keep track of the shards in a KDS data stream that are being leased and processed by the workers of the KCL consumer application. The lease table must remain in sync (within a worker and across all workers) with the latest shard information from the data stream while the KCL consumer application is running. For more information, see Using a Lease Table to Track the Shards Processed by the KCL Consumer Application.

Record processor – the logic that defines how your KCL consumer application processes the data that it gets from the data streams. At runtime, a KCL consumer application instance instantiates a worker, and this worker instantiates one record processor for every shard to which it holds a lease.

checkpoint: The most recent checkpoint sequence number for the shard. This value is unique across all shards in the data stream.

checkpointSubSequenceNumber: When using the Kinesis Producer Library's aggregation feature, this is an extension to checkpoint that tracks individual user records within the Kinesis record.

leaseCounter: Used for lease versioning so that workers can detect that their lease has been taken by another worker.

leaseKey: A unique identifier for a lease. Each lease is particular to a shard in the data stream and is held by one worker at a time.

leaseOwner: The worker that is holding this lease.

ownerSwitchesSinceCheckpoint: How many times this lease has changed workers since the last time a checkpoint was written.

parentShardId: Used to ensure that the parent shard is fully processed before processing starts on the child shards. This ensures that records are processed in the same order they were put into the stream.

hashrange: Used by the PeriodicShardSyncManager to run periodic syncs to find missing shards in the lease table and create leases for them if required.

childshards: Used by the LeaseCleanupManager to review the child shard's processing status and decide whether the parent shard can be deleted from the lease table.

shardID: The ID of the shard.

stream name The identifier of the data stream in the following format: account-id:StreamName:streamCreationTimestamp.

Fetches all the shards for data the stream that is being processed

Fetches all the shard leases from the lease table

Filters out each open shard that does not have a lease in the lease table

Iterates over all found open shards and for each open shard with no open parent:

During application's bootstraping, if the lease table is empty, KCL utilizes the ListShard API's filtering option (the ShardFilter optional request parameter) to retrieve and create leases only for a snapshot of shards open at the time specified by the ShardFilter parameter. The ShardFilter parameter enables you to filter out the response of the ListShards API. The only required property of the ShardFilter parameter is Type. KCL uses the Type filter property and the following of its valid values to identify and return a snapshot of open shards that might require new leases:

ShardFilter is used when creating leases for an empty lease table to initialize leases for a snapshot of shards specified at RetrievalConfig#initialPositionInStreamExtended.

For more information about ShardFilter, see https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Instead of all workers performing the lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard synchronization.

KCL 2.3 uses the ChildShards return parameter of the GetRecords and the SubscribeToShard APIs to perform lease/shard synchronization that happens at SHARD_END for closed shards, allowing a KCL worker to only create leases for the child shards of the shard it finished processing. For shared throughout consumer applications, this optimization of the lease/shard synchronization uses the ChildShards parameter of the GetRecords API. For the dedicated throughput (enhanced fan-out) consumer applications, this optimization of the lease/shard synchronization uses the ChildShards parameter of the SubscribeToShard API. For more information, see GetRecords, SubscribeToShards, and ChildShard.

With the above changes, the behavior of KCL is moving from the model of all workers learning about all existing shards to the model of workers learning only about the children shards of the shards that each worker owns. Therefore, in addition to the synchronization that happens during consumer application bootstraping and reshard events, KCL now also performs additional periodic shard/lease scans in order to identify any potential holes in the lease table (in other words, to learn about all new shards) to ensure the complete hash range of the data stream is being processed and create leases for them if required. PeriodicShardSyncManager is the component that is responsible for running periodic lease/shard scans.

For more information about PeriodicShardSyncManager in KCL 2.3, see https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java#L201-L213.

In KCL 2.3, new configuration options are available to configure PeriodicShardSyncManager in LeaseManagementConfig:

New CloudWatch metrics are also now emitted to monitor the health of the PeriodicShardSyncManager. For more information, see PeriodicShardSyncManager.

Including an optimization to HierarchicalShardSyncer to only create leases for one layer of shards.

During application's bootstraping, if the lease table is empty, KCL utilizes the ListShard API's filtering option (the ShardFilter optional request parameter) to retrieve and create leases only for a snapshot of shards open at the time specified by the ShardFilter parameter. The ShardFilter parameter enables you to filter out the response of the ListShards API. The only required property of the ShardFilter parameter is Type. KCL uses the Type filter property and the following of its valid values to identify and return a snapshot of open shards that might require new leases:

ShardFilter is used when creating leases for an empty lease table to initialize leases for a snapshot of shards specified at KinesisClientLibConfiguration#initialPositionInStreamExtended.

For more information about ShardFilter, see https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Instead of all workers performing the lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard synchronization.

KCL 1.14 uses the ChildShards return parameter of the GetRecords and the SubscribeToShard APIs to perform lease/shard synchronization that happens at SHARD_END for closed shards, allowing a KCL worker to only create leases for the child shards of the shard it finished processing. For more information, see GetRecords and ChildShard.

With the above changes, the behavior of KCL is moving from the model of all workers learning about all existing shards to the model of workers learning only about the children shards of the shards that each worker owns. Therefore, in addition to the synchronization that happens during consumer application bootstraping and reshard events, KCL now also performs additional periodic shard/lease scans in order to identify any potential holes in the lease table (in other words, to learn about all new shards) to ensure the complete hash range of the data stream is being processed and create leases for them if required. PeriodicShardSyncManager is the component that is responsible for running periodic lease/shard scans.

When KinesisClientLibConfiguration#shardSyncStrategyType is set to ShardSyncStrategyType.SHARD_END, PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold is used to determine the threshold for number of consecutive scans containing holes in the lease table after which to enforce a shard synchronization. When KinesisClientLibConfiguration#shardSyncStrategyType is set to ShardSyncStrategyType.PERIODIC, leasesRecoveryAuditorInconsistencyConfidenceThreshold is ignored.

For more information about PeriodicShardSyncManager in KCL 1.14, see https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java#L987-L999.

In KCL 1.14, new configuration option is available to configure PeriodicShardSyncManager in LeaseManagementConfig:

New CloudWatch metrics are also now emitted to monitor the health of the PeriodicShardSyncManager. For more information, see PeriodicShardSyncManager.

KCL 1.14 now also supports deferred lease cleanup. Leases are deleted asynchronously by LeaseCleanupManager upon reaching SHARD_END, when a shard has either expired past the data stream’s retention period or been closed as the result of a resharding operation.

New configuration options are available to configure LeaseCleanupManager.

Including an optimization to KinesisShardSyncer to only create leases for one layer of shards.

MultistreamTracker interface

To build a consumer application that can process multiple streams at the same time, you must implement a new interface called MultistreamTracker. This interface includes the streamConfigList method that returns the list of data streams and their configurations to be processed by the KCL consumer application. Notice that the data streams that are being processed can be changed during the consumer application runtime. streamConfigList is called periodically by the KCL to learn about the changes in data streams to process.

The streamConfigList method populates the StreamConfig list.

package software.amazon.kinesis.common;

import lombok.Data;
import lombok.experimental.Accessors;

@Data
@Accessors(fluent = true)
public class StreamConfig {
    private final StreamIdentifier streamIdentifier;
    private final InitialPositionInStreamExtended initialPositionInStreamExtended;
    private String consumerArn;
}
                

Note that the StreamIdentifier and InitialPositionInStreamExtended are required fields, while consumerArn is optional. You must provide the consumerArn only if you are using KCL 2.x to implement an enhanced fan-out consumer application.

For more information about StreamIdentifier, see https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java#L129. To create a StreamIdentifier, we recommend that you create a multistream instance from the streamArn and the streamCreationEpoch that is available in v2.5.0 and later. In KCL v2.3 and v2.4, which don't support streamArm, create a multistream instance by using the format account-id:StreamName:streamCreationTimestamp. This format will be deprecated and no longer supported starting with the next major release.

MultistreamTracker also includes a strategy for deleting leases of old streams in the lease table (formerStreamsLeasesDeletionStrategy). Notice that the strategy CANNOT be changed during the consumer application runtime. For more information, see https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b/amazon-kinesis-client/src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy.java

ConfigsBuilder is a an application-wide class that you can use to specify all of the KCL 2.x configuration settings to be used when building your KCL consumer application. ConfigsBuilder class now has support for the MultistreamTracker interface. You can initialize ConfigsBuilder either with the name of the one data stream to consume records from:

 /**
     * Constructor to initialize ConfigsBuilder with StreamName
     * @param streamName
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull String streamName, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.right(streamName);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }

Or you can initialize ConfigsBuilder with MultiStreamTracker if you want to implement a KCL consumer application that processes multiple streams at the same time.

* Constructor to initialize ConfigsBuilder with MultiStreamTracker
     * @param multiStreamTracker
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull MultiStreamTracker multiStreamTracker, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.left(multiStreamTracker);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }                    
                

With multistream support implemented for your KCL consumer application, each row of the application's lease table now contains the shard ID and the stream name of the multiple data streams that this application processes.

When multistream support for your KCL consumer application is implemented, the leaseKey takes the following structure: account-id:StreamName:streamCreationTimestamp:ShardId. For example, 111111111:multiStreamTest-1:12345:shardId-000000000336.