DynamoDB Streams Kinesis Adapter — KCL Distributed Processing for MCP Servers

Architecture: how the adapter bridges DynamoDB Streams to KCL

KCL uses three Kinesis Data Streams API calls to consume records: ListShards, GetShardIterator, and GetRecords. DynamoDB Streams exposes equivalent calls under different names (DescribeStream, GetShardIterator, GetRecords), and the adapter translates between them transparently.

The adapter maps DynamoDB Streams records to KCL's Record interface by serializing the DynamoDB Streams record (including eventName, dynamodb.NewImage, dynamodb.OldImage, and dynamodb.Keys) as JSON in the KCL record's data payload. Your ShardRecordProcessor deserializes this JSON to access the DynamoDB change event.

// Gradle dependency for the adapter
// build.gradle
dependencies {
    implementation 'com.amazonaws:dynamodbstreams-kinesis-adapter:1.6.0'
    implementation 'software.amazon.kinesis:amazon-kinesis-client:2.5.0'
    implementation 'com.amazonaws:aws-java-sdk-dynamodb:1.12.500'
}

// Java: configure KCL 2.x with DynamoDB Streams Kinesis Adapter
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.streamsadapter.AmazonDynamoDBStreamsAdapterClient;
import software.amazon.kinesis.coordinator.Scheduler;
import software.amazon.kinesis.common.ConfigsBuilder;

// 1. Create the adapter client wrapping DynamoDB Streams
AmazonDynamoDB streamsClient = AmazonDynamoDBClientBuilder.standard()
    .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
        "https://streams.dynamodb.us-east-1.amazonaws.com", "us-east-1"))
    .build();
AmazonDynamoDBStreamsAdapterClient adapterClient = new AmazonDynamoDBStreamsAdapterClient(streamsClient);

// 2. Standard KCL 2.x setup, passing the adapter client as the Kinesis client
ConfigsBuilder configsBuilder = new ConfigsBuilder(
    streamArn,              // DynamoDB Streams ARN (not a Kinesis stream ARN)
    "alivemcp-probe-processor",  // Application name — becomes DynamoDB lease table name
    adapterClient,          // The adapter, implementing AmazonKinesis interface
    dynamoDBForLeaseTable,  // Separate DynamoDB client for the KCL lease table
    cloudWatchClient,
    UUID.randomUUID().toString(),  // Worker ID — unique per worker process
    new ProbeStateRecordProcessorFactory()
);

Scheduler scheduler = new Scheduler(
    configsBuilder.checkpointConfig(),
    configsBuilder.coordinatorConfig(),
    configsBuilder.leaseManagementConfig(),
    configsBuilder.lifecycleConfig(),
    configsBuilder.metricsConfig(),
    configsBuilder.processorConfig(),
    configsBuilder.retrievalConfig()
);

// Runs in a background thread; call scheduler.startGracefulShutdown() to stop
Thread schedulerThread = new Thread(scheduler);
schedulerThread.start();

ShardRecordProcessor implementation for DynamoDB Streams

Your ShardRecordProcessor receives KCL Record objects whose data payload is the serialized DynamoDB Streams change event. The processor must deserialize this JSON and dispatch based on eventName:

import software.amazon.kinesis.processor.ShardRecordProcessor;
import software.amazon.kinesis.lifecycle.events.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Map;

public class ProbeStateRecordProcessor implements ShardRecordProcessor {
    private static final ObjectMapper MAPPER = new ObjectMapper();

    @Override
    public void initialize(InitializationInput initializationInput) {
        // Called once when this worker takes ownership of a shard
        System.out.println("Initialized shard: " + initializationInput.shardId());
    }

    @Override
    public void processRecords(ProcessRecordsInput processRecordsInput) {
        for (var record : processRecordsInput.records()) {
            try {
                // KCL record data is UTF-8 JSON of the DynamoDB Streams record
                String json = StandardCharsets.UTF_8.decode(record.data()).toString();
                Map<String, Object> streamRecord = MAPPER.readValue(json, Map.class);

                String eventName = (String) streamRecord.get("eventName");
                Map<String, Object> dynamoDb = (Map<String, Object>) streamRecord.get("dynamodb");
                Map<String, Object> newImage = (Map<String, Object>) dynamoDb.get("NewImage");

                if ("INSERT".equals(eventName) || "MODIFY".equals(eventName)) {
                    Map<String, Object> statusAttr = (Map<String, Object>) newImage.get("status");
                    String status = (String) statusAttr.get("S");
                    if ("DEGRADED".equals(status) || "DOWN".equals(status)) {
                        triggerAlert(dynamoDb, status);
                    }
                }
            } catch (Exception e) {
                // Log and continue — do NOT rethrow; rethrowing halts shard processing
                System.err.println("Failed to process record: " + e.getMessage());
            }
        }
        // Checkpoint after processing the batch — at-least-once semantics
        // (records may be re-delivered if worker fails before next checkpoint)
        try {
            processRecordsInput.checkpointer().checkpoint();
        } catch (Exception e) {
            System.err.println("Checkpoint failed: " + e.getMessage());
        }
    }

    @Override
    public void leaseLost(LeaseLostInput leaseLostInput) {
        // Another worker took ownership of this shard — clean up resources
        // Do NOT checkpoint here (we no longer own the lease)
    }

    @Override
    public void shardEnded(ShardEndedInput shardEndedInput) {
        // Shard has expired (24-hour DynamoDB Streams shard lifetime reached)
        // MUST checkpoint SHARD_END or child shards will not be processed
        try {
            shardEndedInput.checkpointer().checkpoint();
        } catch (Exception e) {
            System.err.println("SHARD_END checkpoint failed: " + e.getMessage());
        }
    }

    @Override
    public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
        // Graceful shutdown: checkpoint current position before worker exits
        try {
            shutdownRequestedInput.checkpointer().checkpoint();
        } catch (Exception e) {
            System.err.println("Shutdown checkpoint failed: " + e.getMessage());
        }
    }
}

DynamoDB Streams shard lifecycle — 24-hour windows

DynamoDB Streams shards have a fixed 24-hour lifetime, unlike Kinesis Data Streams shards (which remain active indefinitely unless the stream is resharded). When a shard expires, all records in that shard are still readable for 24 hours after the shard's first record was written — but the shard stops accepting new records. A new "child" shard is created to receive new changes for the same DynamoDB partition.

The most critical operational requirement: always call checkpointer.checkpoint() in shardEnded(). If you do not checkpoint the SHARD_END sentinel sequence number, KCL considers the parent shard incomplete and will never start processing the child shard. This causes your processor to silently fall behind and stop receiving new changes for all partitions that underwent a shard transition — which happens to every DynamoDB partition every 24 hours.

Lambda ESM vs KCL adapter: decision framework

For most MCP server monitoring deployments, Lambda ESM is the right choice. The KCL adapter adds operational complexity (managing a long-lived process fleet, lease table maintenance, JVM heap tuning) that is only justified at scale:

For AliveMCP monitoring 100,000 MCP endpoints with 60-second probe intervals (~1,667 writes/second), a DynamoDB probe state table will have ~100-200 active partitions = ~100-200 active DynamoDB Streams shards. Lambda ESM handles this well: 100 concurrent Lambda invocations (one per shard) at 5 ms average execution time per batch = negligible compute cost. The KCL adapter becomes compelling if the probe result fanout (writing enriched probe events to multiple downstream services) requires stateful aggregation that doesn't fit in Lambda's 15-minute timeout.

Complete failure mode reference