Kinesis Enhanced Fan-Out for MCP Servers — RegisterStreamConsumer, SubscribeToShard, and 2 MiB/s Per-Consumer Bandwidth

Enhanced Fan-Out vs standard polling: throughput model

The fundamental difference between standard GetRecords polling and Enhanced Fan-Out is who bears the read-throughput budget:

For an AliveMCP probe result stream with 4 downstream consumers running simultaneously, standard polling gives each consumer at most 500 KiB/s — a 75% reduction from the shard's write bandwidth. With EFO, each consumer receives a dedicated 2 MiB/s push stream. If the probe result stream produces 1 MiB/s of incoming data, every EFO consumer receives that full 1 MiB/s independently; backpressure on one consumer (e.g., the archiver is slow) does not starve the alerter.

RegisterStreamConsumer: lifecycle and status transitions

EFO requires pre-registering each consumer application by name before it can call SubscribeToShard. The registration is idempotent by consumer name — calling it twice with the same name returns the existing consumer ARN rather than creating a duplicate.

import { KinesisClient, RegisterStreamConsumerCommand,
         DescribeStreamConsumerCommand } from '@aws-sdk/client-kinesis';

const client = new KinesisClient({ region: 'us-east-1' });

async function registerConsumer(streamArn: string, consumerName: string): Promise {
  const reg = await client.send(new RegisterStreamConsumerCommand({
    StreamARN: streamArn,
    ConsumerName: consumerName,
  }));

  const consumerArn = reg.Consumer!.ConsumerARN!;

  // Registration is async — poll until ACTIVE before calling SubscribeToShard
  for (let i = 0; i < 30; i++) {
    const desc = await client.send(new DescribeStreamConsumerCommand({
      ConsumerARN: consumerArn,
    }));
    if (desc.ConsumerDescription!.ConsumerStatus === 'ACTIVE') {
      return consumerArn;
    }
    await new Promise(r => setTimeout(r, 2000));
  }
  throw new Error(`Consumer ${consumerName} did not reach ACTIVE within 60s`);
}

Consumer status transitions: CREATING → ACTIVE typically takes 10–30 seconds. A consumer in CREATING state cannot receive records — SubscribeToShard returns ResourceInUseException if called before the consumer reaches ACTIVE. On deregistration: ACTIVE → DELETING → removed. DELETING consumers still count against the 20-consumer limit; wait for full removal before re-registering the same name if needed within minutes.

Consumer naming strategy for MCP deployments

Consumer names are scoped per stream, not globally. Use a deterministic, deployment-stable name — do not include commit SHAs or timestamps, which create orphaned DELETING registrations that block the 20-consumer limit:

# Good: stable, human-readable, scoped to function role
alivemcp-alerter-consumer
alivemcp-archiver-consumer
alivemcp-status-page-consumer
alivemcp-analytics-consumer

# Bad: includes deployment metadata — creates stale consumer ARNs on each deploy
alivemcp-alerter-consumer-abc123
alivemcp-alerter-consumer-2026-10-30

SubscribeToShard: HTTP/2 push delivery

SubscribeToShard opens a long-lived HTTP/2 server-sent event stream from Kinesis to your consumer. Each shard subscription lasts up to 5 minutes, after which Kinesis closes the stream with a SubscriptionEventStream close event. Your consumer must re-call SubscribeToShard with the continuation sequence number to resume:

import { SubscribeToShardCommand } from '@aws-sdk/client-kinesis';

async function consumeShard(
  client: KinesisClient,
  consumerArn: string,
  shardId: string,
  startingSequenceNumber?: string
): Promise {
  const startingPosition = startingSequenceNumber
    ? { Type: 'AFTER_SEQUENCE_NUMBER' as const, SequenceNumber: startingSequenceNumber }
    : { Type: 'LATEST' as const };

  const response = await client.send(new SubscribeToShardCommand({
    ConsumerARN: consumerArn,
    ShardId: shardId,
    StartingPosition: startingPosition,
  }));

  let lastSequenceNumber: string | undefined;

  for await (const event of response.EventStream!) {
    if (event.SubscribeToShardEvent) {
      const records = event.SubscribeToShardEvent.Records ?? [];
      for (const record of records) {
        await processProbeResult(record.Data);
        lastSequenceNumber = record.SequenceNumber;
      }
      // Checkpoint progress; re-subscribe uses lastSequenceNumber
    }
  }

  // Stream closed after ~5 minutes — re-subscribe with continuation
  if (lastSequenceNumber) {
    await consumeShard(client, consumerArn, shardId, lastSequenceNumber);
  }
}

EFO records arrive with ~70ms propagation latency compared to ~200ms for GetRecords polling. For AliveMCP's real-time alerter, this means a probe failure event reaches the alerter roughly 130ms faster — often the difference between a 60-second check interval appearing as <1 minute or exactly 1 minute of downtime on the status dashboard.

Concurrent shard subscriptions

Each consumer application must maintain one SubscribeToShard connection per shard. For a 4-shard stream with 3 EFO consumers, that is 12 concurrent HTTP/2 connections. AWS Lambda handles this automatically when using the KDS event source mapping with EFO mode enabled — you do not manage shard subscriptions directly. For ECS or EC2 consumers, use the Kinesis Client Library (KCL) 2.x which manages shard assignment and EFO subscriptions per worker instance.

Consumer limits and quota management

The 20-consumer limit is the most common operational constraint. Stale consumers in DELETING state consume slots. In CI/CD pipelines that redeploy consumers frequently, add a pre-deploy step that calls ListStreamConsumers and deregisters any consumers whose names match the current deployment's consumer names that are in ACTIVE or DELETING status before re-registering:

import { ListStreamConsumersCommand, DeregisterStreamConsumerCommand } from '@aws-sdk/client-kinesis';

async function deregisterIfExists(client: KinesisClient, streamArn: string, consumerName: string) {
  const list = await client.send(new ListStreamConsumersCommand({ StreamARN: streamArn }));
  const existing = list.Consumers?.find(c => c.ConsumerName === consumerName);
  if (existing) {
    await client.send(new DeregisterStreamConsumerCommand({
      ConsumerARN: existing.ConsumerARN,
    }));
    console.log(`Deregistered existing consumer ${consumerName}`);
  }
}

IAM permissions for Enhanced Fan-Out

EFO requires separate permissions from standard GetRecords polling. The consumer application's IAM role needs:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:RegisterStreamConsumer",
        "kinesis:DescribeStreamConsumer",
        "kinesis:SubscribeToShard",
        "kinesis:ListStreamConsumers"
      ],
      "Resource": [
        "arn:aws:kinesis:us-east-1:123456789012:stream/alivemcp-probe-results",
        "arn:aws:kinesis:us-east-1:123456789012:stream/alivemcp-probe-results/consumer/*:*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "kinesis:DeregisterStreamConsumer",
      "Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/alivemcp-probe-results/consumer/*:*"
    }
  ]
}

Note the dual resource block: RegisterStreamConsumer acts on the stream ARN, while SubscribeToShard and DeregisterStreamConsumer act on the consumer ARN (a child resource of the stream). If you grant only the stream ARN, SubscribeToShard calls fail with AccessDeniedException.

CloudWatch metrics for EFO consumers

EFO consumers do not contribute to ReadProvisionedThroughputExceeded because they use a dedicated bandwidth pool. If you see this metric spike, it indicates standard GetRecords consumers (not EFO) are being throttled — investigate which application is still using standard polling instead of EFO.

Common failure modes