Guide · MSK Consumer Groups · Parallelism

MSK Consumer Groups for MCP Probe Workers

Kafka consumer groups distribute partition processing across multiple workers, enabling horizontal scaling of MCP probe result processing. For AliveMCP's alert pipeline — where each consumer must maintain per-server state to detect up/down transitions — partition assignment strategy, rebalance behavior, and consumer lag directly affect alert delivery latency. This guide covers the KafkaJS consumer group configuration, partition assignment strategies, cooperative rebalancing, consumer lag monitoring, and graceful shutdown for ECS-hosted probe workers.

Consumer group fundamentals for MCP probe pipelines

A consumer group is a set of Kafka consumers that collectively consume a topic. Each partition is assigned to exactly one consumer at a time. When consumers join or leave the group, Kafka re-assigns partitions — a process called rebalancing.

For the MCP probe result pipeline: the mcp-probe-results topic has 60 partitions. With 6 alert-processor consumers in one group, each consumer owns 10 partitions — 10 partitions × ~33 servers/partition = ~330 servers per consumer. The consumer tracks each server's last known status in memory keyed by server slug. When a new probe result arrives with a different status, the consumer triggers an alert.

This stateful pattern requires that all probes for a given server always route to the same consumer. Since Kafka routes by partition key (the server slug hash), and partition assignment within a group gives each consumer a stable set of partitions, this holds as long as no rebalance occurs. During a rebalance, partitions move between consumers — requiring the new owner to reload state from the DynamoDB source of truth before processing new messages.

Multiple consumer groups: independent pipelines from one topic

// Three consumer groups, each consuming all 60 partitions independently:
// alert-processor-group — stateful up/down transition detection → webhook delivery
// archive-writer-group  — all probe results → S3 + MSK Connect sink (see MSK Connect guide)
// analytics-group       — aggregate metrics computation (p50/p95 latency, registry health score)

// Each group maintains its own committed offset — progress is independent
// Adding a third group does NOT affect the alert processor's throughput or lag

Consumer groups are free to create (no cost, no capacity allocation). Add a new consumer group to add a new pipeline without affecting existing groups. The only shared resource is topic retention: keep topics long enough for the slowest consumer group to process at its own pace.

KafkaJS consumer group configuration

import { Kafka, logLevel } from 'kafkajs';

const consumer = kafka.consumer({
  groupId: 'alert-processor-group',
  sessionTimeout:       45_000,  // broker declares consumer dead after 45s of no heartbeat
  heartbeatInterval:    10_000,  // send heartbeat every 10s (must be < sessionTimeout / 3)
  maxWaitTimeInMs:      5_000,   // max time to wait for batch to accumulate before returning
  minBytes:             1,       // min bytes to fetch before returning (1 = return immediately)
  maxBytes:             10_485_760, // max 10MB per fetch request
  partitionAssigners: [PartitionAssigners.roundRobin], // see assignor comparison below
});

await consumer.connect();
await consumer.subscribe({ topics: ['mcp-probe-results'], fromBeginning: false });

// In-memory state: last known status per server slug
const serverStatus = new Map<string, string>();

await consumer.run({
  autoCommit:              true,
  autoCommitInterval:      5_000,   // commit offsets every 5 seconds
  autoCommitThreshold:     100,     // or every 100 messages (whichever comes first)
  eachBatchAutoResolve:    true,

  async eachMessage({ topic, partition, message }) {
    const payload = JSON.parse(message.value!.toString());
    const { serverSlug, status, checkedAt, responseTimeMs } = payload;

    const lastStatus = serverStatus.get(serverSlug);

    if (lastStatus !== undefined && lastStatus !== status) {
      // Status transition — fire alert
      await sendAlert({
        type:           status === 'down' ? 'DOWN' : status === 'up' ? 'RECOVERED' : 'DEGRADED',
        serverSlug,
        previousStatus: lastStatus,
        currentStatus:  status,
        checkedAt,
        responseTimeMs,
      });
    }

    serverStatus.set(serverSlug, status);  // update in-memory state
  },
});

Session timeout and heartbeat interval relationship

The Kafka broker declares a consumer dead when it hasn't received a heartbeat within sessionTimeout. KafkaJS sends heartbeats on a background timer at heartbeatInterval. The rule: heartbeatInterval < sessionTimeout / 3. With sessionTimeout: 45_000 and heartbeatInterval: 10_000, the broker allows 4+ missed heartbeats before triggering a rebalance — tolerating 30+ seconds of slow message processing before declaring the consumer dead.

Increase sessionTimeout if your alert processing (webhook delivery, DynamoDB writes) sometimes takes more than 10 seconds. Do not decrease heartbeatInterval below 3,000ms — too-frequent heartbeats increase broker load without improving failure detection speed.

Partition assignment strategies

KafkaJS supports multiple partition assignors that determine how partitions are distributed when consumers join or leave:

Assignor	Behavior	Rebalance impact	Best for
RoundRobin	Partitions distributed evenly round-robin across consumers	Full rebalance — all partitions re-assigned	Stateless consumers where assignment history doesn't matter
Range	Partitions 0–N/C assigned to consumer 0, N/C+1–2N/C to consumer 1, etc.	Full rebalance — all re-assigned	Default KafkaJS; simple but uneven for unequal partition counts
CooperativeStickyAssignor	Minimizes partition movement; only revokes partitions from over-assigned consumers	Incremental — only affected partitions move	Stateful consumers that maintain per-partition state (probe pipeline)

For the stateful probe alert processor, CooperativeStickyAssignor is strongly preferred. When a new consumer joins (scale-out), only the partitions that need to move for load balance are revoked — not all 60 partitions. Each consumer keeps its partitions (and in-memory state) unless Kafka explicitly revokes them. This eliminates the full state reload on every scale event.

