Managing Amazon Msk

Data Delivery for General Purpose S3 Buckets

Data Delivery for General Purpose S3 Buckets for Amazon MSK Express brokers delivers topic data from Express brokers to a general-purpose S3 bucket as objects, with configurable compression, storage class, and output key layout.

Serverless with no connectors to manage, minutes-level freshness, and auto-scaling throughput (see Amazon MSK Data Delivery quotas for the per-channel ceiling). Each record is delivered exactly once by the delivery pipeline. Data Delivery does not consume broker egress throughput or impact producer or consumer workloads. This optimizes cost as it enables delivery to S3 without requiring additional cluster capacity on MSK Express clusters. Additionally, you can fan out multiple delivery channels from the same topic.

Note: Records are batched into S3 objects — multiple Kafka records land in a single S3 object. The output key template determines the key for each object, and the !{sequence-number} token provides uniqueness across objects within the same prefix.

Data Delivery for General Purpose S3 Buckets Constraints

Check Data Delivery for General Purpose S3 Buckets documentation for constraints. Some key constraints are:

• Data Delivery for General Purpose S3 Buckets is ONLY available on Express brokers — Standard brokers and MSK Serverless are NOT supported

• Data freshness is bounded and not adjustable — get the current bounds and default from Amazon MSK Data Delivery quotas. The documented minimum-throughput floor for the tightest freshness setting applies to the S3 Tables (Iceberg) destination, not to General Purpose S3

• No backfill — only records produced AFTER Data Delivery for General Purpose S3 Buckets is enabled are delivered

If the user needs any of the above functionality, recommend Managed Service for Apache Flink instead. Data Delivery for General Purpose S3 Buckets is the most cost-effective way to deliver data to General Purpose S3 Buckets, so if it can be used it should be preferred in general.

Broker-type routing (Express vs Standard/Serverless). Data Delivery and Streaming Tables are Express-only. On Standard or Serverless clusters they are not available — to land topic data in S3 there, use one of: the Amazon Data Firehose MSK-source integration (fully managed; delivers to Amazon S3 or to Apache Iceberg tables in self-managed S3 or S3 Tables), a self-managed Kafka Connect S3 sink connector, or Managed Service for Apache Flink. On Express, prefer Data Delivery / Streaming Tables over all three — it is fully managed, consumes no broker egress, and is the lowest-cost option.

Prerequisites

• MSK Provisioned cluster with Express brokers

• Kafka topic producing data (JSON, ByteArray, or String)

• General-purpose S3 bucket for delivery

• S3 bucket for dead-letter queue (DLQ) — required

Input Formats

No schema registry is required for S3 bucket delivery.

Destination Options

Supported values for compressionType and storageClass may change over time. Run aws kafka create-channel help or check the Amazon MSK Data Delivery documentation for the latest accepted values.

IAM Setup

Trust Policy

{
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": {"Service": "kafka.amazonaws.com"},
        "Action": "sts:AssumeRole",
        "Condition": {
            "StringEquals": {"aws:SourceAccount": "ACCOUNT_ID"},
            "ArnLike": {"aws:SourceArn": "arn:aws:kafka:REGION:ACCOUNT_ID:channel/*"}
        }
    }]
}

Permission Policy

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "DeliveryBucketList",
            "Effect": "Allow",
            "Action": ["s3:ListBucket", "s3:ListBucketMultipartUploads", "s3:GetBucketLocation"],
            "Resource": ["arn:aws:s3:::BUCKET", "arn:aws:s3:::BUCKET/*"]
        },
        {
            "Sid": "DeliveryBucketWrite",
            "Effect": "Allow",
            "Action": [
                "s3:UploadPart", "s3:CompleteMultipartUpload", "s3:CreateMultipartUpload",
                "s3:PutObject", "s3:ListMultipartUploads", "s3:ListMultipartUploadParts"
            ],
            "Resource": "arn:aws:s3:::BUCKET/PREFIX*"
        },
        {
            "Sid": "DLQBucketAccess",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject", "s3:GetBucketLocation", "s3:ListBucket",
                "s3:ListBucketMultipartUploads", "s3:UploadPart",
                "s3:CompleteMultipartUpload", "s3:CreateMultipartUpload",
                "s3:ListMultipartUploadParts"
            ],
            "Resource": ["arn:aws:s3:::DLQ_BUCKET", "arn:aws:s3:::DLQ_BUCKET/*"]
        },
        {
            "Sid": "KMSAccess",
            "Effect": "Allow",
            "Action": ["kms:Decrypt", "kms:GenerateDataKey"],
            "Resource": "arn:aws:kms:REGION:ACCOUNT_ID:key/KEY_ID",
            "Condition": {
                "StringEquals": {"kms:ViaService": "s3.REGION.amazonaws.com"},
                "StringLike": {"kms:EncryptionContext:aws:s3:arn": "arn:aws:s3:::BUCKET/PREFIX*"}
            }
        },
        {
            "Sid": "KMSAccessDLQ",
            "Effect": "Allow",
            "Action": ["kms:Decrypt", "kms:GenerateDataKey"],
            "Resource": "arn:aws:kms:REGION:ACCOUNT_ID:key/DLQ_KEY_ID",
            "Condition": {
                "StringEquals": {"kms:ViaService": "s3.REGION.amazonaws.com"},
                "StringLike": {"kms:EncryptionContext:aws:s3:arn": "arn:aws:s3:::DLQ_BUCKET/*"}
            }
        },
        {
            "Sid": "CloudWatchLogsAccess",
            "Effect": "Allow",
            "Action": ["logs:CreateLogStream", "logs:PutLogEvents"],
            "Resource": "arn:aws:logs:REGION:ACCOUNT_ID:log-group:/aws/msk/data-delivery/CHANNEL_NAME:*"
        }
    ]
}

