AWS MSK for MCP Server Probe Pipelines: Event Architecture, IAM Auth, and Operational Reliability

Pattern 1 — The Event Backbone Decision: MSK Deployment Mode, Topic Design, and Producer Configuration

Choosing between MSK Serverless and Provisioned

MSK Serverless removes the broker sizing decision entirely. Capacity autoscales to demand; you pay per throughput consumed rather than per broker-hour. For a new MCP monitoring deployment, Serverless is the correct starting point — no capacity planning needed, and the IAM-only authentication (Serverless does not support SASL/SCRAM or mTLS) simplifies the security model by removing secret rotation entirely.

The limits that matter for probe pipelines: MSK Serverless caps each topic at 120 partitions, and the cluster at 120 Kafka Units (RUs), where one RU roughly equals 1 MB/s write plus 2 MB/s read. At AliveMCP's monitoring cadence — 5,000 MCP endpoints at 60-second intervals — the write rate is approximately 83 probe results per second. A typical probe result JSON payload is under 500 bytes, putting write throughput well under 1 MB/s and well within Serverless limits.

Provisioned MSK with three kafka.m5.large brokers (one per AZ) handles up to ~300 MB/s aggregate throughput and supports all four authentication mechanisms. The fixed broker-hour cost makes it cheaper than Serverless once probe throughput is consistently high enough that per-throughput charges exceed the broker-hour rate — typically around 10,000+ endpoints at 60-second cadence.

Topic design: message key routing for stateful consumers

The choice of Kafka message key determines which partition each probe result lands on, and which consumer receives it. For the alert pipeline — where each consumer tracks the last known status of each server and detects transitions — the key must be the server slug, not the probe worker ID.

Using the probe worker ID as the key is the most common topic design mistake. A probe worker monitoring 200 servers routes all 200 servers' results to the same partition, creating a hot partition while leaving others idle. More critically, the consumer receiving that partition holds state for 200 servers, while other consumers hold none — load is uneven and state is siloed by worker rather than by server.

Using the server slug as the key distributes servers evenly across partitions by hash, and guarantees that all probe results for a given server arrive at the same partition. The consumer processing that partition always sees the full ordered history for each server in its assigned partitions, making the in-memory serverSlug → lastStatus map consistent without shared state or external coordination.

// CDK: MSK Serverless cluster for probe pipeline
import { aws_msk as msk } from 'aws-cdk-lib';

const serverlessCluster = new msk.CfnServerlessCluster(this, 'McpProbeMsk', {
  clusterName: 'mcp-probe-cluster',
  vpcConfigs: [{
    subnetIds: vpc.selectSubnets({ subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS }).subnetIds,
    securityGroups: [mskSg.securityGroupId],
  }],
  clientAuthentication: {
    sasl: { iam: { enabled: true } },
  },
});

// Topics created via AdminClient custom resource Lambda at deploy time
await admin.createTopics({
  topics: [
    {
      topic: 'mcp-probe-results',
      numPartitions: 60,            // 60 partitions = 60 max parallel consumers
      replicationFactor: 3,         // survives loss of 1 AZ
      configEntries: [
        { name: 'retention.ms',        value: String(7 * 24 * 60 * 60 * 1000) }, // 7 days
        { name: 'compression.type',    value: 'lz4' },
        { name: 'min.insync.replicas', value: '2' },    // require 2 of 3 AZ ACKs
      ],
    },
    {
      topic: 'mcp-status-changes',   // filtered: only up↔down transitions
      numPartitions: 10,
      replicationFactor: 3,
      configEntries: [{ name: 'retention.ms', value: String(30 * 24 * 60 * 60 * 1000) }],
    },
  ],
});

Idempotent producer configuration

Probe workers run as long-lived processes and must tolerate transient network failures without creating duplicate probe results. The KafkaJS idempotent: true producer option enables exactly-once delivery: the broker tracks a sequence number per producer session and deduplicates retried sends.

import { Kafka } from 'kafkajs';
import { generateAuthToken } from 'aws-msk-iam-sasl-signer-js';

