Amazon Kinesis Client Library (KCL) for Java

KCL performs lease balancing to protect against worker interruptions (e.g., host failures). If a worker stops updating its leases, other workers can take them over.

Key Concepts:

• Balancing Interval: Balancing occurs at an interval configured by leaseDuration and epsilonMillis.
• Lease Stealing: If a worker wants more leases and there are no expired leases available, it can "steal" a lease. Stolen leases are randomly selected from the worker currently holding the most leases.
• Configuration: The maximum number of leases a worker can steal in a single loop is controlled by maxLeasesToStealAtOneTime.

Trade-offs to consider:

1. DynamoDB Cost: LeaseRefresher performs a scan on the lease table; frequent scans increase costs proportional to the number of leases.
2. Turnover vs. Stability: Frequent balancing increases DynamoDB write costs and can cause redundant work due to high lease turnover.
3. Recovery Speed: A low maxLeasesToStealAtOneTime value may slow down the reassignment of leases during major events like deployments or host failures.

KCL 3.x uses a leader-based approach for lease assignment and rebalancing, which differs significantly from the distributed approach in KCL 2.x.

In KCL 3.x, a single worker is elected as the leader. This leader is responsible for:

1. Scanning the DynamoDB lease table to gather global state.
2. Deciding on an optimal lease assignment for all workers.
3. Updating the leases in the table.

Key improvements in the 3.x rebalancing model include:

• Throughput-aware rebalancing: The leader considers actual shard throughput (bytes processed) to distribute "hot shards" more evenly, rather than just counting shards.
• CPU utilization metrics: The leader uses worker metrics (primarily CPU utilization) to keep each worker's load within a specified threshold, redistributing leases away from overloaded workers.
• Reduced DynamoDB usage: Only the leader performs full table scans. Other workers use a Global Secondary Index (GSI) that mirrors the leaseKey attribute with a partition key of leaseOwner for efficient discovery.
• Graceful Lease Handover: To minimize duplicate processing, the worker relinquishing a lease checkpoints the last processed record before the new worker resumes processing.

In Amazon ECS, KCL 3.x retrieves container-level metrics (CPU, memory, and network) via the ECS task metadata endpoint.

Requirements:

• The container must have the ${ECS_CONTAINER_METADATA_URI_V4} environment variable injected by the ECS agent.

Mechanism:

• KCL accesses the local endpoint provided by ${ECS_CONTAINER_METADATA_URI_V4} to retrieve Docker stats.
• CPU utilization is calculated using the formula: cpuUtilization = (cpuDelta / SystemDelta) * online_cpus * 100.0 where:
  • cpuDelta = total_usage - prev_total_usage
  • systemDelta = system_cpu_usage - prev_system_cpu_usage

Note on CPU Limits: KCL determines the maximum number of CPUs available to the container by calculating the ratio of the container's CPU shares to the total task CPU size. This ensures that utilization is reported relative to the container's allocated capacity rather than the entire EC2 host.

KCL 3.x uses a centralized leader decider to manage coordination tasks. The elected leader performs the following:

1. PeriodicShardSync: Periodically ensures the lease table matches the active shards in the stream.
2. Stream Discovery: Discovers new streams that need processing (in multi-stream mode).
3. Lease Assignment: Performs the global lease assignment for all KCL workers.

Leader Election Mechanism

KCL 3.x has moved away from the KCL 2.x DeterministicShuffleShardSyncLeaderDecider (which relied on expensive DynamoDB table scans and a 5-minute heartbeat) to the DynamoDBLockBasedLeaderDecider.

This new implementation uses the DynamoDBLockClient library. The election process follows a first-worker-wins model:

1. A worker reads a single lock entry from the DynamoDB lock table.
2. It validates the active status of the current lock owner by checking for a constant heartbeat.
3. If the previous owner's heartbeat has expired, the worker claims leadership.

A lease is a data object that binds a specific Kinesis shard to a specific KCL worker. This binding allows a fleet of workers to partition data processing across the stream.

Leases follow a progressive state machine:

1. DISCOVERY: KCL identifies new shards via shard syncing (e.g., from an empty lease table, stream splits/merges, or new streams in multi-stream mode).
2. CREATION: A 1:1 lease is created for each discovered shard. Child shards only get leases once their parent(s) reach SHARD_END. Child leases are initialized at TRIM_HORIZON to prevent processing gaps.
3. PROCESSING: The primary state where the worker processes records and continually updates checkpoints in the lease table.
4. SHARD_END: The shard is marked as finished, meaning all records have been processed.
5. DELETION: The lease is removed from the lease table. To ensure durability, deletion only occurs after its child leases have entered the PROCESSING state. This behavior is configurable via lease-cleanup-config.

KCL 3.x uses a variance-based approach to trigger rebalancing based on worker metrics (typically CPU utilization).

Rebalancing Logic

1. Calculate Average: Determine the average worker metric (e.g., CPU utilization) across all workers.
2. Compute Thresholds: Calculate upper and lower limits using the reBalanceThresholdPercentage (default: 10).
   • Upper limit: average worker metric * (1 + reBalanceThresholdPercentage / 100)
   • Lower limit: average worker metric * (1 - reBalanceThresholdPercentage / 100)
3. Trigger: Rebalancing is triggered if any worker's metric falls outside these limits.
4. Calculate Load Delta: Determine the amount of load to move to bring the worker back to the average.
5. Apply Dampening: Apply the dampeningPercentageValue (default: 80) to the calculated load to prevent oscillation and achieve critical damping.
6. Convert to Throughput: Calculate the specific throughput target based on the dampened load.
7. Reassign: Select and reassign specific leases from over-utilized workers to under-utilized workers to match the target.

Fallback Mechanism

On platforms that do not support CPU utilization metrics (e.g., Windows), KCL 3.x automatically falls back to a throughput-based balancing mechanism. In this mode, it distributes leases based on shard throughput to ensure all workers process an equal throughput of shards.