CloudWatchLogsAccess is required whenever CloudWatch Logs delivery logging is enabled (the default) — without it the service role cannot write delivery logs and failures stay invisible. If you route delivery logs to S3 or Firehose instead, swap this for the corresponding s3:PutObject (log bucket) or firehose:PutRecord/firehose:PutRecordBatch permissions per the Delivery Logging section.

KMSAccess covers the delivery bucket (its kms:EncryptionContext condition is scoped to BUCKET/PREFIX*). KMSAccessDLQ is required when the DLQ bucket uses SSE-KMS — the delivery-bucket condition does not cover the DLQ, so without this statement the service role gets AccessDenied writing failed records to an SSE-KMS DLQ. Both KMS statements are recommended for production — omit only if the corresponding bucket relies on SSE-S3 (AES-256) default encryption. SSE-KMS (or at minimum SSE-S3) should always be enabled on both the delivery and DLQ buckets.

Create the Channel

Prerequisite: create the CloudWatch log group first. The log group referenced by LoggingInfo.CloudWatchLogs.LogGroup must exist before create-channel is called. MSK will not auto-create it, and create-channel will fail if the group is missing.

# 1. Create the log group up front
aws logs create-log-group --log-group-name "/aws/msk/data-delivery/CHANNEL_NAME"

# 2. Create the channel
aws kafka create-channel \
    --channel-name "CHANNEL_NAME" \
    --cluster-arn "arn:aws:kafka:REGION:ACCOUNT_ID:cluster/CLUSTER/ID" \
    --topic-configuration-list '[{
        "TopicArn": "arn:aws:kafka:REGION:ACCOUNT_ID:topic/CLUSTER/ID/TOPIC",
        "RecordConverter": {"ValueConverter": "JSON"}
    }]' \
    --s3-destination-configuration '{
        "DataFreshnessInSeconds": 300,
        "DeadLetterQueueS3": {
            "BucketArn": "arn:aws:s3:::DLQ_BUCKET",
            "ErrorOutputPrefix": "dlq/"
        },
        "ServiceExecutionRoleArn": "arn:aws:iam::ACCOUNT_ID:role/ROLE",
        "Storage": {
            "BucketArn": "arn:aws:s3:::DELIVERY_BUCKET",
            "CompressionType": "GZIP",
            "OutputPrefix": "data/",
            "OutputKeyTemplate": "!{channel-id}/!{topic-name}/!{yyyy}/!{MM}/!{dd}/!{HH}/!{partition-id}+!{sequence-number}",
            "StorageClass": "STANDARD"
        }
    }' \
    --logging-info '{
        "CloudWatchLogs": {"Enabled": true, "LogGroup": "/aws/msk/data-delivery/CHANNEL_NAME"},
        "S3": {"Enabled": true, "Bucket": "LOG_BUCKET", "Prefix": "msk-channel-logs/"},
        "Firehose": {"Enabled": false, "DeliveryStream": ""}
    }'

For ByteArray or String input, change RecordConverter.ValueConverter accordingly (BYTE_ARRAY / STRING).

Delivery Logging

Always enable at least one logging destination on the channel. Without it, delivery errors (AccessDenied on the delivery bucket / DLQ, format-mismatch failures, KMS Decrypt failures on the delivery-bucket key, template-rendering errors) are invisible — the only visible symptom is missing objects at the expected S3 key. Logging is also a prerequisite for streaming-tables-troubleshooting.md.

The --logging-info field on aws kafka create-channel accepts three independent destinations — any subset can be enabled at once:

Log group prerequisite: MSK does not auto-create the CloudWatch log group. Create it in advance with aws logs create-log-group --log-group-name "<name>" and, for production, set a retention policy (aws logs put-retention-policy --log-group-name "<name>" --retention-in-days 30). Passing a nonexistent log group to create-channel returns a validation error.

IAM: The service role must be able to write to whichever destinations are enabled — for CloudWatch Logs: logs:CreateLogStream, logs:PutLogEvents on the target log group; for S3: s3:PutObject on the log bucket/prefix (plus kms:GenerateDataKey if SSE-KMS); for Firehose: firehose:PutRecord, firehose:PutRecordBatch on the delivery stream. Use a separate log bucket/prefix from the delivery bucket — routing logs through the same prefix that stores delivered records makes downstream analytics unreliable.

Encryption: Enable SSE-KMS on the CloudWatch log group (via aws logs associate-kms-key) and the S3 log bucket — delivery error records may contain payload fragments and topic names.

Logging is set at channel creation and is not updatable. aws kafka update-channel only accepts --iceberg-destination-update and --s3-destination-update with a single field (DataFreshnessInSeconds). To change logging destinations, delete and recreate the channel.

Output Key Template

Variables

Rules

• Must contain exactly ONE uniqueness token: !{sequence-number} OR !{kafka-offset} (mutually exclusive)

• Uniqueness token must be in the last /-separated segment

• If using !{kafka-offset}, must also include !{partition-id}

• Must NOT end with /

• Max 1024 chars (prefix + template combined)

• Allowed literals: [a-zA-Z0-9/_\-.+=] (no spaces)

• No path traversal (.. or ./)

Valid examples

!{channel-id}/!{yyyy}/!{MM}/!{dd}/!{sequence-number}
topic=!{topic-name}/!{partition-id}/!{kafka-offset}
data/!{channel-id}/!{HH}/!{sequence-number}.json

Invalid examples

Monitoring

Data Delivery emits its own CloudWatch metrics in the AWS/Kafka namespace with dimensions ClusterName, ChannelName, Topic. Recommended alarms — DeliveryToS3.DataFreshness above the configured freshness interval, FailedRecordCount > 0 (records not conforming to input format), and DLQDeliverySuccess > 0 (records routed to DLQ). Full metric list, alarm patterns, and troubleshooting steps live in streaming-tables-troubleshooting.md. Do not fall back to generic Express cluster health metrics or S3-bucket-object counts for delivery monitoring — the channel-scoped metrics above are the correct primitives.

Channel Management

Every channel operation is scoped to a specific cluster — --cluster-arn is required on describe-channel, list-channels, update-channel, and delete-channel.

# Check state (both flags required)
aws kafka describe-channel \
    --cluster-arn "arn:aws:kafka:REGION:ACCOUNT_ID:cluster/CLUSTER/ID" \
    --channel-arn "CHANNEL_ARN"

# List channels on a cluster. Only --topic-name-filter is available server-side —
# filter by destination type client-side from the returned DestinationType field.
aws kafka list-channels \
    --cluster-arn "arn:aws:kafka:REGION:ACCOUNT_ID:cluster/CLUSTER/ID" \
    --query "Channels[?DestinationType=='S3']"

# Update freshness — only DataFreshnessInSeconds is updatable
aws kafka update-channel \
    --cluster-arn "arn:aws:kafka:REGION:ACCOUNT_ID:cluster/CLUSTER/ID" \
    --channel-arn "CHANNEL_ARN" \
    --s3-destination-update '{"DataFreshnessInSeconds": 600}'

# Delete (irreversible — already-delivered data is NOT deleted)
aws kafka delete-channel \
    --cluster-arn "arn:aws:kafka:REGION:ACCOUNT_ID:cluster/CLUSTER/ID" \
    --channel-arn "CHANNEL_ARN"

You CANNOT update the topic, record converter, destination bucket/template, service role, or logging config. Only DataFreshnessInSeconds is mutable. Delete and recreate for anything else.

Security Considerations

• Enable server-side encryption (SSE-KMS or SSE-S3) on both delivery and DLQ buckets

• Enable S3 access logging on the delivery bucket

• Apply bucket policies restricting access to authorized principals only

• DLQ bucket may contain sensitive customer data (PII, financial data) — encrypt and restrict access

• Encrypt CloudWatch Logs log groups that receive Data Delivery for General Purpose S3 Buckets delivery logs

• Use VPC endpoints for S3 where applicable to keep traffic off the public internet

• For additional hardening guidance, see the MSK Security chapter and Amazon S3 security best practices

References

• Amazon MSK Data Delivery