import { PartitionAssigners } from 'kafkajs';

const consumer = kafka.consumer({
  groupId: 'alert-processor-group',
  partitionAssigners: [PartitionAssigners.cooperative],  // CooperativeStickyAssignor
  // ...
});

State reload on partition assignment

Even with cooperative rebalancing, new partitions will occasionally be assigned (scale-out, new deployment). When a new partition arrives, the consumer's in-memory serverStatus map doesn't have entries for those servers yet. The first probe result for each server will have lastStatus === undefined, so no alert fires — correct behavior, since we don't know the previous state. Subsequent probes correctly detect transitions.

For faster state recovery, preload from DynamoDB on partition assignment:

consumer.on(consumer.events.GROUP_JOIN, async ({ payload }) => {
  // Payload includes memberAssignment — the partitions this consumer now owns
  // Query DynamoDB StatusIndex for servers assigned to these partitions
  // (requires knowing which server slugs hash to which partitions)
  // Simpler: on assignment, query ALL servers with status='down' and seed serverStatus
  const downServers = await queryDynamoStatusIndex('down');
  for (const server of downServers) {
    serverStatus.set(server.serverSlug, 'down');
  }
});

Consumer lag monitoring with CloudWatch

Consumer lag — the number of unconsumed messages between the consumer's committed offset and the topic's latest offset — is the primary health metric for the probe pipeline. A growing lag means probes are completing faster than alerts are firing; alert latency is increasing.

MSK publishes consumer lag to CloudWatch when enhanced monitoring is set to PER_TOPIC_PER_PARTITION:

// CloudWatch metric: kafka.consumer_group.offset_lag
// Dimensions: ClusterName, ConsumerGroup, Topic, Partition

// CDK alarm: alert if any partition lag exceeds 2 minutes of probes
new cloudwatch.Alarm(this, 'ConsumerLagAlarm', {
  alarmName: 'mcp-probe-consumer-lag',
  metric: new cloudwatch.Metric({
    namespace:  'AWS/Kafka',
    metricName: 'OffsetLag',
    dimensionsMap: {
      ClusterName:   clusterName,
      ConsumerGroup: 'alert-processor-group',
      Topic:         'mcp-probe-results',
    },
    statistic: 'Maximum',       // Maximum across all partitions — any lagging partition triggers
    period:    cdk.Duration.minutes(1),
  }),
  threshold:          120,      // 120 unconsumed messages ≈ 2 minutes of probes at 1/min cadence
  evaluationPeriods:  3,        // 3 consecutive minutes before alarm
  comparisonOperator: cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
  alarmDescription:  'Alert processor falling behind — probe alert delivery delayed',
});

Lag spike patterns and their causes

Lag pattern	Likely cause	Remediation
Sudden spike on all partitions, then recovers	Rebalance — all partitions paused during rebalance	Normal with full rebalance; switch to CooperativeStickyAssignor to reduce
Slow monotonic growth on specific partitions	One consumer is slow (webhook delivery timeout, DynamoDB throttle)	Tune downstream timeouts; add consumer concurrency; check DynamoDB on-demand capacity
Lag drops to zero then immediately grows again	Consumer restarting in crash loop; re-reading from committed offset each time	Fix crash cause; check ECS service events for OOM or unhandled promise rejections
Lag constant but non-zero	Producer throughput exceeds consumer throughput; consumer count insufficient	Add consumer replicas (up to partition count max)
Lag stuck at exact offset, consumer appears active	Poison message — consumer processing it, failing, retrying infinitely	Check consumer logs for repeated error; route to DLQ or skip via offset reset

Graceful shutdown for ECS probe workers

ECS sends SIGTERM before stopping a task. A consumer that exits without calling consumer.disconnect() leaves its partitions unassigned until the session timeout expires (45 seconds). During that window, those partitions are not being processed — probe results accumulate. Graceful shutdown reduces this window to near-zero:

// Graceful SIGTERM handler for ECS tasks
const gracefulShutdown = async (signal: string) => {
  console.log(`Received ${signal} — starting graceful shutdown`);
  isShuttingDown = true;

  // Stop accepting new messages
  await consumer.pause([{ topic: 'mcp-probe-results' }]);

  // Wait for in-flight message processing to complete (max 10s)
  await new Promise(resolve => setTimeout(resolve, 10_000));

  // Commit current offsets and leave consumer group
  // Kafka triggers immediate rebalance to reassign partitions to remaining consumers
  await consumer.disconnect();

  console.log('Consumer disconnected cleanly');
  process.exit(0);
};

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

The consumer.disconnect() call triggers a LeaveGroup request to the broker, which immediately initiates a rebalance to reassign partitions to remaining consumers. Without this, the broker waits for the session timeout (45 seconds) before declaring the consumer dead and triggering rebalance — 45 seconds of unprocessed probes.

How AliveMCP monitors consumer group health

AliveMCP tracks consumer group health as part of the MSK probe pipeline monitoring. In addition to the lag alarm described above, AliveMCP monitors:

Rebalance frequency: frequent rebalances (more than 2 per hour) indicate unstable consumers — crash loops, GC pauses exceeding session timeout, or rolling deployments without cooperative rebalancing. Frequent rebalances pause processing during each rebalance window.
Consumer count: if the consumer group drops below expected replica count (e.g., 3 replicas but CloudWatch shows 2 active consumers), AliveMCP alerts before lag grows — the missing consumer's partitions are covered by rebalancing but at reduced redundancy.
Offset commit rate: if offset commits stop (consumer still consuming but AlterGroup IAM permission revoked), consumer lag metrics become stale and the consumer re-reads from the last committed offset on every restart. AliveMCP detects this via lag that matches the re-read distance rather than the accumulation rate.