const kafka = new Kafka({
  clientId: `probe-worker-${process.env.WORKER_ID}`,
  brokers:  process.env.MSK_BOOTSTRAP_BROKERS!.split(','),
  ssl:  true,
  sasl: {
    mechanism: 'oauthbearer',
    oauthBearerProvider: async () => {
      const { token, expiryTime } = await generateAuthToken({ region: process.env.AWS_REGION! });
      return { value: token, extensions: { expiryMs: String(expiryTime) } };
    },
  },
});

const producer = kafka.producer({
  idempotent: true,           // broker deduplicates retried sends via sequence numbers
  transactionTimeout: 30_000,
});

await producer.connect();

// After each probe run
await producer.send({
  topic: 'mcp-probe-results',
  messages: [{
    key:   serverSlug,         // consistent partition routing per server
    value: JSON.stringify({ serverSlug, status, responseTimeMs, checkedAt }),
    headers: { 'content-type': 'application/json' },
  }],
  acks: -1,                   // wait for all in-sync replicas (min.insync.replicas = 2)
});

The acks: -1 setting paired with min.insync.replicas: 2 on the topic provides the durability guarantee: the broker only ACKs the produce when at least 2 AZs have written the message. This means no acknowledged probe result is lost even if a broker in one AZ fails immediately after the ACK.

AliveMCP supplements this with a 60-second canary produce to a dedicated alivemcp-canary topic. A timeout on the canary produce flags the cluster as degraded before any CloudWatch alarm fires — broker leader election stalls, for example, cause produce timeouts well before CloudWatch metrics register. The canary also validates the IAM token refresh path end-to-end, not just the cluster's reported health status.

Pattern 2 — The IAM Security Model: SASL/IAM Mechanism, Least-Privilege Roles, and the AlterGroup Trap

How SASL/IAM authentication works

MSK's IAM authentication uses the Kafka SASL/OAUTHBEARER extension with a custom AWS-signed token format. The client calls generateAuthToken from the aws-msk-iam-sasl-signer-js package, which signs a request using the EC2 instance's IMDS credentials (or Lambda execution role, or ECS task role). The resulting token is sent to the broker during the SASL handshake; the broker validates the signature against IAM and checks the principal's policy for the requested Kafka operation.

Tokens expire in approximately 15 minutes. KafkaJS schedules a refresh before expiry, but only if the oauthBearerProvider function returns the expiryMs extension value:

// Correct: returns expiryMs so KafkaJS can schedule refresh
async function mskIamProvider() {
  const { token, expiryTime } = await generateAuthToken({ region: 'us-east-1' });
  return {
    value: token,
    extensions: { expiryMs: String(expiryTime) },  // required for auto-refresh
  };
}

// Incorrect: no expiryMs — KafkaJS treats token as non-expiring
// First connection succeeds; ~15 minutes later the connection drops
async function mskIamProviderBroken() {
  const { token } = await generateAuthToken({ region: 'us-east-1' });
  return { value: token };  // KafkaJS never refreshes — silent failure after 15 minutes
}

Bootstrap broker ports and the ssl: true requirement

MSK exposes different bootstrap endpoints for each authentication mechanism on different ports. Connecting to the wrong port produces an opaque connection reset, not an authentication error. The port table:

Port 9098 requires TLS — ssl: true in the KafkaJS config. Omitting it on port 9098 produces a Connection reset by peer error that is identical to a port mismatch error. This is the most common connection failure when migrating probe workers from a dev environment (where plaintext may have been used) to production (where IAM auth is enforced).

Least-privilege IAM roles for probe workers and consumers

MSK IAM authorization operates at the Kafka cluster, topic, and consumer group level using kafka-cluster: namespace actions. The resource ARN format includes the cluster UUID — not just the cluster name — which is only available after the cluster is created and must be retrieved from the CfnOutput or SSM Parameter Store at deploy time.

// Probe worker role: produce-only on mcp-probe-results
probeWorkerRole.addToPrincipalPolicy(new iam.PolicyStatement({
  actions: [
    'kafka-cluster:Connect',       // connect to cluster
    'kafka-cluster:WriteData',     // produce to topic
    'kafka-cluster:DescribeTopic', // leader discovery during metadata fetch
  ],
  resources: [
    cluster.clusterArn,
    // UUID required — common mistake: ARN with cluster name only fails silently
    `arn:aws:kafka:${region}:${account}:topic/${clusterName}/${clusterId}/mcp-probe-results`,
  ],
}));

