KCL 2.x for MCP Servers — Kinesis Client Library Lease Management, Checkpointing, and SHARD_END Handling

DynamoDB lease table structure

KCL 2.x automatically creates a lease table named after your application (e.g., alivemcp-probe-consumer). Each row corresponds to one Kinesis shard:

Lease renewal happens every 10 seconds by default (failoverTimeMillis). If a worker fails to renew its lease within this window (e.g., the ECS task is killed), KCL marks the lease as expired and another worker steals it. The new worker resumes from the last checkpointed sequence number — records between the last checkpoint and the crash point are reprocessed. This is why idempotent processing (upsert probe results by probeRunId rather than insert) is essential on the consumer side.

DynamoDB capacity for the lease table

On-demand capacity mode is almost always correct for KCL lease tables. Lease renewals are low-frequency writes (one per worker per 10 seconds), but they spike when the consumer fleet scales out: all new workers immediately claim leases. With 16 shards and 8 workers, steady-state is ~16 writes/10s = 1.6 WCU, but a cold start of 8 workers simultaneously can produce 16 conditional writes in <1 second. On-demand handles this spike without throttling.

ShardRecordProcessor interface

KCL 2.x requires implementing four lifecycle methods:

import { RecordProcessor, RecordProcessorInput,
         ProcessRecordsInput, InitializationInput,
         LeaseLostInput, ShardEndedInput } from 'amazon-kinesis-client';

class ProbeResultProcessor implements RecordProcessor {

  private shardId: string = '';

  async initialize(input: InitializationInput): Promise {
    this.shardId = input.shardId;
    // input.extendedSequenceNumber: starting position (checkpoint or TRIM_HORIZON/LATEST)
    console.log(`Initialized processor for shard ${this.shardId} at ${input.extendedSequenceNumber}`);
  }

  async processRecords(input: ProcessRecordsInput): Promise {
    for (const record of input.records) {
      const probeResult = JSON.parse(
        Buffer.from(record.data as Uint8Array).toString('utf-8')
      );
      await upsertProbeResult(probeResult); // idempotent by probeRunId
    }

    // Checkpoint once per batch — durable as long as DynamoDB write succeeds
    // Checkpoint per-record is safer but 10-1000× more expensive on DynamoDB
    await input.checkpointer.checkpoint();
  }

  async leaseLost(input: LeaseLostInput): Promise {
    // Called when this worker loses the shard lease to another worker
    // Do NOT checkpoint here — the new owner will resume from last checkpoint
    console.log(`Lease lost for shard ${this.shardId}`);
  }

  async shardEnded(input: ShardEndedInput): Promise {
    // Shard has been split or merged — no more records on this shard
    // MUST call checkpoint() here with SHARD_END sentinel
    // Without this, KCL will not mark the shard complete and child shards will not start
    await input.checkpointer.checkpoint();
    console.log(`Shard ${this.shardId} ended — child shards will begin`);
  }

  async shutdownRequested(input: any): Promise {
    // Graceful shutdown — checkpoint current progress before exit
    await input.checkpointer.checkpoint();
  }
}

Checkpointing frequency tradeoff

For AliveMCP probe results, per-batch checkpointing is correct: probe results are idempotent (upsert by probeRunId), so reprocessing a batch on crash simply re-applies the same upsert and the result is identical. The DynamoDB cost of per-record checkpointing (one conditional write per probe result record) is not justified when processing is idempotent.

SHARD_END handling after split and merge

When KDS reshards a stream (splits a hot shard or merges two underloaded shards), the parent shard eventually runs out of records and reaches SHARD_END. KCL 2.x detects this and calls shardEnded on your processor with a special checkpointer that marks the shard complete in the lease table.

The critical invariant: KCL will not start processing child shards until all parent shards have been checkpointed at SHARD_END. If you return from shardEnded without calling checkpoint(), the child shard rows in the lease table remain in waiting state indefinitely, and new probe results written to those child shards accumulate unread.

Split shard topology

# Before split:
shard-0000000000: leaseKey, leaseOwner=worker-1, checkpoint=...latest...

# After split — shard-0 is closed, two children created:
shard-0000000000: checkpoint=SHARD_END (marked complete)
shard-0000000001: parentShardIds=['shardId-000000000000'], checkpoint=TRIM_HORIZON
shard-0000000002: parentShardIds=['shardId-000000000000'], checkpoint=TRIM_HORIZON

# KCL assigns both child shards to available workers only after parent=SHARD_END

Merge shard topology

On merge, two parent shards are closed and one child shard is created. The child's parentShardIds attribute contains both parent shard IDs. KCL waits for both parents to be checkpointed at SHARD_END before beginning the child. This ensures no records from either parent stream are skipped when the merged shard starts processing.

MultiLangDaemon for Node.js consumers

KCL 2.x is a Java library. For Node.js consumers, AWS provides the MultiLangDaemon (MLD) — a Java subprocess that handles all Kinesis API calls and lease management, communicating with your Node.js process via a stdin/stdout JSON protocol. Your Node.js code implements the RecordProcessor interface using the amazon-kinesis-client npm package, which wraps the MLD protocol:

# package.json dependencies
{
  "dependencies": {
    "amazon-kinesis-client": "^3.0.0"  // wraps MLD protocol
  }
}

# Start command — launches the Java MLD subprocess which in turn starts your Node.js processor
node_modules/.bin/kcl-bootstrap \
  --java /usr/bin/java \
  --properties kcl.properties \
  -- node src/consumer.js

The KCL properties file configures the stream name, application name (= lease table name), region, and initial position:

# kcl.properties
streamName=alivemcp-probe-results
applicationName=alivemcp-probe-consumer
AWSCredentialsProvider=DefaultAWSCredentialsProviderChain
processingLanguage=Node.js/v20
regionName=us-east-1
initialPositionInStream=LATEST
# Enhanced Fan-Out: fanout EFO consumer ARN (optional — omit for standard polling)
consumerArn=arn:aws:kinesis:us-east-1:123456789012:stream/alivemcp-probe-results/consumer/alivemcp-probe-consumer:1234567890

MLD requires Java 11+ in the container. For ECS Fargate tasks, extend the Node.js base image to install the JRE: RUN apt-get install -y default-jre-headless. This adds ~200 MB to the image; if image size is a constraint, consider using Lambda with the built-in KDS event source mapping instead (which does not require MLD).

CloudWatch monitoring for KCL consumers

KCL 2.x emits per-shard MillisBehindLatest to CloudWatch using the application name as a dimension. Set up a CloudWatch alarm per-shard or use a metric math expression of MAX(MillisBehindLatest across all shards) to catch any single shard falling behind. For AliveMCP, a shard that is 1 full probe cycle (60 seconds = 60,000ms) behind indicates the consumer is not keeping up with ingest — investigate consumer CPU, DynamoDB throttling, or downstream sink latency.

Common failure modes