MSK Connect for MCP Probe Archiving

MSK Connect architecture

MSK Connect runs Kafka Connect workers as fully managed compute inside your VPC. You supply a connector plugin (a JAR file stored in S3) and a connector configuration. MSK Connect provisions workers, manages IAM authentication to the MSK cluster, scales capacity, and restarts failed tasks automatically.

S3 Sink connector for probe archive

The Confluent S3 Sink Connector reads from the mcp-probe-results topic and writes Parquet files to S3, partitioned by date. This builds a durable, queryable archive of all probe results — the source of truth for SLA reports and historical uptime analysis.

Step 1: upload the connector plugin to S3

# Download Confluent S3 Sink Connector from Confluent Hub
# (or use the kafka-connect-s3 open-source variant)
curl -L -o kafka-connect-s3.zip \
  https://api.hub.confluent.io/api/plugins/confluentinc/kafka-connect-s3/versions/10.5.12/archive

# Upload to your plugins S3 bucket
aws s3 cp kafka-connect-s3.zip s3://mcp-probe-connect-plugins/kafka-connect-s3-10.5.12.zip

Step 2: create the custom plugin in MSK Connect

// CDK: create MSK Connect custom plugin
import { aws_kafkaconnect as kafkaconnect } from 'aws-cdk-lib';

const s3SinkPlugin = new kafkaconnect.CfnCustomPlugin(this, 'S3SinkPlugin', {
  contentType: 'ZIP',
  location: {
    s3Location: {
      bucketArn:  pluginsBucket.bucketArn,
      fileKey:    'kafka-connect-s3-10.5.12.zip',
    },
  },
  name: 'kafka-connect-s3-10-5-12',
  description: 'Confluent S3 Sink Connector for probe archive',
});

Step 3: create the connector

const s3SinkConnector = new kafkaconnect.CfnConnector(this, 'ProbArchiveConnector', {
  connectorName: 'mcp-probe-archive',
  kafkaCluster: {
    apacheKafkaCluster: {
      bootstrapServers: cluster.bootstrapBrokersSaslIam,
      vpc: {
        subnets: vpc.selectSubnets({ subnetType: ec2.SubnetType.PRIVATE_WITH_EGRESS }).subnetIds,
        securityGroups: [mskSg.securityGroupId],
      },
    },
  },
  kafkaClusterClientAuthentication: {
    authenticationType: 'IAM',        // MSK Connect uses the connector's service role
  },
  kafkaClusterEncryptionInTransit: {
    encryptionType: 'TLS',
  },
  plugins: [{ customPlugin: { customPluginArn: s3SinkPlugin.attrCustomPluginArn, revision: 1 } }],
  serviceExecutionRoleArn: connectorRole.roleArn,

  // Connector configuration (Kafka Connect properties format)
  connectorConfiguration: {
    'connector.class':          'io.confluent.connect.s3.S3SinkConnector',
    'tasks.max':                '6',          // 6 tasks = one per 10 partitions
    'topics':                   'mcp-probe-results',

    // S3 settings
    's3.region':                'us-east-1',
    's3.bucket.name':           archiveBucket.bucketName,
    's3.part.size':             '67108864',   // 64 MB parts (default)

    // Partitioning: year=2026/month=10/day=22/hour=14/
    'storage.class':            'io.confluent.connect.s3.storage.S3Storage',
    'format.class':             'io.confluent.connect.s3.format.parquet.ParquetFormat',
    'schema.compatibility':     'NONE',        // no schema registry
    'parquet.codec':            'snappy',

    'partitioner.class':        'io.confluent.connect.storage.partitioner.TimeBasedPartitioner',
    'path.format':              "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    'locale':                   'en_US',
    'timezone':                 'UTC',
    'timestamp.extractor':      'RecordField',
    'timestamp.field':          'checkedAt',   // use probe timestamp, not ingest time

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

    // Error handling
    'errors.tolerance':         'all',         // don't fail on bad records
    'errors.deadletterqueue.topic.name': 'mcp-probe-results-dlq',
    'errors.deadletterqueue.context.headers.enable': 'true',
  },

  capacity: {
    autoScaling: {
      minWorkerCount: 1,
      maxWorkerCount: 4,
      scaleInPolicy:  { cpuUtilizationPercentage: 20 },
      scaleOutPolicy: { cpuUtilizationPercentage: 80 },
      mcuCount: 1,   // 1 MCU per worker (1 vCPU + 4 GB)
    },
  },

  logDelivery: {
    workerLogDelivery: {
      cloudWatchLogs: { enabled: true, logGroup: connectLogGroup.logGroupName },
      s3: { enabled: true, bucket: logsBucket.bucketName, prefix: 'msk-connect-logs/' },
    },
  },
});

IAM role for MSK Connect

The connector's service execution role needs permissions to read from MSK and write to S3 (for the S3 sink) or DynamoDB (for the DynamoDB sink). MSK Connect uses IAM-based authentication to the cluster, so the role needs the same Kafka IAM actions as a regular consumer:

const connectorRole = new iam.Role(this, 'MskConnectRole', {
  assumedBy: new iam.ServicePrincipal('kafkaconnect.amazonaws.com'),
});

// MSK Consumer permissions (same as alert processor consumer role)
connectorRole.addToPrincipalPolicy(new iam.PolicyStatement({
  actions: [
    'kafka-cluster:Connect',
    'kafka-cluster:ReadData',
    'kafka-cluster:DescribeTopic',
    'kafka-cluster:DescribeGroup',
    'kafka-cluster:AlterGroup',
    'kafka-cluster:WriteData',        // needed for DLQ topic
  ],
  resources: [
    cluster.clusterArn,
    `arn:aws:kafka:${region}:${account}:topic/${clusterName}/${clusterId}/mcp-probe-results`,
    `arn:aws:kafka:${region}:${account}:topic/${clusterName}/${clusterId}/mcp-probe-results-dlq`,
    `arn:aws:kafka:${region}:${account}:group/${clusterName}/${clusterId}/connect-mcp-probe-archive-*`,
  ],
}));

// S3 sink permissions
archiveBucket.grantReadWrite(connectorRole);

// Plugin JAR download
pluginsBucket.grantRead(connectorRole);

// CloudWatch for connector logs
connectorLogGroup.grantWrite(connectorRole);

Dead-letter queue handling for failed records

The S3 sink connector configuration above sets errors.tolerance: all and errors.deadletterqueue.topic.name: mcp-probe-results-dlq. Any probe record that fails to serialize or write to S3 is routed to the DLQ topic with error context headers.

The DLQ topic is a standard Kafka topic consumed by a diagnostic Lambda or a separate consumer:

// DLQ consumer: inspect failed records and alert on unexpected failures
const dlqConsumer = kafka.consumer({ groupId: 'connect-dlq-inspector' });
await dlqConsumer.subscribe({ topics: ['mcp-probe-results-dlq'] });

await dlqConsumer.run({
  async eachMessage({ message }) {
    // MSK Connect DLQ headers contain error context
    const errorClass   = message.headers?.['__connect.errors.exception.class.name']?.toString();
    const errorMessage = message.headers?.['__connect.errors.exception.message']?.toString();
    const sourceOffset = message.headers?.['__connect.errors.kafka.topic']?.toString();

    console.error('Connector DLQ record', { errorClass, errorMessage, sourceOffset });

    // Alert on error types that indicate config problems (not transient S3 errors)
    if (errorClass?.includes('SerializationException') || errorClass?.includes('AccessDeniedException')) {
      await sendAlert({ type: 'CONNECTOR_CONFIG_ERROR', errorClass, errorMessage });
    }
  },
});

Common MSK Connect failure modes

Connector throughput and cost sizing

MSK Connect pricing is per MCU-hour (1 MCU = 1 vCPU + 4 GB RAM ≈ $0.11/hour). For a probe pipeline writing 83 records/second to S3:

• 1 worker × 1 MCU handles up to ~500 records/second for a simple S3 sink (mostly I/O bound)
• At 83 records/second, 1 worker with auto-scaling min=1, max=2 is sufficient and costs ~$80/month
• Enable auto-scaling (CPU threshold 80%) so the connector scales out during registry crawl bursts
• The tasks.max setting limits parallelism within a worker — set to the number of topic partitions for maximum throughput (60 for the 60-partition probe topic, but 6 is usually sufficient at this scale)

Compare to the alternative: a KafkaJS consumer in ECS writing to S3 directly would need a custom batching and Parquet encoding implementation. MSK Connect's S3 sink handles batching, Parquet encoding, S3 multipart upload, and retry logic out of the box.

How AliveMCP uses MSK Connect for probe archive

AliveMCP runs an S3 sink connector that archives all probe results to a Parquet data lake partitioned by day. The archive powers:

• SLA PDF reports: Athena queries over the Parquet archive compute 99.9th-percentile uptime and response time for Team tier customers' monthly SLA reports
• State of the MCP Registry audit: the weekly "registry health" report queries 7 days of probe archive to compute per-registry uptime rates, identifying which registries have the most consistently dead endpoints
• Trend analysis: year-on-year queries (possible because the archive is append-only and partitioned by date) power the "MCP ecosystem health" blog posts that drive top-of-funnel traffic

AliveMCP monitors the S3 sink connector as part of its own internal infrastructure health. Connector lag and DLQ depth are surfaced on the internal SRE dashboard. A connector DLQ depth exceeding 0 pages on-call immediately — lost probe records are unrecoverable.

Related guides

• AWS MSK Overview for MCP Servers
• MSK IAM Authentication for MCP Probe Workers
• MSK Consumer Groups for Parallel Probe Processing
• Kafka MCP Server Tools — KafkaJS Producer and Consumer