// Alert processor role: consume from both topics
alertProcessorRole.addToPrincipalPolicy(new iam.PolicyStatement({
  actions: [
    'kafka-cluster:Connect',
    'kafka-cluster:ReadData',
    'kafka-cluster:DescribeTopic',
    'kafka-cluster:DescribeGroup',
    'kafka-cluster:AlterGroup',    // REQUIRED: commit offsets; missing = silent duplicate alerts
  ],
  resources: [
    cluster.clusterArn,
    `arn:aws:kafka:${region}:${account}:topic/${clusterName}/${clusterId}/mcp-probe-results`,
    `arn:aws:kafka:${region}:${account}:topic/${clusterName}/${clusterId}/mcp-status-changes`,
    `arn:aws:kafka:${region}:${account}:group/${clusterName}/${clusterId}/alert-processor-group`,
  ],
}));

The kafka-cluster:AlterGroup action is the most commonly missing permission for MSK consumer roles. Without it, the consumer can read messages but cannot commit its consumed offset. The consumer does not crash and does not log an error — it simply never advances its committed position. On every restart, the consumer re-reads from the last committed offset (which never moved), replaying all previously processed messages and firing every alert again. The only observable symptom is duplicate alert delivery after deployments, which often gets attributed to the deployment process rather than a missing IAM action.

SASL/SCRAM for development tooling

Engineers inspecting the probe topic with kcat, Kafka UI, or Conduktor cannot assume an IAM role. SASL/SCRAM provides a user/password credential for these tools. Two MSK-specific requirements: the Secrets Manager secret must be named with the AmazonMSK_ prefix (AmazonMSK_{cluster-name}_{username}), and the secret must be associated with the cluster using the BatchAssociateScramSecret API (available only on provisioned MSK, not Serverless). SCRAM credentials rotate via Secrets Manager with no cluster restart; the old password remains valid for 5 minutes after rotation to allow in-flight connections to drain.

Token refresh in long-running probe workers

Probe workers are long-lived processes — they produce a message every 60 seconds, indefinitely. IAM tokens expire every ~15 minutes. If the probe worker enters a quiet period (all monitored endpoints paused, no produce operations), the token expires and the next produce triggers re-authentication. KafkaJS handles this automatically via retry logic, but the re-auth cycle emits WARN-level log entries that are the only observability signal for this event. Do not suppress KafkaJS WARN logs in probe workers.

A simple mitigation: send a heartbeat message on a 10-minute interval to keep the connection active and prevent token expiry during quiet periods:

// Keep connection alive during quiet periods; prevents token expiry on idle workers
setInterval(async () => {
  await producer.send({
    topic: 'mcp-probe-results',
    messages: [{
      key:   '__heartbeat__',
      value: JSON.stringify({ type: 'heartbeat', workerId: process.env.WORKER_ID, ts: Date.now() }),
    }],
  });
}, 10 * 60 * 1000); // 10 minutes

Pattern 3 — Consumer Scaling and Operational Reliability: Cooperative Rebalancing, Graceful Shutdown, Lag Monitoring, and MSK Connect

CooperativeStickyAssignor for stateful alert processors

The probe alert processor keeps an in-memory serverStatus map: on each probe result, it checks whether the server's status has changed from the cached value, and fires an alert if so. This state lives in the consumer process's heap, associated with the partitions that consumer currently owns. A full Kafka rebalance — the default behavior with RoundRobin or Range assignors — revokes all partitions from all consumers and redistributes them. Every consumer loses its cached state. The next probe result for each server finds no cached state (lastStatus === undefined), so no transition is detected. The first genuine down-event after a rebalance may be missed if the consumer hasn't yet seen a healthy probe result to establish baseline state.

CooperativeStickyAssignor (KafkaJS: PartitionAssigners.cooperative) performs incremental rebalancing: only partitions that need to move for load balance are revoked, not all partitions. When a 6-consumer group scales out to 7, approximately 8–9 partitions move to give the newcomer its share; the other 51–52 partitions stay with their original owners, preserving their in-memory state. For the alert processor, this means a scale event is nearly invisible to the pipeline — no missed transitions, no full state reload from DynamoDB required.

import { PartitionAssigners } from 'kafkajs';

const consumer = kafka.consumer({
  groupId: 'alert-processor-group',
  partitionAssigners: [PartitionAssigners.cooperative],  // incremental rebalance
  sessionTimeout:    45_000,   // broker marks dead after 45s no heartbeat
  heartbeatInterval: 10_000,   // must be < sessionTimeout / 3
});

// In-memory state per partition assignment
const serverStatus = new Map<string, 'up' | 'degraded' | 'down'>();

await consumer.run({
  autoCommit:             true,
  autoCommitInterval:     5_000,   // commit offsets every 5s
  autoCommitThreshold:    100,     // or every 100 messages

  async eachMessage({ message }) {
    const { serverSlug, status, checkedAt, responseTimeMs } = JSON.parse(message.value!.toString());
    const lastStatus = serverStatus.get(serverSlug);

    if (lastStatus !== undefined && lastStatus !== status) {
      await sendAlert({ serverSlug, previousStatus: lastStatus, currentStatus: status, checkedAt });
    }

    serverStatus.set(serverSlug, status);
  },
});

Graceful shutdown — the 45-second problem

ECS sends SIGTERM before stopping a Fargate task. A consumer that exits without calling consumer.disconnect() does not send a LeaveGroup request to the broker. The broker waits for the session timeout — 45 seconds — before declaring the consumer dead and triggering rebalance to redistribute its partitions. During those 45 seconds, the consumer's partitions are unassigned: probe results accumulate without being processed, and any transitions that occur during this window are not detected until the partition is assigned to a new consumer and state is rebuilt.

let isShuttingDown = false;

const gracefulShutdown = async (signal: string) => {
  console.log(`Received ${signal}`);
  isShuttingDown = true;

  await consumer.pause([{ topic: 'mcp-probe-results' }]);   // stop accepting new messages
  await new Promise(resolve => setTimeout(resolve, 10_000)); // drain in-flight processing

  // disconnect() sends LeaveGroup → broker triggers rebalance immediately
  // partitions reassigned to remaining consumers in seconds, not 45 seconds
  await consumer.disconnect();
  process.exit(0);
};

process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
process.on('SIGINT',  () => gracefulShutdown('SIGINT'));

Set the ECS task definition stopTimeout to 60 seconds to ensure the 10-second drain plus disconnect() latency completes before SIGKILL.

Consumer lag monitoring

Consumer lag — unconsumed messages between the consumer's committed offset and the topic's latest offset — is the primary health indicator for the probe alert pipeline. Growing lag means probe results are accumulating faster than alerts are being fired; alert delivery latency is increasing.

// CDK: alarm on any partition lag exceeding 2 minutes of probes
new cloudwatch.Alarm(this, 'ConsumerLagAlarm', {
  metric: new cloudwatch.Metric({
    namespace:  'AWS/Kafka',
    metricName: 'OffsetLag',
    dimensionsMap: {
      ClusterName:   clusterName,
      ConsumerGroup: 'alert-processor-group',
      Topic:         'mcp-probe-results',
    },
    statistic: 'Maximum',    // worst partition, not average — one bad partition is a problem
    period:    cdk.Duration.minutes(1),
  }),
  threshold:         120,    // 120 messages ≈ 2 minutes of probes at 1/min cadence
  evaluationPeriods: 3,
  alarmDescription: 'Alert processor falling behind — probe alert delivery delayed',
});

Five lag patterns correspond to five distinct root causes. A sudden spike across all partitions that then recovers is a normal full rebalance — switch to CooperativeStickyAssignor to reduce frequency. Slow monotonic growth on specific partitions indicates a slow downstream (webhook delivery timeout, DynamoDB throttle) — tune timeouts or scale the consumer. Lag that drops to zero then immediately grows again is a crash loop — fix the crash cause rather than adding consumers. Lag constant but non-zero means consumer throughput is below producer throughput — add consumer replicas up to the partition count. Lag stuck at exactly the same offset for minutes while the consumer reports active reads is a poison message causing infinite retry — identify the record in consumer logs, route to a DLQ manually, or advance the offset.

MSK Connect for S3 archiving without consumer code

The archive consumer — reading every probe result and writing it to S3 in a queryable format — is mechanical work: batch records, encode as Parquet, handle S3 multipart uploads, retry failed writes, partition by timestamp. MSK Connect's Confluent S3 Sink Connector does all of this without consumer code. Deploy the connector plugin (a JAR uploaded to S3) as a CfnCustomPlugin, then deploy the connector with a CfnConnector configuration:

// Key connector configuration properties
{
  'connector.class':   'io.confluent.connect.s3.S3SinkConnector',
  'tasks.max':         '6',
  'topics':            'mcp-probe-results',
  'format.class':      'io.confluent.connect.s3.format.parquet.ParquetFormat',
  'parquet.codec':     'snappy',

  // Partition by probe timestamp, not Kafka ingest time
  // Using ingest time creates partitions that mix probes from different hours
  'timestamp.extractor': 'RecordField',
  'timestamp.field':     'checkedAt',
  'path.format':         "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
  'partitioner.class':   'io.confluent.connect.storage.partitioner.TimeBasedPartitioner',

  'flush.size':          '1000',      // write S3 object after 1000 records
  'rotate.interval.ms':  '3600000',   // or 1 hour, whichever comes first

  // Route failed records to DLQ with error context headers
  'errors.tolerance':    'all',
  'errors.deadletterqueue.topic.name': 'mcp-probe-results-dlq',
  'errors.deadletterqueue.context.headers.enable': 'true',
}

The connector's service execution role requires kafka-cluster:AlterGroup scoped to connect-{connector-name}-* (MSK Connect auto-generates consumer group IDs following this pattern). Without it, the same silent offset-not-committed failure occurs as with a regular consumer, but it manifests as the connector re-reading the full topic from offset 0 after every worker restart — re-archiving every historical probe result, re-writing thousands of S3 objects, and generating massive redundancy costs.

Set a DLQ depth alarm at threshold 1: any failed record pages immediately. Probe records in the S3 Parquet archive are the input to Athena queries for SLA reports. A record lost to the DLQ without investigation creates a silent gap in the archive that produces incorrect SLA percentages for the affected server during the affected window, with no error surfaced to the customer receiving the report.

Multiple consumer groups: independent pipelines from one topic

One of MSK's most useful properties for probe pipelines is that consumer groups are independent. Three groups — alert-processor-group, analytics-group, and the MSK Connect connect-mcp-probe-archive-* group — each maintain their own committed offset and consume the full topic at their own pace. Adding the analytics group does not affect the alert processor's throughput, lag, or partition assignment. Adding a new downstream consumer never requires modifying producer code. The only shared resource across groups is topic retention — keep the topic long enough (7 days in the configuration above) for the slowest consumer group to process its backlog after a restart.

Failure modes reference

How AliveMCP uses MSK for MCP monitoring infrastructure

AliveMCP's probe pipeline is built on MSK. Every 60 seconds, probe workers publish results for each monitored MCP endpoint to mcp-probe-results with the server slug as the message key. The alert processor consumer group runs with CooperativeStickyAssignor, maintaining per-server status state in memory and firing webhooks on up/down transitions within seconds of detection. The S3 sink connector archives all probe results to a Parquet data lake partitioned by probe timestamp — this archive powers the Athena queries that generate SLA PDF reports for Team tier customers, the weekly "State of the MCP Registry" audit that tracks ecosystem-wide health trends, and the historical trend analysis in the public blog posts that drive top-of-funnel traffic.

For Team tier customers who operate their own MSK clusters as part of MCP server infrastructure, AliveMCP surfaces broker health, per-consumer-group lag, and MSK Connect DLQ depth alongside MCP endpoint probe results in a unified dashboard — so the operations team can distinguish "our probe pipeline is degraded" from "the MCP endpoint we're monitoring is degraded" without switching between AWS consoles and the AliveMCP dashboard.

Related guides

• AWS MSK Overview for MCP Servers
• MSK IAM Authentication for MCP Probe Workers
• MSK Consumer Groups for Parallel Probe Processing
• MSK Connect for Probe Archive and State Sync
• How to Monitor an MCP Server
• MCP Server Uptime Monitoring
• AWS MemoryDB for MCP Servers: Durable State, ACL Security, and Cost-Efficient Probe Archives
• AWS SSM Operations for MCP Servers