migrate-to-msk — Skillopedia

Migrating to MSK Express Overview This skill helps customers migrate self-managed Apache Kafka workloads to Amazon MSK Express. It provides three phases — Discovery , Assessment , and an optional Simulation — that can be run end-to-end or individually depending on the customer's needs. Scope This skill covers migrations from self-managed Apache Kafka (on-premises, EC2, Docker, Kubernetes, or other non-MSK deployments) to MSK Express. Migrations from MSK Standard (Provisioned) to MSK Express are out of scope. Prerequisites The AWS MCP server is recommended for documentation lookups and informa…

; then\n aws cloudwatch put-metric-data --region $REGION --namespace MSKSimulation \\\n --metric-name E2ELatencyMs --unit Milliseconds --value \"$MS\" --dimensions Test=E2ELatency,Cluster=$CL\n fi\n sleep 30\n done\n fi\n wait\n\n RunBrokerRestartTestDoc:\n Type: AWS::SSM::Document\n Properties:\n DocumentType: Command\n Name: !Sub \"${AWS::StackName}-RunBrokerRestartTest\"\n Content:\n schemaVersion: \"2.2\"\n description: >\n Broker Restart Under Load. Producers sustain target throughput while\n consumers read the topic (group simulation-restart) so both the produce and\n consume paths are exercised; at reboot_at_minute the lead producer reboots\n one broker via kafka:RebootBroker and emits a timeseries of produce latency\n (MSKSimulation/ProduceLatencyMs) so the dashboard shows the impact and\n recovery. Branches on the simulation:role tag.\n parameters:\n perInstanceThroughputMbps: {type: String, default: \"50\"}\n durationMinutes: {type: String, default: \"15\"}\n rebootAtMinute: {type: String, default: \"5\"}\n recordSizeBytes: {type: String, default: \"1024\"}\n maxDrainMinutes: {type: String, default: \"20\", description: \"Hard cap (minutes) on the post-producer drain. After producers stop, the consumer keeps reading until the group's lag reaches ~0 (caught up) or this many minutes elapse, whichever first.\"}\n leadInstanceId: {type: String, default: \"\", description: \"Instance id that reboots a broker (the first targeted producer).\"}\n topic: {type: String, default: !Sub \"${TopicName}-restart\"}\n mainSteps:\n - action: aws:runShellScript\n name: run\n inputs:\n timeoutSeconds: \"8400\"\n runCommand:\n - !Sub |\n set -e\n if ! /opt/kafka/bin/kafka-topics.sh --version >/dev/null 2>&1; then echo \"ERROR: Kafka client not installed on this instance; run the InstallKafkaClient SSM document first.\" >&2; exit 1; fi\n REGION=${AWS::Region}\n CL=${MskCluster}\n BS=$(aws kafka get-bootstrap-brokers --cluster-arn $CL --region $REGION --query BootstrapBrokerStringSaslIam --output text)\n DUR={{durationMinutes}}; SECS=$((DUR*60)); RA={{rebootAtMinute}}\n CAP={{maxDrainMinutes}}; CAPS=$((CAP*60)) # hard cap on the adaptive drain (see consumer branch)\n TOPIC={{topic}}; RS={{recordSizeBytes}}\n P=/opt/kafka/client.properties; K=/opt/kafka/bin\n TOK=$(curl -sX PUT http://169.254.169.254/latest/api/token -H \"X-aws-ec2-metadata-token-ttl-seconds: 300\")\n IID=$(curl -s -H \"X-aws-ec2-metadata-token: $TOK\" http://169.254.169.254/latest/meta-data/instance-id)\n - |\n # Consumers read the topic so the restart exercises the consume path too.\n ROLE=$(curl -s -H \"X-aws-ec2-metadata-token: $TOK\" http://169.254.169.254/latest/meta-data/tags/instance/simulation:role)\n if [ \"$ROLE\" = \"consumer\" ]; then\n # Background consumer; drains until caught up (or maxDrainMinutes) so consume-side metrics reflect the test.\n GRP=simulation-restart\n # Reset the consumer group to the topic's latest so each run starts clean.\n $K/kafka-consumer-groups.sh --bootstrap-server $BS --command-config $P --group $GRP --topic $TOPIC --reset-offsets --to-latest --execute >/dev/null 2>&1 || true\n nohup $K/kafka-consumer-perf-test.sh --bootstrap-server $BS --consumer.config $P --topic $TOPIC --group $GRP --from-latest --messages 1000000000 --timeout 2147483647 --reporting-interval 5000 --show-detailed-stats >/tmp/consumer.out 2>&1 &\n CPID=$!\n PRODEND=$(( $(date +%s) + SECS )) # producers stop ~here\n HARDEND=$(( PRODEND + CAPS )) # absolute cap on the drain\n while kill -0 $CPID 2>/dev/null && [ $(date +%s) -lt $HARDEND ]; do\n sleep 15\n if [ $(date +%s) -lt $PRODEND ]; then continue; fi # don't test \"caught up\" until producers stop\n LAG=$($K/kafka-consumer-groups.sh --bootstrap-server $BS --command-config $P --describe --group $GRP 2>/dev/null | awk 'NR>1 && $6 ~ /^[0-9]+$/ {s+=$6; n++} END{ if (n>0) print s; else print \"NA\" }') || LAG=NA\n echo \"adaptive-drain: total_lag=$LAG\"\n if [ \"$LAG\" != \"NA\" ] && [ \"$LAG\" -le 1000 ]; then echo \"adaptive-drain: caught up (lag\u003c=1000), stopping\"; break; fi\n done\n kill $CPID 2>/dev/null || true\n exit 0\n fi\n # Producers generate the aggregate target load.\n RECS=$(awk -v tput={{perInstanceThroughputMbps}} -v rs=$RS 'BEGIN{printf \"%d\", tput*1024*1024/rs}')\n TOTAL=$(( RECS * SECS ))\n nohup $K/kafka-producer-perf-test.sh --topic $TOPIC --num-records $TOTAL --record-size $RS --throughput $RECS \\\n --producer-props bootstrap.servers=$BS --producer.config $P >/tmp/perf.out 2>&1 &\n PERF=$!\n START=$(date +%s)\n if [ \"$IID\" = \"{{leadInstanceId}}\" ]; then\n sleep $(( RA * 60 ))\n # Select a broker to reboot.\n BID=$(aws kafka list-nodes --cluster-arn $CL --region $REGION --query 'NodeInfoList[?NodeType==`BROKER`].BrokerNodeInfo.BrokerId | [0]' --output text)\n BID=${BID%.*}\n if echo \"$BID\" | grep -qE '^[0-9]+

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

; then\n aws kafka reboot-broker --cluster-arn $CL --broker-ids \"$BID\" --region $REGION || true\n else\n echo \"WARN: could not resolve a broker id to reboot (got '$BID'); skipping reboot\" >&2\n fi\n fi\n # Publish produce latency every 30s so the dashboard shows impact and recovery.\n # Anchor the window to the test start (not after the reboot sleep) so the lead\n # producer's reporting loop ends at durationMinutes, not rebootAtMinute + duration.\n END=$(( START + SECS ))\n while [ $(date +%s) -lt $END ]; do\n MS=$(grep -oE '[0-9]+\\.?[0-9]* ms avg latency' /tmp/perf.out | tail -1 | grep -oE '[0-9]+\\.?[0-9]*' | head -1)\n if echo \"$MS\" | grep -qE '^[0-9]+\\.?[0-9]*

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

; then\n aws cloudwatch put-metric-data --region $REGION --namespace MSKSimulation \\\n --metric-name ProduceLatencyMs --unit Milliseconds --value \"$MS\" --dimensions Test=BrokerRestart,Cluster=$CL\n fi\n sleep 30\n done\n wait $PERF || true\n\n InstallKafkaClientDoc:\n Type: AWS::SSM::Document\n Properties:\n DocumentType: Command\n Name: !Sub \"${AWS::StackName}-InstallKafkaClient\"\n Content:\n schemaVersion: \"2.2\"\n description: >\n Install the Kafka client and the aws-msk-iam-auth jar on the fleet. The\n Kafka client tarball URL is provided as a parameter at invocation\n (kafkaUrl); the aws-msk-iam-auth jar is installed from its AWS-owned source.\n parameters:\n kafkaUrl:\n type: String\n description: \"URL to the Kafka client tarball to install.\"\n kafkaSha512:\n type: String\n description: \"SHA-512 checksum of the Kafka client tarball; verified before install.\"\n mainSteps:\n - action: aws:runShellScript\n name: install\n inputs:\n runCommand:\n - |\n set -xe\n mkdir -p /opt/kafka/bin /opt/kafka/libs\n cd /opt\n # Require an HTTPS URL for the customer-provided client tarball before downloading.\n KURL=\"{{kafkaUrl}}\"\n case \"$KURL\" in\n https://*) : ;;\n *) echo \"ERROR: kafkaUrl must be an HTTPS URL (got: $KURL).\" >&2; exit 1 ;;\n esac\n # Install the Kafka client from the provided URL.\n curl -fsSL \"$KURL\" -o kafka.tgz\n # Validate the checksum is a 128-char lowercase hex string before use\n # (it is interpolated into a shell command, so reject metacharacters).\n SHA=\"{{kafkaSha512}}\"\n if ! echo \"$SHA\" | grep -qxE '[0-9a-f]{128}'; then\n echo \"ERROR: kafkaSha512 must be a 128-char lowercase hex string.\" >&2; exit 1\n fi\n # Verify the tarball against the customer-provided SHA-512 before using it.\n echo \"$SHA kafka.tgz\" | sha512sum -c -\n tar xzf kafka.tgz --strip-components=1 -C /opt/kafka\n # Install the aws-msk-iam-auth jar from its AWS-owned source (required for MSK IAM auth).\n curl -fsSL https://github.com/aws/aws-msk-iam-auth/releases/download/v2.2.0/aws-msk-iam-auth-2.2.0-all.jar -o /opt/kafka/libs/aws-msk-iam-auth.jar\n # Verify the pinned jar against its known SHA-256.\n echo \"735dd058b81d088e767770d6cc4e7f1a1beea815ced18b66a23b35fd87651e77 /opt/kafka/libs/aws-msk-iam-auth.jar\" | sha256sum -c -\n cat > /opt/kafka/client.properties \u003c\u003c'EOF'\n security.protocol=SASL_SSL\n sasl.mechanism=AWS_MSK_IAM\n sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;\n sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler\n EOF\n /opt/kafka/bin/kafka-topics.sh --version\n\n ValidateClientDoc:\n Type: AWS::SSM::Document\n Properties:\n DocumentType: Command\n Name: !Sub \"${AWS::StackName}-ValidateClient\"\n Content:\n schemaVersion: \"2.2\"\n description: >\n Validate the installed Kafka client against the cluster: confirms the client\n runs, its version is >= the cluster's Kafka version, and it can authenticate\n (IAM over TLS) and list topics. Fails with a clear message if the client is\n missing or incompatible.\n mainSteps:\n - action: aws:runShellScript\n name: validate\n inputs:\n runCommand:\n - !Sub |\n set -e\n if ! /opt/kafka/bin/kafka-topics.sh --version >/dev/null 2>&1; then\n echo \"ERROR: Kafka client not installed/working. Run ${AWS::StackName}-InstallKafkaClient first.\" >&2; exit 1\n fi\n CLIENT_VER=$(/opt/kafka/bin/kafka-topics.sh --version 2>/dev/null | awk '{print $1}')\n CLUSTER_VER=$(echo \"${KafkaVersion}\" | sed 's/\\.kraft$//; s/\\.x/.0/')\n LOWER=$(printf '%s\\n%s\\n' \"$CLIENT_VER\" \"$CLUSTER_VER\" | sort -V | head -n1)\n if [ \"$CLIENT_VER\" != \"$CLUSTER_VER\" ] && [ \"$LOWER\" = \"$CLIENT_VER\" ]; then\n echo \"ERROR: Kafka client version $CLIENT_VER is older than the cluster's $CLUSTER_VER; install a client >= the cluster version.\" >&2; exit 1\n fi\n BS=$(aws kafka get-bootstrap-brokers --cluster-arn ${MskCluster} --region ${AWS::Region} --query BootstrapBrokerStringSaslIam --output text)\n if ! /opt/kafka/bin/kafka-topics.sh --bootstrap-server \"$BS\" --command-config /opt/kafka/client.properties --list >/dev/null 2>&1; then\n echo \"ERROR: client $CLIENT_VER cannot authenticate/connect to the cluster (IAM/TLS/list failed). If you installed a custom client, fix it or re-run ${AWS::StackName}-InstallKafkaClient to use the official client.\" >&2; exit 1\n fi\n echo \"Kafka client OK: version $CLIENT_VER >= cluster $CLUSTER_VER, authenticated and listed topics on the cluster.\"\n\n # ---------------- Dashboard ----------------\n Dashboard:\n Type: AWS::CloudWatch::Dashboard\n Properties:\n DashboardName: !Sub \"${AWS::StackName}-loadtest\"\n DashboardBody: !Sub |\n {\n \"widgets\": [\n {\n \"type\": \"text\",\n \"x\": 0,\n \"y\": 0,\n \"width\": 24,\n \"height\": 5,\n \"properties\": {\n \"markdown\": \"## Client metrics \\u2014 client fleet, consumers & probes\\n- **Fleet CPU (EC2 CPUUtilization)**: CPU on the producer/consumer instances. Saturation here indicates the client fleet, not the cluster, is the bottleneck.\\n- **Fleet network (EC2 NetworkOut/NetworkIn)**: per-NIC throughput on the client instances; compare against instance bandwidth to detect fleet saturation.\\n- **End-to-end latency (E2ELatencyMs)**: probe-measured produce-to-consume round trip; the primary latency SLA indicator.\\n- **Produce latency during restart (ProduceLatencyMs)**: client-observed produce latency across the broker-restart window, showing impact and recovery.\\n- **Consumer lag (MaxOffsetLag)**: messages by which the test consumers trail the log head. A sustained rise indicates consumers cannot keep pace.\"\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 0,\n \"y\": 5,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"Fleet CPU % \\u2014 client fleet (maxed = fleet is the bottleneck)\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"AVG(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\\\"${AWS::StackName}-producer\\\" MetricName=\\\"CPUUtilization\\\"', 'Average', 60))\",\n \"label\": \"producers\",\n \"id\": \"pcpu\"\n }\n ],\n [\n {\n \"expression\": \"AVG(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\\\"${AWS::StackName}-consumer\\\" MetricName=\\\"CPUUtilization\\\"', 'Average', 60))\",\n \"label\": \"consumers\",\n \"id\": \"ccpu\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 12,\n \"y\": 5,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"Fleet network bytes/min \\u2014 producer egress / consumer ingress\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"SUM(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\\\"${AWS::StackName}-producer\\\" MetricName=\\\"NetworkOut\\\"', 'Average', 60))\",\n \"label\": \"producers NetworkOut\",\n \"id\": \"pno\"\n }\n ],\n [\n {\n \"expression\": \"SUM(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\\\"${AWS::StackName}-consumer\\\" MetricName=\\\"NetworkIn\\\"', 'Average', 60))\",\n \"label\": \"consumers NetworkIn\",\n \"id\": \"cni\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 0,\n \"y\": 11,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"End-to-end latency (ms) \\u2014 produce->consume round trip (probe)\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"MAX(SEARCH('{MSKSimulation,Cluster,Test} Test=\\\"E2ELatency\\\" MetricName=\\\"E2ELatencyMs\\\"', 'Average', 60))\",\n \"label\": \"E2ELatencyMs\",\n \"id\": \"e2elat\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 12,\n \"y\": 11,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"Produce latency (ms) during broker restart \\u2014 watch spike & recovery\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"MAX(SEARCH('{MSKSimulation,Cluster,Test} Test=\\\"BrokerRestart\\\" MetricName=\\\"ProduceLatencyMs\\\"', 'Average', 60))\",\n \"label\": \"ProduceLatencyMs\",\n \"id\": \"prodlat\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 0,\n \"y\": 17,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"Consumer lag \\u2014 how far consumers trail (messages behind, per test)\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"MAX(SEARCH('{AWS/Kafka,\\\"Cluster Name\\\",\\\"Consumer Group\\\",\\\"Topic\\\"} \\\"Cluster Name\\\"=\\\"${AWS::StackName}-express\\\" \\\"Consumer Group\\\"=\\\"simulation-e2e\\\" MetricName=\\\"MaxOffsetLag\\\"', 'Average', 60))\",\n \"label\": \"E2E lag - simulation-e2e (msgs behind)\",\n \"id\": \"lage2e\"\n }\n ],\n [\n {\n \"expression\": \"MAX(SEARCH('{AWS/Kafka,\\\"Cluster Name\\\",\\\"Consumer Group\\\",\\\"Topic\\\"} \\\"Cluster Name\\\"=\\\"${AWS::StackName}-express\\\" \\\"Consumer Group\\\"=\\\"simulation-restart\\\" MetricName=\\\"MaxOffsetLag\\\"', 'Average', 60))\",\n \"label\": \"Broker Restart lag - simulation-restart (msgs behind)\",\n \"id\": \"lagrestart\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"text\",\n \"x\": 0,\n \"y\": 23,\n \"width\": 24,\n \"height\": 5,\n \"properties\": {\n \"markdown\": \"## Cluster metrics \\u2014 Amazon MSK (broker side)\\n- **Bytes In/Out per sec**: aggregate write/read throughput across all brokers. A flat write ceiling indicates the per-broker bandwidth quota has been reached.\\n- **CPU user (CpuUser)**: broker CPU spent on Kafka. Sustained >60% indicates an overloaded cluster.\\n- **Health (ActiveControllerCount / UnderReplicatedPartitions / OfflinePartitionsCount)**: expected steady state is 1 / 0 / 0. The latter two spike transiently during a broker restart.\\n- **Replication bytes/sec (ReplicationBytesOutPerSec)**: inter-broker replication; spikes as a restarted broker catches up.\"\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 0,\n \"y\": 28,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"Bytes In/Out per sec \\u2014 total write/read traffic (all brokers)\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"SUM(SEARCH('{AWS/Kafka,\\\"Broker ID\\\",\\\"Cluster Name\\\"} \\\"Cluster Name\\\"=\\\"${AWS::StackName}-express\\\" MetricName=\\\"BytesInPerSec\\\"', 'Average', 60))\",\n \"label\": \"BytesInPerSec (cluster)\",\n \"id\": \"bin\"\n }\n ],\n [\n {\n \"expression\": \"SUM(SEARCH('{AWS/Kafka,\\\"Broker ID\\\",\\\"Cluster Name\\\"} \\\"Cluster Name\\\"=\\\"${AWS::StackName}-express\\\" MetricName=\\\"BytesOutPerSec\\\"', 'Average', 60))\",\n \"label\": \"BytesOutPerSec (cluster)\",\n \"id\": \"bout\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 12,\n \"y\": 28,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"CPU user % \\u2014 broker CPU on Kafka (keep \u003c 60%)\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"AVG(SEARCH('{AWS/Kafka,\\\"Broker ID\\\",\\\"Cluster Name\\\"} \\\"Cluster Name\\\"=\\\"${AWS::StackName}-express\\\" MetricName=\\\"CpuUser\\\"', 'Average', 60))\",\n \"label\": \"CpuUser (avg %)\",\n \"id\": \"cpu\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 0,\n \"y\": 34,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"Health \\u2014 controller=1; under-replicated & offline=0\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n \"AWS/Kafka\",\n \"ActiveControllerCount\",\n \"Cluster Name\",\n \"${AWS::StackName}-express\"\n ],\n [\n \"AWS/Kafka\",\n \"UnderReplicatedPartitions\",\n \"Cluster Name\",\n \"${AWS::StackName}-express\"\n ],\n [\n \"AWS/Kafka\",\n \"OfflinePartitionsCount\",\n \"Cluster Name\",\n \"${AWS::StackName}-express\"\n ],\n [\n {\n \"expression\": \"SUM(SEARCH('{AWS/Kafka,\\\"Broker ID\\\",\\\"Cluster Name\\\"} \\\"Cluster Name\\\"=\\\"${AWS::StackName}-express\\\" MetricName=\\\"UnderMinIsrPartitionCount\\\"', 'Average', 60))\",\n \"label\": \"UnderMinIsrPartitionCount (sum)\",\n \"id\": \"uminisr\"\n }\n ]\n ]\n }\n },\n {\n \"type\": \"metric\",\n \"x\": 12,\n \"y\": 34,\n \"width\": 12,\n \"height\": 6,\n \"properties\": {\n \"title\": \"Replication bytes/sec \\u2014 re-replication (spikes as a restarted broker catches up)\",\n \"region\": \"${AWS::Region}\",\n \"stat\": \"Average\",\n \"period\": 60,\n \"metrics\": [\n [\n {\n \"expression\": \"SUM(SEARCH('{AWS/Kafka,\\\"Broker ID\\\",\\\"Cluster Name\\\"} \\\"Cluster Name\\\"=\\\"${AWS::StackName}-express\\\" MetricName=\\\"ReplicationBytesOutPerSec\\\"', 'Average', 60))\",\n \"label\": \"ReplicationBytesOutPerSec (cluster)\",\n \"id\": \"repout\"\n }\n ]\n ]\n }\n }\n ]\n }\nOutputs:\n ClusterArn:\n Value: !Ref MskCluster\n DashboardUrl:\n Value: !Sub \"https://${AWS::Region}.console.aws.amazon.com/cloudwatch/home?region=${AWS::Region}#dashboards:name=${AWS::StackName}-loadtest\"\n DashboardArn:\n Value: !Sub \"arn:${AWS::Partition}:cloudwatch::${AWS::AccountId}:dashboard/${AWS::StackName}-loadtest\"\n CreateTopicsCommand:\n Value: !Sub \"aws ssm send-command --document-name ${AWS::StackName}-CreateTopics --targets Key=tag:simulation:role,Values=producer --max-concurrency 1 --region ${AWS::Region}\"\n InstallKafkaClientCommand:\n Description: >\n Installs the Kafka client (from the kafkaUrl you pass) and the aws-msk-iam-auth\n jar (from its AWS-owned source) on the fleet.\n Value: !Sub \"aws ssm send-command --document-name ${AWS::StackName}-InstallKafkaClient --targets Key=tag:simulation:role,Values=producer,consumer --parameters 'kafkaUrl=\u003ckafka-client-url>,kafkaSha512=\u003csha512>' --region ${AWS::Region}\"\n ValidateClientCommand:\n Description: >\n Validates the installed Kafka client against this cluster (version >= cluster,\n and live IAM/TLS auth + topic list). Run after install and before tests; re-run\n after any client fix.\n Value: !Sub \"aws ssm send-command --document-name ${AWS::StackName}-ValidateClient --targets Key=tag:simulation:role,Values=producer,consumer --region ${AWS::Region}\"\n FleetSummary:\n Value: !Sub \"${ProducerCount} x ${ClientInstanceType} producers, ${ConsumerCount} consumers; cluster ${BrokerCount} x ${InstanceType}\"\n","content_type":"application/yaml; charset=utf-8","language":"yaml","size":49078,"content_sha256":"e77824730db2221ddefdc2578b11338eceb69acd650bde6028599d44477a0190"},{"filename":"references/assessment-compatibility.md","content":"# Assessment — Compatibility\n\nThis reference describes the compatibility assessment for migrating a\nself-managed Apache Kafka cluster to MSK Express. It documents the five\npillars that `scripts/compatibility.py` evaluates — topology, Kafka version,\nconfigs, auth, and quotas — together with the AWS-doc-anchored thresholds and\nthe per-finding verdicts each pillar emits.\n\n`compatibility.py` is a pure file processor: input is the discovery contract\n`migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json`; output is\n`compatibility.\u003ccluster_name>.json`. No live cluster or AWS API calls.\n\n## Guardrail — where this skill's own files live (MCP vs local install)\n\nThis skill can be loaded two ways, and they resolve the skill's **own bundled\nfiles** — the `references/` documents and the `scripts/` files\n(`compatibility.py`, `simulation_load_test_config.py`) — from different places.\nDetermine how the skill was loaded before you read a reference or run a script:\n\n- **Loaded through the AWS MCP `retrieve_skill` tool call.** The skill is **not\n installed on the local filesystem**; its reference files and scripts do not\n exist on disk. You MUST fetch each reference or script through the same\n `retrieve_skill` tool by passing the `file` parameter (for example,\n `file=\"references/assessment-compatibility.md\"` or `file=\"scripts/compatibility.py\"`),\n and run a script from the content that tool returns. Do NOT `file_read` these\n paths from the local or working directory, and do NOT search the filesystem\n for them — they are not there, and any local file that happens to match the\n name is unrelated to this skill.\n- **Installed locally** (the skill lives in a local skills directory such as\n `.claude/skills/migrate-to-msk/`, `~/.claude/skills/migrate-to-msk/`, or\n `.kiro/skills/migrate-to-msk/`). Read references and run scripts from the\n local skill directory using the relative paths shown throughout this\n documentation (`uv run scripts/compatibility.py ...`).\n\nThis distinction applies **only** to the skill's own packaged files. The\ncustomer's data is always on the local filesystem in both modes: the input IaC\nfiles, the discovery `cluster-config.json`, and every artifact under\n`migrate-to-msk-skill-artifacts/\u003ccluster_name>/` are read from and written to\nthe customer's working directory regardless of how the skill was loaded. Never\nfetch or write customer data through `retrieve_skill`.\n\n## Running the assessment\n\nRun both scripts with `uv run`, which resolves each script's declared\ndependencies automatically. Resolve the script location first per the guardrail\nabove — when this skill was loaded through the MCP `retrieve_skill` tool, fetch\neach script via `retrieve_skill` with the `file` parameter and run that content;\nwhen it is installed locally, run it from the local skill directory. Pass the\ndiscovery `cluster-config.json` as input, and write the outputs into the same\nper-cluster artifacts directory:\n\n```bash\n# Compatibility -> compatibility.\u003ccluster_name>.json\nuv run scripts/compatibility.py \\\n migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json \\\n --out-dir migrate-to-msk-skill-artifacts/\u003ccluster_name>\n\n```\n\nSizing is not a script in this skill. Load the managing-amazon-msk Skill and run\nits `scripts/msk_sizing.py` with the workload inputs derived from\n`cluster-config.json`, passing `--broker-classes express`. Write\nthe summary to\n`migrate-to-msk-skill-artifacts/\u003ccluster_name>/msk_sizing_pricing.md` and the\ninput record to `msk-sizing-inputs.\u003ccluster_name>.json`. See\n[assessment-sizing.md](./assessment-sizing.md) for the full flow, the Express-only\npresentation rule, and the source-footprint comparison.\n\nCompatibility and sizing are independent: run them in either order, and a failure\nin one does not block the other.\n\n## Source of truth (AWS public docs)\n\nEvery threshold and rule below is anchored to one of these AWS public\ndocumentation pages. When AWS publishes updates, refresh the constants in\n`compatibility.py` and re-run tests.\n\n- [Express broker overview](https://docs.aws.amazon.com/msk/latest/developerguide/msk-broker-types-express.html)\n- [Express read/write broker and topic configurations](https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-express-read-write.html)\n- [Express read-only broker configurations](https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-express-read-only.html)\n- [Express broker quotas](https://docs.aws.amazon.com/msk/latest/developerguide/limits.html#msk-express-quota)\n- [Express broker best practices](https://docs.aws.amazon.com/msk/latest/developerguide/bestpractices-express.html)\n\nApache Kafka per-version defaults (used to filter the full config dump):\n\n- [Kafka broker configs](https://kafka.apache.org/documentation/#brokerconfigs)\n- [Kafka topic configs](https://kafka.apache.org/documentation/#topicconfigs)\n\n## Assessment scope — forbidden behavior\n\nAssessment operates on the existing `cluster-config.json` produced by Phase 1\n(Discovery). It is NOT discovery, and it does NOT pivot back into discovery\nwhen the input is incomplete.\n\nWhen in Assessment, you MUST NOT:\n\n- Discuss the discovery phase, mention \"Phase 1\", or describe how\n `cluster-config.json` is produced.\n- Propose, suggest, or display any method for gathering more cluster\n information — no Kafka CLI commands (`kafka-topics.sh`, `kafka-configs.sh`,\n `kafka-broker-api-versions.sh`, etc.), no IaC file walks, no Python\n discovery scripts, no `boto3` calls, no manual questionnaires, no\n \"would you like me to fetch X from your cluster?\" prompts.\n- Ask the user to provide additional cluster fields, peaks, configs, or\n topics. The contract is fixed at the input file.\n\n**Two exceptions,** both cost assumptions about the target rather than facts about the\nsource cluster, and both material to the estimate: whether consumers will use\nrack-aware fetching (`target.rack_affined_consumers`) and whether the customer has a\nnegotiated discount to apply (`target.pricing_discount_pct`). Discovery collects both\ninto the `target` block — read them from `cluster-config.json` first. Ask only for the\nones that are `null` or absent, once, together, before running sizing — see\n\"Negotiated pricing\" and \"Consumer rack affinity\" in\n[assessment-sizing.md](./assessment-sizing.md). Asking them does not license any\nother question.\n\n**Partial data is fine.** When required fields are missing or empty\n(no `metrics` block, no `num_azs`, no broker configs, etc.), the scripts\nalready emit ADVISORY evidence describing the gap (`METRICS_MISSING`,\n`AZ_COUNT_UNKNOWN`, etc.). Surface those verdicts to the user as-is and\nstop. Do not offer to gather the missing data yourself. If the user wants\nbetter assessment fidelity, they can re-run Phase 1 — but suggesting that\nis a routing decision the user makes, not something Assessment proposes\nmid-flow.\n\nIf the input is malformed (`compatibility.py`'s `validate_input` raises),\nreport the error from the script and stop. Do not improvise around it.\n\n## Response Template\n\nAfter running `compatibility.py` and the managing-amazon-msk sizing script, your\nresponse MUST follow the template below exactly. One template covers both\nartifacts in a single response.\n\nFORBIDDEN content — do NOT include any of the following:\n\n- Free-form narrative summaries beyond what the template allows.\n- Editorial framings (\"looks good overall\", \"this is a clean migration\",\n \"you're in great shape\", \"minor issues only\", etc.). State facts; let the\n user judge.\n- Numeric scores, percentages, or \"readiness scores\". The skill emits\n categorical verdicts only.\n- Recommendations to run discovery again, fetch more cluster data, or\n invoke any Kafka CLI / IaC walk / questionnaire (see \"Assessment scope\"\n above).\n- MSK Replicator commands or any migration execution detail. That belongs\n to a later conversation, not the assessment response.\n- \"Action items\", \"Next steps\", or \"Recommendations\" sections beyond the\n one mandated below.\n- Broker counts or cost figures you did not read from the sizing script output.\n Report the script's Express pick verbatim; never round, re-derive, or estimate.\n- Standard instance types, broker counts, or costs, unless the customer asked to\n compare classes (see \"Present Express only\" in\n [assessment-sizing.md](./assessment-sizing.md)).\n\n### Template\n\n```\n## Assessment Complete — \u003ccluster_name>\n\n**How to read these results:**\n\n- **`INFO`** — This setting already matches how MSK Express works. No action needed — it's listed so you can see it was checked.\n- **`ADVISORY`** — Your cluster differs from MSK Express here, and Express adjusts or replaces this for you during migration. You can migrate as-is, but we recommend reviewing it so the change in behavior is expected; where a setting behaves differently, validate in a test environment first.\n- **`ACTION_REQUIRED`** — MSK Express won't accept this configuration as-is, so it can't be migrated unchanged. We'll walk you through what to adjust before you migrate.\n**Overall verdict:** \u003cINFO | ADVISORY | ACTION_REQUIRED>\n\n### Compatibility — by pillar\n\n| Pillar | Verdict | Findings |\n|---|---|---|\n| Topology | \u003cverdict> | \u003ccount> finding(s) |\n| Kafka version | \u003cverdict> | \u003ccount> finding(s) |\n| Configs | \u003cverdict> | \u003ccount> finding(s) |\n| Auth | \u003cverdict> | \u003ccount> finding(s) |\n| Quotas | \u003cverdict> | \u003ccount> finding(s) |\n\n### Findings\n\n(One bullet per evidence object emitted by compatibility.py. List only\nnon-INFO findings. Use the `code` and `detail` from the evidence verbatim.\nGroup by pillar in the same order as the table.)\n\n- **`\u003cEVIDENCE_CODE>`** [\u003cseverity>] — \u003cdetail string from compatibility.py>\n- ...\n\nIf there are zero non-INFO findings, replace this section with:\n\"No advisories or action-required items. All five pillars: INFO.\"\n\n### Recommended Express target\n\n**`\u003cexpress instance type>` — \u003cN> brokers — $\u003ctotal>/mo (us-east-1)**\n\nBottleneck: \u003cingress | egress | partitions | storage>. \u003cOne sentence on the demand\nvs per-broker capacity that drives the count, from the script's output.>\n\nYour source cluster runs \u003csource broker count> broker(s)\u003c, instance type\n`\u003cbroker_instance_type>` when present>. \u003cState the delta against the recommended\ntarget. When the target is smaller, explain why per \"Compare the recommendation\nagainst the source footprint\" in assessment-sizing.md. When\n`broker_instance_type` is absent, compare broker counts only and say the\ninstance-level comparison was not possible.>\n\nFull breakdown — inputs, per-instance options, cost components, and cost-optimization\nopportunities:\n\n`migrate-to-msk-skill-artifacts/\u003ccluster_name>/msk_sizing_pricing.md`\n\nVerify the per-cluster broker quota before provisioning: **60 brokers with KRaft,\n30 with ZooKeeper** (adjustable via Service Quotas). Cost figures use public\non-demand pricing for the Region you sized against — see the\n[Amazon MSK pricing page](https://aws.amazon.com/msk/pricing/). State the Region and,\nwhen the sizing run reported a `static` pricing source, say the rates came from the\nscript's embedded snapshot rather than a live lookup.\n\n### Artifacts produced\n\n- `migrate-to-msk-skill-artifacts/\u003ccluster_name>/compatibility.\u003ccluster_name>.json`\n- `migrate-to-msk-skill-artifacts/\u003ccluster_name>/msk_sizing_pricing.md`\n- `migrate-to-msk-skill-artifacts/\u003ccluster_name>/msk-sizing-inputs.\u003ccluster_name>.json`\n\n---\n\nWould you like to discuss data replication strategy or revisit any of the\nfindings above?\n```\n\n### Rules\n\n- The response MUST start with `## Assessment Complete — \u003ccluster_name>`\n using the `cluster_name` from the input.\n- The response MUST open with the \"How to read these results:\" legend\n exactly as shown, then the overall verdict.\n- Use the verdict strings (`INFO` / `ADVISORY` / `ACTION_REQUIRED`)\n verbatim in the legend, verdict, table, and findings. Do not translate to\n \"Pass / Warn / Fail\" or any other vocabulary.\n- The findings list reproduces evidence `code` and `detail` strings as-is.\n Do not paraphrase the script's wording, do not drop the code, do not\n reorder severity components within a finding.\n- The \"Recommended Express target\" section carries the instance type, broker\n count, and monthly cost read verbatim from the sizing script. Do not add cost\n sections beyond it, and do not restate the per-instance option table — that\n lives in `msk_sizing_pricing.md`.\n- Do NOT add a confidence rating, risk score, or quality bar.\n- Do NOT add migration timeline estimates.\n- The closing question is fixed: ask whether to discuss data replication\n strategy or revisit findings. No alternative phrasings.\n\n## Pillars\n\n`compatibility.py` runs five pillars; the pillar-roll-up is the worst per-finding severity.\n\n### 1. Topology (`assess_topology`)\n\nThe 3-AZ requirement, KRaft availability, and the 3-broker minimum come from\nthe Express broker overview page. The target broker count is determined by the\nsizing script, not carried over from the source, so this pillar does not\ncompare the source broker count against a per-cluster ceiling.\n\n| Code | Severity | When |\n|---|---|---|\n| `AZ_COUNT_UNKNOWN` | ADVISORY | `topology.num_azs` missing |\n| `AZ_COUNT_NOT_3` | ADVISORY | `topology.num_azs` ≠ 3 (note: Express always uses 3 AZs) |\n| `BROKER_COUNT_LT_3` | ADVISORY | `topology.num_brokers` \u003c 3 (note: Express minimum is 3; exact count comes from the sizing script) |\n| `KRAFT_REQUIRED_FOR_VERSION` | ADVISORY | `kafka.version` is 3.9 and `kafka.coordination_mechanism` is `ZooKeeper` |\n\nA single-broker or single-AZ source emits these two advisories rather than an\naction item: Express fixes it at the target by deploying across 3 AZs with ≥3\nbrokers regardless of the source layout.\n\n### 2. Kafka version (`assess_kafka_version`)\n\nPer the Express broker overview page, MSK Express supports Apache Kafka versions\n3.6, 3.8, and 3.9. Any version outside that set is ADVISORY: your cluster runs a\ndifferent version, so after migrating your workload will run on a new Kafka\nversion. Confirm your client libraries and applications are compatible with the\nExpress Kafka version you choose — Kafka clients are generally compatible across\nminor versions, but we recommend validating in a test environment before\nmigrating. See the Apache Kafka upgrade notes at\nhttps://kafka.apache.org/documentation/#upgrade for details. Sources older than\nApache Kafka\n2.8.1 additionally cannot be data-migrated with MSK Replicator (which requires a\n2.8.1+ source); use a MirrorMaker 2 based solution for data migration in that\ncase.\n\n| Code | Severity | When |\n|---|---|---|\n| `VERSION_SUPPORTED` | INFO | `kafka.version` ∈ {3.6, 3.8, 3.9} |\n| `VERSION_NOT_IN_EXPRESS_SET` | ADVISORY | any other version (older, newer, or a gap such as 3.7); message adds the MirrorMaker 2 note when `kafka.version` \u003c 2.8.1 |\n\n### 3. Configs (`assess_configs`)\n\nLargest pillar. Covers broker-level (`broker_configs`) and topic-level (`topics[].configs`) configs.\n\n**Default-value filtering.** Discovery passes the FULL Kafka config dump (every config the source exposed). `compatibility.py` compares each value against the Apache Kafka default for the source's `kafka.version`; values matching the default produce **no evidence**. Only divergences from default are evaluated against the rules below. See `BROKER_DEFAULTS_BY_VERSION` and `TOPIC_DEFAULTS_BY_VERSION` in the script.\n\n**Express config buckets** (sourced from the Express read/write and read-only\nbroker configuration pages):\n\n- **Editable** — configurable on Express. Your values carry over.\n- **Read-only** — Express enforces a fixed value (sometimes instance-derived). Custom overrides are silently replaced.\n- **Range-restricted** — editable, but with a documented bounded range; out-of-range values are rejected.\n- **Forced** — read-only with a known fixed enforcement value.\n- **Non-exposed** — not a configurable property on MSK Express. This value is managed internally and may differ from your current setting. We recommend validating on a test cluster to confirm the behavior meets your expectations before migrating production traffic.\n\nThe exact sets are constants in `compatibility.py` (`EXPRESS_BROKER_RW`, `EXPRESS_BROKER_RO`, `EXPRESS_BROKER_RANGES`, `EXPRESS_BROKER_FORCED`, and the topic-level equivalents).\n\n**Decision matrix** (applied to non-default values only):\n\n| Membership of key | Verdict |\n|---|---|\n| Editable and within range | INFO (no evidence emitted) |\n| Editable with a range, and value out of range | **ACTION_REQUIRED** (`*_CONFIG_OUT_OF_RANGE`) |\n| In forced set, value ≠ forced value | ADVISORY (`*_CONFIG_FORCED_VALUE`) |\n| Read-only (not in forced) | ADVISORY (`*_CONFIG_READ_ONLY`) |\n| Not in editable or read-only sets | ADVISORY (`*_CONFIG_NOT_EXPOSED`) |\n\nPer-topic replication factor: `replication_factor ≠ 3` → ADVISORY (`TOPIC_RF_NOT_3`); Express creates topics with a replication factor of 3 regardless.\n\n**Documented bounded ranges** (Express read/write configurations page):\n\n| Config | Bound |\n|---|---|\n| `log.cleaner.max.compaction.lag.ms` | [1 day = 86_400_000 ms, +∞] |\n| `max.compaction.lag.ms` (topic) | [1 day = 86_400_000 ms, +∞] |\n\n`max.compaction.lag.ms \u003c 1 day` is the only Express-rejected topic-config bound\nencoded in `compatibility.py`. Adjust it on the source before migration.\n\n**Forced values** (Express read-only configurations page):\n\n| Config | Express value |\n|---|---|\n| `default.replication.factor` | 3 |\n| `min.insync.replicas` (broker + topic) | 2 |\n| `transaction.state.log.min.isr` | 2 |\n| `unclean.leader.election.enable` (broker + topic) | false |\n\n### 4. Auth (`assess_auth`)\n\nMSK Express supports four client authentication mechanisms — unauthenticated,\nTLS (AWS Private CA), SASL/SCRAM (Secrets Manager), and IAM (`SASL/AWS_MSK_IAM`\nor `SASL/OAUTHBEARER`, both carrying an AWS IAM token via the AWS MSK IAM\nlibraries). TLS in transit is\n**required** for every authenticated mechanism; plaintext is only possible with\nunauthenticated access. The discovery contract carries two closed-enum fields,\n`security.encryption_in_transit` and `security.authentication`; a missing field\nor an explicit `UNKNOWN` is treated as undetermined and flagged ADVISORY (it is\nnot a hard failure). Unrecognized values are rejected at validation time.\n\nThe supported mechanisms (unauthenticated, TLS, SASL/SCRAM, IAM) carry over\nas-is and emit no evidence (INFO).\n\n| Code | Severity | When |\n|---|---|---|\n| `AUTH_OAUTHBEARER_NOT_SUPPORTED` | ACTION_REQUIRED | `security.authentication` = `SASL_OAUTHBEARER` (custom OAuth provider; MSK Express accepts OAUTHBEARER only as an AWS IAM token transport, not with external identity providers) |\n| `AUTH_UNKNOWN` | ADVISORY | `security.authentication` = `UNKNOWN` or missing (verify it is a supported mechanism) |\n| `AUTH_MECHANISM_NOT_SUPPORTED` | ACTION_REQUIRED | `security.authentication` = `OTHER` (e.g. SASL/GSSAPI/Kerberos, SASL/PLAIN — not supported by Express) |\n| `ENCRYPTION_NOT_TLS` | ACTION_REQUIRED | `security.encryption_in_transit` ≠ `TLS` **and** an authenticated mechanism is in use (TLS / SASL/SCRAM / IAM) — Express requires TLS for these; update clients before migrating |\n| `ENCRYPTION_UNKNOWN` | ADVISORY | `security.encryption_in_transit` = `UNKNOWN` or missing (confirm clients can use TLS) |\n\nUnauthenticated access is `INFO` — Express supports it, and plaintext is\npermitted in that case, so no encryption finding fires. Mechanism-specific\nre-credentialing work (SASL/SCRAM secret prefix, IAM policies, TLS Private CA\nassociation) is out of scope for compatibility — handled by migration planning.\n\n### 5. Quotas (`assess_quotas`)\n\nCompatibility checks **absolute** Express ceilings — workload can't fit *any* Express configuration. Sizing checks per-instance fit.\n\nAll values below come from the Express broker quotas page.\n\n| Limit | Value |\n|---|---|\n| Max ingress / broker (max-quota at m7g.16xlarge) | 750 MBps |\n| Max egress / broker (max-quota at m7g.16xlarge) | 1875 MBps |\n| Max partitions / broker | 32_000 |\n| Max IAM connections / broker | 3_000 |\n| Max throughput / partition | 15 MB/s |\n\n| Code | Severity | When |\n|---|---|---|\n| `METRICS_MISSING` | ADVISORY | `metrics` block absent |\n| `INGRESS_OVER_MAX_BROKER` | ADVISORY | `peak_bytes_in_per_broker_mbps` > 750 |\n| `EGRESS_OVER_MAX_BROKER` | ADVISORY | `peak_bytes_out_per_broker_mbps` > 1875 |\n| `PARTITIONS_OVER_MAX_BROKER` | ADVISORY | `peak_partitions_per_broker` > 32_000 |\n| `CONNECTIONS_OVER_IAM_LIMIT` | ADVISORY | `security.authentication` ∈ {`SASL_IAM`, `SASL_OAUTHBEARER`} AND `peak_connections_per_broker` > 3_000 |\n| `PARTITION_THROUGHPUT_OVER_LIMIT` | ADVISORY | average per-partition throughput (`peak_in × num_brokers / total_partitions`) > 15 MB/s |\n\nThe per-partition check is approximate; hot partitions can exceed 15 MB/s while the cluster average is fine.\n\n## Output schema\n\n```json\n{\n \"cluster_name\": \"\u003cfrom input>\",\n \"assessed_at\": \"\u003cISO-8601>\",\n \"overall\": \"INFO | ADVISORY | ACTION_REQUIRED\",\n \"pillars\": {\n \"topology\": {\"verdict\": \"...\", \"evidence\": [...]},\n \"kafka_version\": {\"verdict\": \"...\", \"evidence\": [...]},\n \"configs\": {\"verdict\": \"...\", \"evidence\": [...]},\n \"auth\": {\"verdict\": \"...\", \"evidence\": [...]},\n \"quotas\": {\"verdict\": \"...\", \"evidence\": [...]}\n },\n \"summary\": {\n \"action_required_codes\": [...],\n \"advisory_codes\": [...],\n \"info_codes\": [...]\n }\n}\n```\n\nEach evidence object carries `code`, `severity` (per-finding, used for summary bucketing), `detail`, plus optional `topic`, `config_key`, `observed`, `limit`, `enforced`.\n\n## Refresh procedure\n\n> **The script is the source of truth.** This reference documents the *rules*\n> `scripts/compatibility.py` enforces — the AWS-doc-anchored data and the\n> classification matrix. **Do not re-implement the script from this document.**\n> Always invoke `scripts/compatibility.py` directly. If the script's behavior\n> diverges from this reference, the script wins; update this reference instead.\n\nWhen AWS publishes updates to the anchor docs:\n\n1. Re-fetch the AWS public docs linked at the top of this file.\n2. Diff against the constants at the top of `compatibility.py`.\n3. Update `BROKER_DEFAULTS_BY_VERSION` / `TOPIC_DEFAULTS_BY_VERSION` if Apache Kafka adds a new supported version.\n4. Run `python3 -m pytest scripts/tests/`.\n\n## Security considerations\n\n- **Discovery input may contain credentials.** `cluster-config.json` is produced\n by the discovery phase. Before passing it to `compatibility.py`, verify it\n does not contain SASL passwords, API keys, TLS private keys, or other\n credential material. The discovery contract intentionally captures\n `security.authentication` (the mechanism) but never the secret itself;\n if your discovery output includes anything resembling a credential,\n redact it before processing or sharing.\n- **Output files reveal cluster topology and configuration.** The\n `compatibility.\u003ccluster_name>.json` output enumerates broker configs, topic\n configs, partition counts, and authentication mode. Treat it as sensitive:\n store with restrictive permissions (e.g., `chmod 600`), keep it inside the\n `migrate-to-msk-skill-artifacts/\u003ccluster_name>/` directory, and do not\n paste it into public channels, ticketing systems, or shared chat without\n redaction.\n- **Do not log raw config values.** The script's evidence detail strings\n include observed config values; this is intentional for human review but\n means logs containing this output should be retained under the same\n controls as the cluster-config input.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":23672,"content_sha256":"e1e428d5d1a13dd873014fe3893c2dcf5227cec5f88d16fede20419f4f9db6b6"},{"filename":"references/assessment-sizing.md","content":"# Assessment — Sizing\n\nSizing is delegated to the managing-amazon-msk Skill's pricing logic. This skill\nimplements no sizing math and packages no sizing script of its own. Derive the\nworkload inputs from the discovery contract, then use that skill's pricing logic\nto produce the recommended instance type, broker count, and monthly cost.\n\n> **Response format**: the assessment response template that covers both\n> compatibility AND sizing artifacts lives in\n> [`assessment-compatibility.md`](./assessment-compatibility.md) under\n> \"Response Template\". Do not invent a separate sizing-only response shape;\n> the user gets one combined response with both artifact paths.\n\n## Using managing-amazon-msk Skill's pricing logic (agent flow)\n\n1. **Load the managing-amazon-msk Skill.** How you reach its sizing script depends\n on how skills are available in this environment — the same MCP-vs-local split\n described in the guardrail section of\n [assessment-compatibility.md](./assessment-compatibility.md), applied to the\n other skill:\n - **AWS MCP available.** Call `retrieve_skill` for the `managing-amazon-msk`\n skill, then fetch `scripts/msk_sizing.py` through the same tool by passing\n `file=\"scripts/msk_sizing.py\"`, and run the content it returns.\n - **Skills installed locally.** Run the script from the local\n `managing-amazon-msk` skill directory, for example\n `uv run \u003cpath to skills>/managing-amazon-msk/scripts/msk_sizing.py ...`. Sibling\n skill directories live alongside this one (`.claude/skills/`,\n `~/.claude/skills/`, `.kiro/skills/`, etc.).\n2. **Read `target.rack_affined_consumers` and `target.pricing_discount_pct` from the\n discovery contract.** Ask the user only for the ones that are `null` or absent.\n Both are covered in the next two sections.\n3. **Derive the workload inputs** from `cluster-config.json` per \"Deriving the\n sizing inputs\" below. Two of the conversions are not one-to-one; getting them\n wrong produces a silently incorrect broker count.\n4. **Run sizing with `--broker-classes express`** to understand the options, sizing constraints, and cost explanations. `--broker-classes express` keeps Standard instance types out of the output entirely. Verify the script's \"Sizing inputs\" block matches the values you intended to pass before reading any broker count. Add `standard` to the flag only in the comparison cases listed below. Add `--discount-pct` and `--no-rack-affined-consumers` per the answers from step 2. Pass `--region` when the target is not `us-east-1` — see \"Region and pricing source\" below.\n5. **Read the recommendations.** Create an `msk_sizing_pricing.md` summary explaining the recommendation, inputs, sizing scenario, constraints, and cost-optimization tradeoffs for the workload. Record the rack-affinity assumption, any discount applied, and the `pricing source` value from the script's \"Sizing inputs\" block that produced the figures.\n\n## Deriving the sizing inputs\n\n`cluster-config.json` records throughput **per broker** and partitions as **leader\ncounts**. `msk_sizing.py` expects **total cluster** throughput and **total partition\nreplicas**. Apply these conversions — do not pass the contract values through\nunchanged.\n\n| `msk_sizing.py` flag | Derivation from `cluster-config.json` |\n|---|---|\n| `--peak-data-in-mbs` | `metrics.peak_bytes_in_per_broker_mbps` **× `topology.num_brokers`** |\n| `--peak-data-out-mbs` | `metrics.peak_bytes_out_per_broker_mbps` **× `topology.num_brokers`** |\n| `--avg-data-in-mbs` | `metrics.avg_bytes_in_per_broker_mbps` **× `topology.num_brokers`**. When the field is `null`, use `peak_in / 2` and say so in `msk_sizing_pricing.md`. |\n| `--avg-data-out-mbs` | `metrics.avg_bytes_out_per_broker_mbps` **× `topology.num_brokers`**. When the field is `null`, use `peak_out / 2` and say so in `msk_sizing_pricing.md`. |\n| `--num-partitions` | `sum(topics[].num_partitions)` **× 3**. The contract stores leader counts; the flag wants total replicas, and Express always uses RF=3. Never use the source cluster's own replication factor here. |\n| `--retention-hours` | `ceil(max(topics[].configs[\"retention.ms\"]) ÷ 3_600_000)` — the flag takes an integer hour count, so round up so retention is never under-estimated. Default to 24 when no topic declares it. |\n| `--primary-retention-hours` | Same value as `--retention-hours` — equal values disable Tiered Storage, which is correct for an Express target. If customer asks to compare Standard, and uses Tiered Storage today, pass the customer's Tiered Storage configuration. Only relevant for when customer asks for Standard comparison. |\n| `--replication-factor` | Omit. The default of 3 matches Express. |\n| `--broker-classes` | `express` (see \"Present Express only\"). |\n| `--discount-pct` | `target.pricing_discount_pct` (see \"Negotiated pricing\"). |\n| `--no-rack-affined-consumers` | Pass when `target.rack_affined_consumers` is `false` (see \"Consumer rack affinity\"). |\n| `--region` | The Region the customer is migrating into. Omit only when that is `us-east-1` (the script default). |\n| `--pricing-source` | Omit — the default handles both the credentialed and credential-free cases. See \"Region and pricing source\". |\n\nRecord the derived values — not the raw contract values — in\n`msk-sizing-inputs.\u003ccluster_name>.json`, so the sizing run is reproducible.\n\nIf `metrics` is absent entirely or `topology.num_brokers` is `null`, you cannot\nderive total throughput. Report that sizing could not be produced and surface the\ncompatibility `METRICS_MISSING` advisory. Do not substitute guessed throughput\nfigures, and do not pivot back into discovery to collect them.\n\n## Negotiated pricing\n\nRead `target.pricing_discount_pct` from `cluster-config.json`. A number is the\npercentage the customer stated (`0` means they confirmed no discount). When the field\nis `null` or absent, ask — alongside the rack affinity question if that one is also\nmissing:\n\n> Does your organization have a Private Pricing Agreement (PPA), Enterprise Discount\n> Program (EDP), or other negotiated AWS pricing? If so, what discount percentage\n> should I apply? Without one I will report public on-demand pricing.\n\nPass it as `--discount-pct N`, which applies the discount uniformly to brokers,\nstorage, Express data-in, and cross-AZ transfer. Broker counts do not change.\n\nDo not guess a percentage and do not apply one the customer did not state. When they\ndo not have a discount or do not know it, run at public pricing and include this note\nin `msk_sizing_pricing.md`:\n\n> **Note on pricing:** All cost estimates are based on AWS public on-demand pricing. If your organization has a Private Pricing Agreement (PPA), Enterprise Discount Program (EDP), or other negotiated pricing, your actual costs will differ. Contact your AWS account team for pricing that reflects your agreements.\n\nWhen a discount was applied, state the percentage in `msk_sizing_pricing.md` and tell\nthe customer to confirm it with their AWS account team before using the figures for\nbudgeting.\n\n## Region and pricing source\n\nPass `--region` for the Region the customer is migrating into; the script defaults to `us-east-1` and rates differ per Region.\n\nWith AWS credentials the script fetches live rates. Without them it falls back to an embedded `us-east-1` snapshot — expected here, since Assessment often has no credentials for the target account. Read the `pricing source` row in the \"Sizing inputs\" block and record it in `msk_sizing_pricing.md`; when it reads `static`, add this note:\n\n> **Note on pricing source:** These figures use `us-east-1` rates embedded in the sizing script and captured on *<date from the script banner>*, not a live pricing lookup. Re-run with AWS credentials available to confirm against current published rates.\n\nPricing a non-`us-east-1` Region needs credentials. Without them the script exits non-zero and prints nothing at all — not even a broker count, since pricing resolves before any broker math. Report that sizing could not be produced and ask the customer for credentials or a confirmed Region. Never re-run against `us-east-1` and present the result as the target Region's cost.\n\n## Consumer rack affinity\n\nCross-AZ consumer fetch is often one of the largest line items in the estimate. Read\n`target.rack_affined_consumers` from `cluster-config.json`. When it is `null` or\nabsent, ask before you run sizing — this and negotiated pricing are the only inputs\nAssessment may request (see \"Assessment scope — forbidden behavior\" in\n[assessment-compatibility.md](./assessment-compatibility.md)):\n\n> Will your consumers fetch from local-AZ replicas on the Express target — rack-aware\n> fetching, with `client.rack` set on each consumer and\n> `replica.selector.class=RackAwareReplicaSelector` on the cluster — or will they fetch\n> from partition leaders in any AZ?\n\nMap the contract value (or the answer) to the sizing flags:\n\n| `target.rack_affined_consumers` | Answer | Flag |\n|---|---|---|\n| `true` | Rack-aware / local-AZ fetching | Omit the flag — rack affinity is the script default |\n| `false` | Fetch from leaders in any AZ, or no rack configuration planned | `--no-rack-affined-consumers` |\n| `null` | Unknown | Run both ways |\n\nFor a self-managed source there may be no AZ topology to carry over, so treat this as\na decision about the target rather than a fact about the source. Ask whether they\nintend to configure rack-aware fetching on the Express cluster.\n\nWhen the answer is unknown, run sizing both ways, report the rack-affined figure as\nthe recommendation, and present the difference as avoidable cost. Recommend enabling\nrack-aware fetching on the target either way — see\n`references/configure-clients.md` in the `managing-amazon-msk` skill for\nthe `client.rack` and `replica.selector.class` settings.\n\nThis flag changes cost only, never broker count. State which assumption you used in\n`msk_sizing_pricing.md`.\n\n## Present Express only\n\nThis skill migrates workloads to Express. Report the Express pick; do not put Standard instance types, broker counts, or costs in `msk_sizing_pricing.md` or in your response.\n\nRun with `--broker-classes standard,express` only when the customer explicitly asks to compare classes, asks why not Standard, or hits a Standard-only condition. Refer to `references/size-and-choose-cluster.md` in the managing-amazon-msk skill, lead with the Express pick and present Standard as the alternative, giving the context from the decision framework on why Express is better than Standard even with cost differences.\n\nDo not volunteer Standard as a cost-optimization path, including Tiered Storage. Express storage is billed per GB-hour on data actually retained, which is the cost-effective option for nearly all workloads.\n\nIf the workload exceeds the broker quota at every Express size, recommend a quota increase, not Standard (which would also exceed broker quotas).\n\n## Compare the recommendation against the source footprint\n\nSource clusters are frequently over-provisioned — sized for peak plus a safety margin, then never scaled back. Recommend the smaller, lower-cost target when the workload supports it, and show the delta.\n\n1. Pull the source footprint from `cluster-config.json`: `topology.num_brokers`, `topology.num_azs`, `topology.broker_instance_type`.\n2. Compare it against the Express recommendation. Recommend the Express pick even when it is materially smaller than the source. The script already models peak throughput, 1-AZ-down headroom, and partition limits, so a smaller broker count is not a reduction in resilience.\n3. Explain why the target is smaller in terms the customer can verify: up to 3x ingress per broker, storage billed on retained data instead of provisioned EBS, and capacity for replication and rebalancing reserved by the service rather than bought as spare nodes.\n4. State what you could not verify. Node-for-node comparison needs `topology.broker_instance_type`, which is optional in the discovery contract. Without it, compare broker counts only and say so.\n\nDo not inflate the recommendation to match the source footprint. If the customer wants headroom beyond what the script models, name the recommended pick first and the larger option second, with the cost difference.\n\n## Caveats\n\n1. **Average throughput is optional in the discovery contract.** The sizing script\n uses average throughput for storage volume and cost projection (e.g. cross-AZ\n data transfer). When `metrics.avg_bytes_*_per_broker_mbps` is `null`, the\n `peak / 2` fallback from the derivation table applies and the cost figures are\n correspondingly rough. Real average values give more accurate cost projections;\n broker count is driven by peak either way.\n\n2. **Retention is per-topic in the discovery contract; the skill takes one\n number.** The skill emits the max over topics as an upper-bound storage\n estimate. Override by passing a smaller `--retention-hours` (derived per the\n table above) if the source has a small set of low-retention topics dominating\n the storage picture, and say so in `msk_sizing_pricing.md`. Express storage is billed\n per GB-hour on retained data, so right-sizing per-topic retention is the\n storage cost lever — not a change of broker class.\n\n3. **Pricing is public on-demand, and `us-east-1` unless you pass `--region`.**\n Apply a customer-stated Private Pricing Agreement (PPA) or Enterprise Discount\n Program (EDP) discount with `--discount-pct` — see \"Negotiated pricing\" above.\n Rates may also come from an embedded snapshot rather than a live lookup — see\n \"Region and pricing source\" below.\n\n## Key recommendations\n\n- **Express recommendations**: Broker instance size (express.m7g.large through\n express.m7g.16xlarge) and count\n- **Bottleneck breakdown**: per instance, which\n constraint (ingress / egress / partitions / storage) drives the broker\n count.\n\n## Refresh procedure\n\nWhen new skill updates are released, pricing logic can be re-run with the managing-amazon-msk Skill's pricing logic.\n\n## Security considerations\n\n- **Discovery input may contain credentials.** The sizing inputs are derived from\n `cluster-config.json`, whose `security` block may carry credential material\n (SASL passwords, API keys, TLS private keys) if discovery captured more than the\n authentication mechanism. Verify the input has none before deriving inputs or\n sharing the file; redact anything resembling a credential.\n- **Sizing inputs reveal capacity and topology details.** The\n `msk-sizing-inputs.\u003ccluster_name>.json` and `msk_sizing_pricing.md` artifacts\n contain peak throughput, partition count, retention, and the recommended\n broker count and instance type for the workload. Treat them as sensitive —\n these details are useful inputs for targeted attacks. Do not share via\n unencrypted email, public channels, or public ticketing systems without\n redaction.\n- **Store with restrictive permissions.** Keep the artifact inside\n `migrate-to-msk-skill-artifacts/\u003ccluster_name>/` and apply restrictive\n permissions (e.g., `chmod 600`) appropriate for your environment.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":15130,"content_sha256":"cca138be714932e0948d41960971a39c6d6b80eae37e451b284d9a8b24910a57"},{"filename":"references/discovery.md","content":"# Discovery Phase\n\nProcess ONE cluster at a time. If multiple clusters are found in the IaC files,\nlist them and ask the user which one to process first. Do NOT process all clusters\nat once.\n\nYour response MUST follow the template exactly.\n\n## Input methods\n\nUse whichever of these the customer's input supports; they combine.\n\n1. **IaC analysis** — Read infrastructure files and extract cluster metadata.\n2. **Kafka CLI commands** — Display standard Kafka CLI commands for the customer to\n run on their cluster (`kafka-topics.sh`, `kafka-configs.sh`,\n `kafka-broker-api-versions.sh`). Do NOT generate or offer Python scripts.\n3. **Runtime metrics intake** — Ingest metrics provided by the customer.\n4. **Manual conversation** — Ask the customer for cluster details.\n5. **Target inputs** — Ask the customer whether consumers will use rack-aware\n (local-AZ) fetching on the target and whether they have a PPA or enterprise\n discount to apply. Record both in the `target` block; record `null` for\n \"unknown\". Never guess a discount percentage.\n\nFORBIDDEN content — do NOT include any of the following:\n\n- Compatibility observations (\"not supported by MSK\", \"should migrate smoothly\")\n- Blockers, warnings, or recommendations\n- Migration steps or deployment commands\n- Sections called \"Key Observations\", \"Important Notes\", \"Next Steps\", or similar\n- Any assessment of whether configs will work on Express\n- Python scripts\n\n## Response Template\n\n```\n## Discovery Complete — \u003ccluster_name>\n\n### Kafka\n- **Version:** \u003cversion>\n- **Coordination Mechanism:** \u003cKRaft | ZooKeeper | Unknown>\n\n### Topology\n- **Brokers:** \u003ccount>\n- **Availability Zones:** \u003ccount or \"unknown\">\n- **Instance Type:** \u003ctype or \"not determined\">\n\n### Security\n- **Authentication:** \u003cmethod>\n- **Encryption in transit:** \u003cTLS | PLAINTEXT | TLS_PLAINTEXT | UNKNOWN>\n\n### Topics Found\n| Topic | Partitions | Replication | Non-default configs |\n|-------|-----------|-------------|---------------------|\n| \u003cname> | \u003ccount> | \u003cfactor> | \u003cconfigs or \"none\"> |\n\n### Non-default Broker Configs\n- `\u003cconfig.key>`: `\u003cvalue>`\n\n### Target Inputs\n- **Consumer rack affinity:** \u003clocal-AZ (rack-aware) fetching | leaders in any AZ | unknown>\n- **Negotiated pricing discount:** \u003cN% | none | unknown>\n\n---\n\n### Information I Could Not Determine\n\nThe following require runtime data that isn't available in IaC:\n- Topics, partitions, replication factors, and topic-level configs\n- Broker-level configuration overrides\n- Peak throughput (MB/s ingress/egress per broker)\n- Connection count per broker\n\n**Options to fill these gaps:**\n- **A.** Run these commands on your cluster and share the output files:\n\n ```bash\n kafka-topics.sh --bootstrap-server \u003caddr> --describe > kafka-topics-output.txt\n kafka-configs.sh --bootstrap-server \u003caddr> --entity-type brokers --entity-default --describe > kafka-configs-output.txt\n kafka-broker-api-versions.sh --bootstrap-server \u003caddr> > kafka-versions-output.txt\n ```\n\n Then share the files or paste their contents here.\n\n- **B.** Proceed with partial data (assessment will be less accurate)\n\n### Target Inputs I Need From You\n\nThese two are choices about the target cluster, not properties of your source, and\nthey drive the cost estimate:\n\n1. **Consumer rack affinity** — will your consumers fetch from local-AZ replicas\n (rack-aware fetching, with `client.rack` set on each consumer and\n `replica.selector.class=RackAwareReplicaSelector` on the cluster), or will they\n fetch from partition leaders in any AZ?\n2. **Negotiated pricing** — does your organization have a Private Pricing Agreement\n (PPA), Enterprise Discount Program (EDP), or other negotiated AWS pricing? If so,\n what discount percentage should I apply? Without one I will use public on-demand\n pricing.\n\nAnswer \"unknown\" to either and I will record it as unknown.\n\nWould you like to proceed to assessment, or provide additional information first?\n\n```\n\n## Rules\n\n- Fill in values from IaC files/commands. Remove sections where no data was found.\n- Save to `migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json` for the selected cluster.\n- Do NOT add extra sections beyond what the template shows.\n- Do NOT summarize, characterize, or editorialize the findings (\"solid foundation\",\n \"straightforward migration\", etc.). Only state facts.\n- Do NOT mention Express, MSK, compatibility, blockers, or migration steps. The two\n Target Inputs questions are the sole exception: they may refer to the target\n cluster, since they are decisions about it. Ask them as written — no compatibility\n commentary, no cost figures, no recommendation about which option to pick.\n- ALWAYS ask the two Target Inputs questions unless the user already supplied both\n answers. Record what they said in the `target` block; record `null` for anything\n they answered \"unknown\" or did not answer. Never guess a discount percentage and\n never infer rack affinity from the source cluster's AZ layout.\n- Skip the \"Target Inputs I Need From You\" section only when both answers are\n already in hand; the `### Target Inputs` section of the report is always present.\n- Do NOT offer or generate Python scripts. Only show Kafka CLI commands.\n- Do NOT reference CLI commands unless you actually displayed them in the response.\n- If the user provided all information manually and there are no gaps, skip the\n \"Information I Could Not Determine\" section entirely. Keep the Target Inputs\n questions if either answer is still missing, then end with:\n \"Would you like to proceed to assessment, or provide additional information first?\"\n\n---\n\n## JSON Output Schema\n\nSave to the following path:\n`migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json`\n\nWhere `\u003ccluster_name>` is the cluster name (lowercase, hyphenated).\nCreate the directories if they don't exist.\n\nALL fields MUST be present in the output file unless marked as `(optional)`.\nFields marked `(optional)` MAY be omitted entirely. For all other fields, if the\nvalue cannot be determined, use `null` for strings/numbers, `false` for booleans,\n`[]` for arrays, and `{}` for objects.\n\nThe output file contains sensitive data (broker addresses, authentication details).\nTreat it accordingly — do not commit to version control or share in public channels.\nDo NOT store passwords, private keys, or secret values in this file. For\n`auth_identity`, record only the username or a Secrets Manager ARN reference —\nnever the password or key material.\n\nNote: `broker_configs` and `topics[].configs` carry the **full** Kafka config\ndump (every config the source exposed), not just non-default values.\n`compatibility.py` filters against per-Kafka-version Apache defaults internally,\nso values matching the default for the source's `kafka.version` produce no\nevidence — only divergences from default are evaluated against Express's\nconstraints.\n\n### `security` enum values\n\n`encryption_in_transit` is a closed enum describing how client-broker traffic\nis encrypted on the source:\n\n- `TLS` — clients connect on a TLS-only listener (port 9094 or 9096 typically).\n- `PLAINTEXT` — clients connect on a plaintext listener (port 9092 typically).\n- `TLS_PLAINTEXT` — the cluster exposes both; some clients use TLS, others plaintext.\n- `UNKNOWN` — use when the listener configuration cannot be determined from available IaC or CLI output.\n\nTo determine the value: check the broker's `listeners` / `advertised.listeners`\nconfig, or look at the `security.protocol` clients use (`SSL` / `SASL_SSL` →\nTLS; `PLAINTEXT` / `SASL_PLAINTEXT` → PLAINTEXT; mix → TLS_PLAINTEXT).\n\n`authentication` is a closed enum covering the mechanism the source cluster\nexpects from Kafka clients:\n\n- `UNAUTHENTICATED` — no `sasl.mechanism` or `ssl.keystore` configured on clients; the broker allows anonymous connections.\n- `TLS` — clients present X.509 certificates (`ssl.keystore.location` configured); broker has `ssl.client.auth=required`.\n- `SASL_SCRAM` — `sasl.mechanism=SCRAM-SHA-256` or `SCRAM-SHA-512` in client config.\n- `SASL_IAM` — `sasl.mechanism=AWS_MSK_IAM`, or clients use the AWS MSK IAM signer with `sasl.mechanism=OAUTHBEARER` on an existing MSK cluster.\n- `SASL_OAUTHBEARER` — `sasl.mechanism=OAUTHBEARER` with a **non-AWS** token provider (e.g. Keycloak, Okta, custom OAuth server).\n- `OTHER` — any mechanism not covered above (e.g. `GSSAPI`/Kerberos, `PLAIN`, custom callback handlers).\n- `UNKNOWN` — use when the mechanism cannot be determined from available IaC or CLI output.\n\nTo determine the value: check the client's `sasl.mechanism` property, or the\nbroker's `sasl.enabled.mechanisms` / `listener.security.protocol.map` config.\nIf both `OAUTHBEARER` and the AWS signer library are present, use `SASL_IAM`\n(it's the AWS IAM path). Use `SASL_OAUTHBEARER` only for custom providers.\n\nFor how each value is evaluated against MSK Express, see\n[assessment-compatibility.md](./assessment-compatibility.md) (Pillar 4 — Auth).\n\nDiscovery MUST emit one of these exact strings. `compatibility.py`'s\n`validate_input` rejects unrecognized values.\n\n### `target` block values\n\nBoth fields are user-stated decisions about the target cluster, not measurements of\nthe source. Discovery asks for them; Assessment reads them and passes them to the\nsizing script.\n\n`rack_affined_consumers`:\n\n- `true` — consumers will fetch from local-AZ replicas (`client.rack` on each\n consumer, `replica.selector.class=RackAwareReplicaSelector` on the cluster).\n- `false` — consumers will fetch from partition leaders in any AZ, or no rack\n configuration is planned.\n- `null` — the user answered \"unknown\" or was not asked.\n\n`pricing_discount_pct`:\n\n- A number in `[0, 100)` — the discount percentage the user stated (`0` means they\n confirmed no negotiated discount).\n- `null` — the user answered \"unknown\" or was not asked.\n\nRecord only what the user stated. Do not derive `rack_affined_consumers` from\n`topology.num_azs` or the source's rack configuration, and do not supply a discount\npercentage the user did not give.\n\n### Partition counts: leaders vs. total replicas\n\nTwo different partition numbers come up, and they are not interchangeable. Be\nexplicit about which one you are recording.\n\n- **Configured (leader) partitions** — the partition count set on a topic, one\n leader per partition. This is the `PartitionCount` shown by\n `kafka-topics.sh --describe`, and the number set with `--partitions`. It does\n **not** include replicas.\n- **Total partition replicas** — configured partitions multiplied by the\n replication factor (leaders + followers). This is the basis AWS uses for\n per-broker partition limits (see `peak_partitions_per_broker` below and the\n MSK Express broker partition quota).\n\nHow each contract field is counted:\n\n- `topics[].num_partitions` — the **configured (leader)** count for that topic.\n Record the per-topic partition count, never a pre-multiplied total.\n- `topics[].replication_factor` — the source topic's replication factor.\n- `metrics.peak_partitions_per_broker` — **total replicas** (leaders +\n followers) hosted on the busiest broker, matching the AWS quota basis.\n\nAssessment converts when it needs the total: it multiplies the summed leader count\nby the Express target replication factor (always 3) before passing\n`--num-partitions` to the sizing script (see \"Deriving the sizing inputs\" in\n[assessment-sizing.md](./assessment-sizing.md)). So you only ever enter leader\ncounts in `num_partitions` — do not pre-multiply by RF.\n\n**If a user reports a partition number conversationally and it is ambiguous\nwhich count they mean, ask before recording it:** \"Is that the configured\npartition count per topic (leaders), or the total including replicas?\" If it is\na total-including-replicas figure, divide it back to the configured count (and\ncapture the replication factor separately) before writing `num_partitions`.\n\n```json\n{\n \"cluster_name\": \"\u003cstring>\",\n \"source_type\": \"\u003cstring: 'self-managed' | 'cloud-hosted'>\",\n \"discovered_at\": \"\u003cstring: ISO 8601 timestamp>\",\n\n \"kafka\": {\n \"version\": \"\u003cstring>\",\n \"coordination_mechanism\": \"\u003cstring: 'KRaft' | 'ZooKeeper' | 'Unknown'>\"\n },\n\n \"topology\": {\n \"num_brokers\": \"\u003cinteger>\",\n \"num_azs\": \"\u003cinteger> (optional)\",\n \"broker_instance_type\": \"\u003cstring> (optional)\"\n },\n\n \"topics\": [\n {\n \"name\": \"\u003cstring>\",\n \"num_partitions\": \"\u003cinteger>\",\n \"replication_factor\": \"\u003cinteger>\",\n \"configs\": {\n \"\u003cconfig.key>\": \"\u003cstring: full Kafka topic config dump>\"\n }\n }\n ],\n\n \"broker_configs\": {\n \"\u003cconfig.key>\": \"\u003cstring: full Kafka broker config dump>\"\n },\n\n \"security\": {\n \"encryption_in_transit\": \"\u003cenum: 'TLS' | 'PLAINTEXT' | 'TLS_PLAINTEXT' | 'UNKNOWN'>\",\n \"authentication\": \"\u003cenum: 'UNAUTHENTICATED' | 'TLS' | 'SASL_SCRAM' | 'SASL_IAM' | 'SASL_OAUTHBEARER' | 'OTHER' | 'UNKNOWN'>\",\n \"auth_identity\": \"\u003cstring> (optional)\"\n },\n\n \"target\": {\n \"rack_affined_consumers\": \"\u003cboolean | null>\",\n \"pricing_discount_pct\": \"\u003cnumber | null>\"\n },\n\n \"metrics\": {\n \"source\": \"\u003cstring: 'manual' | 'other' | null>\",\n \"lookback_hours\": \"\u003cinteger | null>\",\n \"peak_bytes_in_per_broker_mbps\": \"\u003cnumber | null>\",\n \"peak_bytes_out_per_broker_mbps\": \"\u003cnumber | null>\",\n \"avg_bytes_in_per_broker_mbps\": \"\u003cnumber | null>\",\n \"avg_bytes_out_per_broker_mbps\": \"\u003cnumber | null>\",\n \"peak_connections_per_broker\": \"\u003cinteger | null>\",\n \"peak_partitions_per_broker\": \"\u003cinteger | null>\"\n },\n\n \"iac_index\": [\n {\n \"iac_type\": \"\u003cstring: 'terraform' | 'cdk' | 'cloudformation' | 'k8s_manifest' | 'docker_compose'>\",\n \"iac_file\": \"\u003cstring: relative path>\",\n \"service_name\": \"\u003cstring>\",\n \"bootstrap_hints\": [\"\u003cstring>\"]\n }\n ]\n}\n```\n\n### Example\n\n```json\n{\n \"cluster_name\": \"orders-staging\",\n \"source_type\": \"self-managed\",\n \"discovered_at\": \"2026-05-27T18:00:00Z\",\n \"kafka\": {\n \"version\": \"3.6.0\",\n \"coordination_mechanism\": \"KRaft\"\n },\n \"topology\": {\n \"num_brokers\": 12,\n \"num_azs\": 3,\n \"broker_instance_type\": null\n },\n \"topics\": [\n {\n \"name\": \"orders\",\n \"num_partitions\": 24,\n \"replication_factor\": 3,\n \"configs\": {\n \"cleanup.policy\": \"delete\",\n \"retention.ms\": \"604800000\"\n }\n }\n ],\n \"broker_configs\": {\n \"log.segment.bytes\": \"536870912\",\n \"min.insync.replicas\": \"3\"\n },\n \"security\": {\n \"encryption_in_transit\": \"TLS\",\n \"authentication\": \"SASL_SCRAM\",\n \"auth_identity\": \"kafka-admin\"\n },\n \"target\": {\n \"rack_affined_consumers\": true,\n \"pricing_discount_pct\": 15\n },\n \"metrics\": {\n \"source\": \"manual\",\n \"lookback_hours\": 24,\n \"peak_bytes_in_per_broker_mbps\": 142.7,\n \"peak_bytes_out_per_broker_mbps\": 285.3,\n \"avg_bytes_in_per_broker_mbps\": 78.4,\n \"avg_bytes_out_per_broker_mbps\": 156.8,\n \"peak_connections_per_broker\": 1240,\n \"peak_partitions_per_broker\": 380\n },\n \"iac_index\": [\n {\n \"iac_type\": \"k8s_manifest\",\n \"iac_file\": \"deploy/orders.yaml\",\n \"service_name\": \"order-svc\",\n \"bootstrap_hints\": [\"kafka:9092\"]\n }\n ]\n}\n```\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":15118,"content_sha256":"eb325597384cbfe990d1aaa98d4471e86e422ce7db819ddd73c49180d3e2b507"},{"filename":"references/simulation.md","content":"# Phase 3 — MSK Express Load-Test Simulation\n\nDeploy a complete MSK Express load-testing environment in the customer's AWS\naccount so they can validate Express performance before cutover. The simulation is\n**optional** and is the final phase of the skill. Declining at any decision point\nends the skill — there is no subsequent phase.\n\nYou orchestrate the conversation and run AWS CLI commands (CloudFormation, SSM,\nEC2). The AWS MCP server is recommended for executing these commands but is not\nrequired — the skill works with plain AWS CLI as well. Two deterministic\nartifacts do the rest — never hand-compute or hand-edit them:\n\n- [`scripts/simulation_load_test_config.py`](../scripts/simulation_load_test_config.py) — sizing math,\n template rendering, and guardrail validation (`compute`, `render`, and\n `validate` actions).\n- [`assets/simulation-stack.yaml`](../assets/simulation-stack.yaml) — a **static**\n parameterized CloudFormation template. It is the internal source: never edit it,\n and never hand the customer its install path. There are two deploy paths:\n - **You deploy it (customer consents):** deploy in one `create-stack` straight\n from this source template, passing the compute-derived `--parameters`. No render.\n - **Customer deploys it themselves (self-guided):** `render` reads this template\n internally and writes a **filled local copy** (the customer's sizing and the\n computed fleet baked in as parameter `Default`s) to the customer's working\n directory; the customer deploys from that local copy with no `--parameters`.\n To resize, redeploy with new parameters (or re-render, on the self-guided path).\n\n## Never expose the skill source path\n\nThe customer must never see, depend on, or be told to `cd` into the skill install\ndirectory. You (the agent) run the scripts and — on the consented path — the\n`create-stack` from wherever the skill lives, but **nothing you hand the customer**\n(a file to keep, a self-guided command, teardown) may name a skill-package path\n(anything under the skill directory — `assets/…`, `scripts/…`, or the symlinked\n`~/.claude/skills/…`). On the self-guided path, the only template path the customer\nsees is the **local rendered artifact** in their working directory (the absolute\n`output_path` that `render` reports). The customer should never see or depend on\nwhere the skill is installed.\n\n## Contents\n\n- [Flow](#flow)\n- [Confirm the target account](#confirm-the-target-account)\n- [Single simulation per account](#single-simulation-per-account)\n- [Constrained choices](#constrained-choices)\n- [Communicating with the customer](#communicating-with-the-customer)\n- [Sizing inputs](#sizing-inputs)\n- [Install the Kafka client](#install-the-kafka-client)\n- [Env computation](#env-computation)\n- [Render the local template](#render-the-local-template)\n- [Throughput & fleet reference](#throughput--fleet-reference)\n- [Deploy](#deploy)\n- [Test templates](#test-templates)\n- [Guardrails](#guardrails)\n- [Triggering a test](#triggering-a-test)\n- [Teardown](#teardown)\n- [Off-trail handling](#off-trail-handling)\n\n## Flow\n\n1. Introduce the simulation and what it does; ask consent. **No → end the skill.**\n Mention that only **one simulation can exist per account at a time** (to avoid\n duplicate test costs).\n2. Gather the target AWS region. Then **confirm the AWS account** (see\n [Confirm the target account](#confirm-the-target-account)): run\n `aws sts get-caller-identity`, show the account id + role, and ask the customer to\n confirm it is the account they intend to deploy into. **Do not run any\n create/deploy/delete action until the customer confirms the account.** If they say\n it is the wrong account, stop and let them switch credentials first.\n3. **Check for an existing simulation** (see [Single simulation per account](#single-simulation-per-account)).\n Only one may exist per account/region. If one already exists, tell the customer\n and ask whether to **reuse** it (present its dashboard, then **verify the Kafka\n client** before tests — see below) or **delete and redeploy** (tear it down and\n wait, then continue). If none exists, continue.\n - On **reuse**, a simulation that is `CREATE_COMPLETE` does **not** guarantee a\n working Kafka client — the client is installed as a post-deploy step, so a\n reused simulation may have an empty `/opt/kafka` (e.g. install never ran, or ran\n against a different client). Before offering tests, run `ValidateClient` (see\n [Install the Kafka client](#install-the-kafka-client)). If it **passes**, go to\n the test options at step 8. If it **fails** with a \"client not installed/working\"\n (or version) error, run the install + validate loop (have the customer install a\n Kafka client themselves — a `kafkaUrl` + `kafkaSha512` they provide; you may point\n them to Apache in words but don't source the URL or run it yourself — then re-run\n `ValidateClient` until it passes)\n — then go to step 8. Do **not** present test options until\n `ValidateClient` passes.\n4. Ask the customer for cluster sizing (see [Sizing inputs](#sizing-inputs)) —\n `instance_type`, `broker_count`, and `kafka_version` + mode. Offer only\n supported choices (see [Constrained choices](#constrained-choices)). Mention\n once, in passing, that the skill's assessment phase (Phase 2) inventories their\n source cluster and produces a sizing/pricing workbook to help pick the right\n Express size — so if they have not run it and want a data-driven\n recommendation, they can run the assessment first. Do not block on it; proceed\n with whatever sizing they give.\n5. Run `simulation_load_test_config.py compute` → derive fleet + CloudFormation parameters.\n Do **not** render yet — render is only for the self-guided path below.\n (The stack downloads no Kafka client; the client is installed after deploy — see\n [Install the Kafka client](#install-the-kafka-client).)\n6. Present the stack resource list and **inform there will be extra cost** and it\n takes ~1 hour. Ask the customer to confirm deployment.\n - **Yes (you deploy) →** go to step 7. Deploy in **one** `create-stack` command\n from the static source template, passing the compute-derived values as\n `--parameters` (see [Deploy](#deploy)). Do **not** render a local template on\n this path — passing parameters to the source template is the single deploy step.\n - **No (self-guided) →** *now* run `simulation_load_test_config.py render` (see\n [Render the local template](#render-the-local-template)) to write a filled,\n ready-to-deploy template to the customer's working directory, then hand them the\n self-guided `create-stack` command that deploys from **that local file** (no\n `--parameters` needed — sizing is baked in) plus the install/validate/test\n commands. End the skill. Rendering exists so the customer who deploys themselves\n gets a single self-contained file rather than a long parameterized command\n pointing at a path they don't control.\n7. Deploy the stack with one `create-stack` (source template + `--parameters` from\n compute); wait for `CREATE_COMPLETE`. Then **install and validate the Kafka client**\n (see [Install the Kafka client](#install-the-kafka-client)): install the Kafka\n client on the fleet (the `kafkaUrl` comes from the customer — you don't source the\n URL), then run `ValidateClient` and loop\n until it passes. Then create the topics via the\n `CreateTopicsCommand` output. Once that\n is done, present the CloudWatch dashboard to the customer **once** —\n both the `DashboardUrl` and the `DashboardArn` stack outputs — and tell them this\n is where every test's metrics will appear.\n8. Present the two test options (below). **Before the options, state the deployed\n cluster's ingress capacity** so the customer can pick a sensible throughput:\n its **sustained** ingress (the recommended ceiling, no degradation up to here)\n and its **maximum** ingress (the hard quota — MSK throttles beyond it). Both come\n from the `compute` output (`cluster_sustained_ingress_mbps` and\n `cluster_max_ingress_mbps`); never hand-compute them.\n9. Fill the chosen template with computed defaults + bounds; ask for edits.\n10. Run `simulation_load_test_config.py validate`. **REJECT → show errors, loop back to 9.**\n **PASS →** surface any warnings as one sentence each and continue.\n11. Trigger the test via SSM and confirm it started, and name the metric(s) it\n emits. The dashboard was already shared at deploy — point the customer back to it\n rather than re-posting the URL each run. **Tell the customer that after the\n producers stop, the consumers keep reading and adaptively drain until the\n consumer-lag metric returns to ~0 (capped at `maxDrainMinutes`, default 20) —\n so the run takes a little longer than `duration_minutes` to fully settle, and\n that trailing drain is expected.**\n12. Ask the customer which of these they want next:\n - **Keep the stack running and finish** — leave the simulation in place and end\n the skill. Remind them it keeps incurring cost until they tear it down, and\n give them the teardown command for later.\n - **Run another test** → go to 8.\n - **Tear down the stack** → [Teardown](#teardown), then end the skill.\n\n## Confirm the target account\n\nBefore the first AWS action that creates, deploys, modifies, or deletes anything,\nconfirm you are operating in the account the customer intends:\n\n```bash\naws sts get-caller-identity --query '{Account:Account,Arn:Arn}' --output json\n```\n\nShow the customer the **account id** and **role/identity**, and ask them to confirm\nit is the correct account for this simulation. Do **not** proceed past this point until\nthey explicitly confirm. If it is the wrong account, stop — ask them to switch\ncredentials (profile / refreshed session) and re-run the check. Re-confirm whenever\ncredentials are refreshed mid-session, since the identity may have changed.\n\n## Single simulation per account\n\nTo avoid paying for duplicate load-test infrastructure, **only one simulation may exist\nper account/region at a time**. The stack name is fixed (`msk-express-simulation`), so a\nsecond `create-stack` fails with `AlreadyExistsException`. Always check before\ndeploying, and tell the customer up front that only one simulation is allowed at a time:\n\n```bash\naws cloudformation describe-stacks --stack-name msk-express-simulation \\\n --region \u003cregion> --query 'Stacks[0].StackStatus' --output text 2>/dev/null\n```\n\n- **No stack** (command errors / empty) — proceed with a fresh deployment.\n- **`CREATE_COMPLETE`** — a healthy simulation already exists. Inform the customer (and\n that only one is allowed), then ask which they want:\n - **Reuse it** — read `DashboardUrl` + `DashboardArn` (and `FleetSummary`) from\n `describe-stacks --query 'Stacks[0].Outputs'` and present the dashboard. No new\n deploy, no new cost. **Then verify the Kafka client before offering tests:** a\n `CREATE_COMPLETE` stack does not imply an installed client (it is a post-deploy\n step and may never have run, or ran against a different client), so run\n `ValidateClient` (see [Install the Kafka client](#install-the-kafka-client)).\n If it passes, go to the test options. If it fails, run the install + validate\n loop (`InstallKafkaClient` with a Kafka client URL, then\n `ValidateClient` until it passes) first. Do not offer tests until it passes.\n - **Delete and redeploy** — `delete-stack` + `wait stack-delete-complete`, then\n continue with sizing + deploy.\n- **Any other state** (`ROLLBACK_COMPLETE`, `*_FAILED`, `*_IN_PROGRESS`) — it cannot\n be reused. If it is settled in a failed/rollback state, delete and redeploy; if an\n operation is still in progress, wait for it to finish before deciding.\n\n## Constrained choices\n\nWhen asking the customer to choose cluster parameters, offer **only the supported\nvalues** as an explicit list. The question UI always appends its own \"Other\"\nfree-form option that cannot be suppressed — so do not rely on the prompt to\nconstrain input. Instead, treat any free-form value as untrusted and run it\nthrough `simulation_load_test_config.py compute` (which rejects unsupported `instance_type`,\n`broker_count`, and `kafka_version` locally) before deploying. Never deploy a\nvalue the script rejected — an unsupported value otherwise only fails later at\n`create-stack`.\n\n**Picker cap.** The `AskUserQuestion` tool hard-limits each question to 4\noptions (plus the auto-appended \"Other\") and silently drops the rest. There are\n**7 supported `instance_type` values**, more than the picker can show. You may\nstill use the picker, but the **question text MUST enumerate all 7 instance\ntypes** (with their specs) so the customer can see every option even when some are\nnot rendered as buttons — instruct them to use \"Other\" to pick a type that is\nnot shown as a button. Never present `instance_type` in a way that names fewer\nthan all 7 types. `broker_count` and `kafka_version` have ≤ 4 options each and\nfit the picker directly.\n\n- **instance_type** — exactly these Express types (no others), always present\n all seven:\n `express.m7g.large`, `express.m7g.xlarge`, `express.m7g.2xlarge`,\n `express.m7g.4xlarge`, `express.m7g.8xlarge`, `express.m7g.12xlarge`,\n `express.m7g.16xlarge`.\n- **broker_count** — a multiple of 3, ≥ 3 (Express is always a 3-AZ topology).\n- **kafka_version + metadata mode** — present as one combined list (mode is implied\n by the choice). Express supports **only 3.6, 3.8, 3.9**. On Express **3.9 is\n KRaft-only** and **KRaft is not available below 3.9**, so the only valid combos\n are these three:\n\n | Choice | Mode | CloudFormation `KafkaVersion` |\n |---|---|---|\n | 3.9 (KRaft) — recommended | KRaft | `3.9.x.kraft` |\n | 3.8 (ZooKeeper) | ZooKeeper | `3.8.x` |\n | 3.6 (ZooKeeper) | ZooKeeper | `3.6.0` |\n\n Do **not** offer **3.9 in ZooKeeper mode** — it is not a supported Express\n combination and only fails later at `create-stack`. KRaft is not available below\n 3.9; 3.7 and 4.0 are not Express versions at all. Do not offer any of these.\n\nTreat the official AWS MSK Express documentation as the source of truth for the\nsupported instance types and versions. If you can access AWS documentation, verify\nthese values against it — especially if a value here is rejected at `create-stack`.\nIf that documentation is not accessible, this list is a trustworthy source to use.\n\n## Communicating with the customer\n\nSpeak to the customer in plain, outward-facing language. Do **not** expose internal\nor process framing — no \"per the v1 flow\", \"the skill says\", \"I won't interpret\nresults because…\", version numbers of this workflow, step numbers, or references to\nthese instructions. Just do the right thing.\n\nFrame the simulation as a way to see MSK Express perform on the customer's own\nworkload — a confidence-building demonstration, not a safeguard. MSK Express is\nproduction-ready; present the simulation as confirming expected performance on their\nspecific workload, and avoid phrasing that implies doubt about Express (e.g. \"just to\nbe safe before committing\" or \"in case Express can't keep up\"). Let the results speak\nfor themselves.\n\nWhen presenting test results: point the customer to the dashboard already shared at\ndeploy and name the metric(s) the test emits, and let them read it. If they ask what\nthe numbers mean, you may describe what each metric represents, but do not assert a\npass/fail verdict — phrase it as something they evaluate against their own targets.\n\nWhen you preview or list the upcoming steps, keep each to a short phrase (e.g.\n*Install the Kafka client on the fleet*) — do not inline the install options or other\ndetails; cover those at the step itself.\n\n## Sizing inputs\n\nAlways ask the customer for the sizing directly — `instance_type`, `broker_count`,\nand a `kafka_version` + metadata mode — offering only the supported values listed\nin [Constrained choices](#constrained-choices). Do not look for a local\nassessment file; the assessment now hands the customer a sizing/pricing workbook\nrather than writing a JSON artifact this skill can read.\n\nWhen you ask, briefly note that the skill's assessment phase (Phase 2) inventories\ntheir source cluster and produces that sizing/pricing workbook, so if they have\nnot run it and want a data-driven Express size recommendation, they can run the\nassessment first. Keep it to a sentence and do not block on it — if they already\nhave a size in mind (from the workbook or otherwise), take it and proceed.\n\nValidate: `instance_type` must be one of the supported Express types; `broker_count`\n≥ 3 and a multiple of 3; the version/mode must be one of the three supported\ncombinations. Map the customer's version + mode choice to the exact CloudFormation\n`KafkaVersion` string from the table. 3.9\nis always KRaft on Express (`3.9.x.kraft`); reject a request for 3.9 in ZooKeeper\nmode.\n\n## Install the Kafka client\n\nThe Kafka client is installed after the stack is created, then validated before any\ntest. At `CREATE_COMPLETE` the fleet has Java, an empty `/opt/kafka`, and a\n`client.properties` with the IAM/TLS settings Express requires; this step installs\nthe client itself. Two SSM documents handle it:\n\n- **`InstallKafkaClient`** — installs the Kafka client from `kafkaUrl` (passed at\n invocation, verified against the customer-provided `kafkaSha512`), plus the\n `aws-msk-iam-auth` jar (required for IAM auth) from its AWS-owned source (installed\n automatically — no parameter).\n- **`ValidateClient`** — confirms the client runs, its version is `>=` the cluster's\n Kafka version, and it can authenticate and list topics on the cluster.\n\n### Step 1 — install\n\nInstall the Kafka client on the fleet. The tarball URL is the runtime parameter\n`kafkaUrl`, and **it comes from the customer** — they pick a source they trust and\ngive you the URL; you plug it in. Help *in words*: the Apache Software Foundation\npublishes Kafka — point them to a source they trust (their org's approved software\nchannel, their package manager, or the official Apache distribution on an\n`apache.org` domain) and away from third-party mirrors or search-result links. **Do\nnot fabricate, guess, or web-search a URL and present it as authoritative** — you may\nfill in a URL the customer provides. The customer also gives the tarball's\n**SHA-512** (`kafkaSha512`); the install verifies it before use (copy the value Apache\npublishes, lowercased/no spaces, or compute `sha512sum \u003cfile>.tgz`). The\n`aws-msk-iam-auth` jar installs automatically from its AWS-owned source (no parameter).\n\nIf the customer asks you to pick the URL or run the install for them, don't flatly\nrefuse — briefly explain you can't vouch for a specific link on their behalf, then\nhelp: confirm the version, name Apache as the publisher, point to an official channel\nby description, and show them how to get the checksum.\n\n**Pre-flight before invoking:** confirm the customer-provided `kafkaUrl` starts with\n`https://` (reject and ask for an HTTPS URL otherwise) — catch this before the\n`send-command`, not as a failed run. The install document re-checks the scheme as a\nbackstop (e.g. for the self-guided path where the customer runs the command directly).\n\n```bash\n# Single-quote --parameters: a URL with a ?/& query string would otherwise be\n# glob-expanded by the shell (e.g. zsh: \"no matches found\").\naws ssm send-command --document-name msk-express-simulation-InstallKafkaClient \\\n --targets Key=tag:simulation:role,Values=producer,consumer \\\n --parameters 'kafkaUrl=\u003ckafka-client-tarball-url>,kafkaSha512=\u003csha512>' \\\n --region \u003cregion>\n```\n\nWait for it to succeed, then validate.\n\n### Step 2 — validate\n\nAfter install, run `ValidateClient` and loop until it passes:\n\n```bash\naws ssm send-command --document-name msk-express-simulation-ValidateClient \\\n --targets Key=tag:simulation:role,Values=producer,consumer --region \u003cregion>\n```\n\n- **PASS** → proceed to topic creation + tests.\n- **FAIL** → show the error (version too low, or cannot authenticate/connect). Help\n the customer fix or re-install the Kafka client by re-running the install command\n with a different `kafkaUrl` + `kafkaSha512` they provide. Re-run\n `ValidateClient` until it passes.\n\nThe test documents also carry a lightweight client check, so a test never runs\nagainst a fleet with no client.\n\n### If a test hits a client-related error\n\nIf a test fails with a client-related error, the cause may be the Kafka client that\nwas installed. Have the customer re-install with a different client (a `kafkaUrl` +\n`kafkaSha512` they provide — you don't source the URL) and re-run the test.\n\n## Env computation\n\n```bash\nuv run scripts/simulation_load_test_config.py compute --instance-type \u003ctype> --broker-count \u003cN> --kafka-version \u003cver>\n```\n\nAlways pass `--kafka-version` with the CloudFormation `KafkaVersion` string you intend to\ndeploy. The question UI always appends an \"Other\" free-form option to every\nchoice prompt, which a customer can use to enter an unsupported `instance_type`,\n`broker_count`, or version. `compute` rejects all three locally (non-zero exit\nwith an `error` message) so a free-form value fails here in seconds rather than\n~40 min into `create-stack`. On rejection, re-present the supported choices\n(`supported_kafka_versions` is in the output) and do not deploy.\n\nReturns cluster capacity, the **auto-sized** client fleet, and each test's\ndefaults/bounds. Map the output to CloudFormation parameters:\n\n| CloudFormation parameter | From compute output |\n|---|---|\n| InstanceType / BrokerCount / KafkaVersion | sizing inputs |\n| ClientInstanceType | `fleet_instance_type` |\n| ProducerCount | `fleet_producer_count` |\n| ConsumerCount | `fleet_consumer_count` |\n\nThe fleet is sized to ~1.5× the cluster's **maximum** ingress quota (its throttle\nceiling), so it can drive the cluster all the way to its limit and is never the\nbottleneck. The customer cannot change fleet size; to change cluster sizing, redeploy.\n\n## Render the local template\n\n**Self-guided path only.** Render exclusively when the customer **declines** to let\nyou deploy (flow step 6, \"No\"). If they consent, skip render entirely and deploy in\none `create-stack` with `--parameters` (see [Deploy](#deploy)).\n\nOn the self-guided path, render a local, ready-to-deploy copy of the template with the\nsizing and computed fleet baked in as parameter `Default`s. This is what the\ncustomer downloads and deploys from — never the skill's source template.\n\nPass `--output` as an **absolute path under the customer's working directory** (not\nthe skill directory). `render` resolves whatever you pass to an absolute path —\na relative path resolves against the current working directory, which is ambiguous\nand could land the file inside the skill package — so anchor it explicitly with\n`$(pwd)`:\n\n```bash\nuv run scripts/simulation_load_test_config.py render \\\n --instance-type \u003ctype> --broker-count \u003cN> --kafka-version \u003cver> \\\n --output \"$(pwd)/migrate-to-msk-skill-artifacts/simulation/simulation-stack.yaml\"\n```\n\nAfter the customer deploys the rendered template, they install + validate the Kafka\nclient exactly like the consented path (via the `InstallKafkaClient` /\n`ValidateClient` SSM documents — see [Install the Kafka client](#install-the-kafka-client)).\nNothing client-related is baked into the rendered template.\n\n`render` re-runs the same sizing math as `compute` (so it rejects an unsupported\n`instance_type`, `broker_count`, or `kafka_version` the same way, before writing\nanything), reads the static source template internally, and bakes these parameters\nas `Default`s — so a plain `create-stack` with no `--parameters` deploys the\ncustomer's exact sizing:\n\n| Baked parameter | Value |\n|---|---|\n| InstanceType / BrokerCount / KafkaVersion | sizing inputs |\n| ClientInstanceType | `fleet_instance_type` |\n| ProducerCount | `fleet_producer_count` |\n| ConsumerCount | `fleet_consumer_count` |\n\n`render` returns the **resolved absolute** `output_path` (and a ready-to-run\n`deploy_command` that points at it). Use that absolute path verbatim — it is the\n**only** template path you ever show the customer. Tell them where it was saved so\nthey have the exact artifact being deployed. `render` refuses to write inside the\nskill package, so if you accidentally point `--output` there it errors instead of\nleaking the artifact into the install directory.\n\n## Throughput & fleet reference\n\nThese tables are the source of the numbers `simulation_load_test_config.py` computes from. They\nare reference only — the script is authoritative; do not hand-compute from them. The figures\nreflect published Express performance characteristics. If you can access AWS documentation, treat\nthe official AWS MSK Express documentation as the source of truth and validate against it; if that\ndocumentation is not accessible, these values are a trustworthy source to use.\n\n### Express per-broker throughput throttle limits (MB/s)\n\n`sustained` = recommended threshold (no degradation up to here). `max` = hard quota\n(MSK throttles read/write traffic beyond it). Cluster totals = per-broker × broker\ncount. The sizing math uses **ingress**; egress is listed for reference (it can bind\nfirst when many consumer groups read the stream).\n\n| Instance | Ingress sustained | Ingress max | Egress sustained | Egress max |\n|---|---|---|---|---|\n| express.m7g.large | 15.6 | 23.4 | 31.2 | 58.5 |\n| express.m7g.xlarge | 31.2 | 46.8 | 62.5 | 117 |\n| express.m7g.2xlarge | 62.5 | 93.7 | 125 | 234.2 |\n| express.m7g.4xlarge | 124.9 | 187.5 | 249.8 | 468.7 |\n| express.m7g.8xlarge | 250 | 375 | 500 | 937.5 |\n| express.m7g.12xlarge | 375 | 562.5 | 750 | 1406.2 |\n| express.m7g.16xlarge | 500 | 750 | 1000 | 1875 |\n\n### Client fleet sizing\n\nThe fleet runs `kafka-producer-perf-test` / `kafka-e2e-latency` — network- and\nCPU-bound. All clusters use a fixed fleet instance type (`c5.2xlarge`); only the\ncount scales: `fleet_producer_count = max(2, ceil(cluster_max_ingress × 1.5 /\n400))`, and the same count of consumers.\n\n- **Instance type:** `c5.2xlarge` (10 Gbps sustained baseline, no burst credits)\n- **Per-instance throughput target:** 400 MB/s (conservative for Kafka with TLS + acks=all)\n- **Why c5.2xlarge:** stable non-burstable network; smaller c5 sizes have burstable\n baselines unsuitable for sustained load generation.\n\n## Deploy\n\nThere are two deploy commands depending on which path the customer chose at flow\nstep 6. Both create the same stack; they differ only in where the template comes\nfrom and whether sizing is passed as parameters or baked in.\n\n### Consented path — you deploy (no render)\n\nDeploy in **one** `create-stack` straight from the static source template, passing\nthe compute-derived values as `--parameters`. This is the single deploy step; do\nnot render on this path.\n\n```bash\naws cloudformation create-stack --stack-name msk-express-simulation \\\n --template-body file://\u003cskill>/assets/simulation-stack.yaml \\\n --parameters ParameterKey=InstanceType,ParameterValue=\u003ctype> \\\n ParameterKey=BrokerCount,ParameterValue=\u003cN> \\\n ParameterKey=KafkaVersion,ParameterValue=\u003cver> \\\n ParameterKey=ClientInstanceType,ParameterValue=\u003cfleet_instance_type> \\\n ParameterKey=ProducerCount,ParameterValue=\u003cfleet_producer_count> \\\n ParameterKey=ConsumerCount,ParameterValue=\u003cfleet_consumer_count> \\\n --capabilities CAPABILITY_IAM --region \u003cregion>\n\naws cloudformation wait stack-create-complete --stack-name msk-express-simulation --region \u003cregion>\n```\n\nThe stack downloads no Kafka client — install it after `CREATE_COMPLETE` (see\n[Install the Kafka client](#install-the-kafka-client)).\n\n`\u003cskill>/assets/simulation-stack.yaml` is the install path — you (the agent) use it\nhere, but never surface it to the customer. The customer consented to you deploying,\nso they never need the template path on this path; you run the command on their\nbehalf.\n\n### Self-guided path — customer deploys (render first)\n\nOnly on the self-guided path. After [rendering](#render-the-local-template) the\nfilled local template, hand them this command using the absolute `output_path`\n`render` reported (its `deploy_command` field is exactly this, minus the region).\nSizing is baked into the parameter `Default`s, so no `--parameters` are needed:\n\n```bash\naws cloudformation create-stack --stack-name msk-express-simulation \\\n --template-body file://\u003cabsolute-output_path-from-render> \\\n --capabilities CAPABILITY_IAM --region \u003cregion>\n\naws cloudformation wait stack-create-complete --stack-name msk-express-simulation --region \u003cregion>\n```\n\nOn this path the customer only ever sees the local rendered artifact path, never the\nskill's source template path.\n\nThe fleet ASGs gate `CREATE_COMPLETE` on a `cfn-signal` from each instance's\nUserData, so reaching `CREATE_COMPLETE` means the fleet finished bootstrapping\n(Java + dirs + `client.properties`). The **Kafka client is not installed yet** —\nthat's the post-deploy step ([Install the Kafka client](#install-the-kafka-client)).\nAs a belt-and-suspenders check before the first SSM command, confirm the producer\ninstances pass their EC2 status checks — the SSM agent reports `Online` before an\ninstance is through boot, and a command issued too early is silently `Terminated`\nwith no output.\n\n> **Shell-safe instance-ID handling (do not skip).** `--output text` returns\n> instance IDs **tab-separated on one line**. Do **not** capture them in a plain\n> variable and pass it unquoted to `--instance-ids` (`... --instance-ids\n> $PRODUCERS`): the session shell is often **zsh**, which — unlike bash — does\n> **not** word-split unquoted parameter expansions, so every ID is passed as a\n> single malformed argument and the call fails with\n> `InvalidInstanceID.Malformed`. Pipe the IDs through `xargs` instead, which\n> splits on whitespace in **any** shell. This same rule applies anywhere you feed\n> a list of IDs to a CLI flag (e.g. the `send-command --instance-ids` calls when\n> [triggering a test](#triggering-a-test)) — never rely on the shell to split a\n> captured variable.\n\n```bash\naws ec2 describe-instances --region \u003cregion> \\\n --filters Name=tag:simulation:role,Values=producer Name=instance-state-name,Values=running \\\n --query 'Reservations[].Instances[].InstanceId' --output text \\\n | tr '\\t' '\\n' \\\n | xargs aws ec2 wait instance-status-ok --region \u003cregion> --instance-ids\n```\n\n**Before creating topics, install and validate the Kafka client** (see\n[Install the Kafka client](#install-the-kafka-client)): install the Kafka client on\nthe fleet (the `kafkaUrl` comes from the customer — you don't source the URL), then\nrun `ValidateClient` and loop until it\npasses. The test documents fail fast without a working client, so install +\nvalidation must succeed first.\n\nThen create the test topics (cluster must be ACTIVE first) using the `CreateTopicsCommand`\nstack output. It creates one topic per test (`\u003cbase>-e2e`, `\u003cbase>-restart`) plus the\ne2e probe topic, so each test's traffic stays isolated.\n\n## Test templates\n\nPresent exactly these two. Show the **Description** and **User Configuration**\nsections only. Fill defaults/bounds from the `compute` output for the deployed\ncluster (the values below are for the 12× express.m7g.4xlarge default).\n\nLead in with the deployed cluster's **ingress capacity** so the customer can\nchoose a throughput with the right context — its **sustained** ingress\n(`cluster_sustained_ingress_mbps`, the recommended ceiling with no degradation)\nand its **maximum** ingress (`cluster_max_ingress_mbps`, the hard quota beyond\nwhich MSK throttles). Phrase it plainly, e.g. *\"This cluster sustains up to\n**X MB/s** ingress and is throttled at a hard maximum of **Y MB/s** — pick a test\nthroughput with those in mind.\"* Take both numbers straight from the `compute`\noutput; do not hand-compute them.\n\n### 1. End-to-End Latency\n\n**Description:** Measures produce-to-consume round-trip latency under a steady\nload. Producers drive the target throughput while one producer samples end-to-end\nlatency; the value is emitted as the `MSKSimulation/E2ELatencyMs` custom metric.\nConsumers drain the topic (consumer group `simulation-e2e`) so the consume path is\nexercised too — watch BytesOut and consumer lag on the dashboard alongside latency.\n\n**User Configuration:**\n\n| Parameter | Default | Range |\n|---|---|---|\n| throughput_mbps | 10 | 1 – fleet_max_mbps |\n| duration_minutes | 10 | 5 – 120 |\n| num_producers | 1 | 1 – fleet_producer_count |\n| num_consumers | 1 | 1 – fleet_consumer_count |\n\n### 2. Broker Restart Under Load\n\n**Description:** Sustains target throughput, then reboots one broker mid-test to\nobserve failover behavior and recovery. Producers emit `MSKSimulation/ProduceLatencyMs`\nover the window, and consumers read the topic (consumer group `simulation-restart`) so\nthe consume path is exercised too. The dashboard's ActiveControllerCount, per-cluster\nBytesIn/BytesOut, and consumer lag show the dip and recovery on both sides.\n\n**User Configuration:**\n\n| Parameter | Default | Range |\n|---|---|---|\n| target_throughput_mbps | default_target_throughput | 1 – fleet_max_mbps |\n| duration_minutes | 15 | 10 – 120 |\n| reboot_at_minute | 5 | 2 – (duration_minutes − 2) |\n| num_producers | fleet_producer_count | 1 – fleet_producer_count |\n\n## Guardrails\n\nAlways validate before triggering:\n\n```bash\nuv run scripts/simulation_load_test_config.py validate --test \u003ce2e_latency|broker_restart> \\\n --config '\u003cjson>' --instance-type \u003ctype> --broker-count \u003cN>\n```\n\nTwo severity levels:\n\n- **REJECT** (`decision: REJECT`, exit 1) — value exceeds a physical fleet limit\n or is an invalid timing (e.g. `reboot_at_minute` not inside the run). Show the\n error and loop back to let the customer change the value.\n- **WARN** (`warnings: [...]`, still `PASS`) — value exceeds a cluster best\n practice (e.g. above cluster max ingress, or too few producers to sustain the\n rate). Surface as **one sentence** at trigger time and continue.\n\n## Triggering a test\n\nSelect which fleet instances participate by passing their instance IDs to\n`send-command`. Per-instance throughput = total ÷ number of producers targeted.\nSet `leadInstanceId` to the **first** targeted producer (it samples latency /\nreboots the broker). For **E2E Latency**, `num_producers` / `num_consumers` set how\nmany of each you target. For **Broker Restart**, target the chosen `num_producers`\nproducers and all\nconsumers — consumers drain the topic for read-health monitoring. Each test uses\nits **own** load topic (E2E -> `\u003cbase>-e2e`, Broker Restart -> `\u003cbase>-restart`),\nso the two tests are fully isolated -- an idle test's consumer group never\naccumulates phantom lag from the other test writing to a shared topic. On top of\nthat, each run **resets its consumer group's offset to its topic's latest** at\nstart, so every run begins at ~0 lag (ignoring any leftover from a prior run of the\nsame test) and its consumer-lag / BytesOut reflect only its own traffic. After the\nproducers stop, the consumer keeps reading and **adaptively drains**: it polls the\ngroup's lag and runs until the lag reaches ~0 (caught up) or `maxDrainMinutes`\n(default 20) elapses, whichever first — then exits. Without this tail the consumer\nwould stop with the producers and `MaxOffsetLag` would freeze at its peak (a stale\nflat line that wrongly looks like the cluster can't keep up). The adaptive drain\nself-tunes to any cluster size / rate and exits early once caught up (no idle\nwaste); raise `maxDrainMinutes` only if a very large backlog needs more than 20 min.\n\n**Tell the customer at trigger time:** after the producers stop, the consumers keep\nreading until the dashboard's consumer-lag drains back to ~0 — so the run takes a\nlittle longer than `duration_minutes` to fully settle, and that trailing drain is\nexpected, not a problem.\n\n```bash\n# E2E Latency (targets both roles; the doc branches on the simulation:role IMDS tag)\naws ssm send-command --document-name msk-express-simulation-RunE2ELatency \\\n --instance-ids \u003cproducer-ids...> \u003cconsumer-ids...> \\\n --parameters perInstanceThroughputMbps=\u003ctotal/num_producers>,durationMinutes=\u003cd>,leadInstanceId=\u003cfirst-producer-id> \\\n --region \u003cregion>\n\n# Broker Restart (targets both roles; producers load + reboot, consumers read + drain)\n# After producers stop, consumers adaptively drain until lag ~0 (capped at maxDrainMinutes,\n# default 20). Only pass maxDrainMinutes to override the cap; the default is usually fine.\naws ssm send-command --document-name msk-express-simulation-RunBrokerRestartTest \\\n --instance-ids \u003cproducer-ids...> \u003cconsumer-ids...> \\\n --parameters perInstanceThroughputMbps=\u003ctotal/num_producers>,durationMinutes=\u003cd>,rebootAtMinute=\u003cm>,leadInstanceId=\u003cfirst-producer-id> \\\n --region \u003cregion>\n```\n\nThe dashboard URL + ARN were already shared once at `CREATE_COMPLETE`; point the\ncustomer back to that dashboard for results rather than re-posting it each run.\n\n## Teardown\n\n```bash\naws cloudformation delete-stack --stack-name msk-express-simulation --region \u003cregion>\n```\n\n## Off-trail handling\n\n- **Asks to change cluster sizing after deploy** — not supported in place. Offer\n to tear down and redeploy with new sizing.\n- **Asks for a third/custom test type** — only the two vended tests are available;\n offer those instead.\n- **Asks to edit the template / infra** — the source template is static and the\n rendered local copy only carries baked sizing; decline hand-edits and offer to\n re-render + redeploy with different sizing instead.\n- **Asks what the results mean** — present the dashboard and explain what the\n metrics represent, but do not assert a pass/fail verdict; the customer evaluates\n the numbers against their own targets.\n- **Asks to deploy a second/another simulation while one exists** — only one simulation is\n allowed per account at a time; offer to reuse the existing one or delete it and\n redeploy (see [Single simulation per account](#single-simulation-per-account)).\n- **Asks where the Kafka client comes from** — the Kafka client is installed\n post-deploy via `InstallKafkaClient` from a `kafkaUrl` the customer provides (you\n may point them to Apache in words, but don't fabricate a URL). The `aws-msk-iam-auth`\n jar is installed automatically from its AWS-owned source. See [Install the Kafka client](#install-the-kafka-client).\n- **Reuses an existing simulation** — a `CREATE_COMPLETE` stack does not guarantee a\n working Kafka client (install is a post-deploy step that may never have run). After\n presenting the dashboard, run `ValidateClient` before offering tests; if it fails,\n run the `InstallKafkaClient` + `ValidateClient` loop until it passes. See\n [Install the Kafka client](#install-the-kafka-client).\n- **A test fails with a \"Kafka client not installed/working\" error** — install the\n client via `InstallKafkaClient`, run `ValidateClient` until it passes, then retry\n the test.\n- **A test hits a client-related error** — it may be the installed Kafka client;\n have the customer re-install with a different client (a `kafkaUrl` + `kafkaSha512`\n they provide — you don't source the URL) and re-run the test. See [Install the Kafka client](#install-the-kafka-client).\n- **Wants to skip the simulation** — the simulation is the last phase, so end the skill.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":39522,"content_sha256":"69d69dae01413a26fd8d4a687647c07650619019233c60177108bf681e69fdb5"},{"filename":"scripts/compatibility.py","content":"#!/usr/bin/env python3\n\"\"\"\nPure file processor: reads a discovery `cluster-config.json` and emits\n`compatibility.\u003ccluster_name>.json` with the five-pillar verdict.\n\nFive pillars: topology, kafka_version, configs, auth, quotas. No live\ncluster or AWS API calls — deterministic and replayable from fixtures.\n\nThe discovery contract carries FULL Kafka config dumps (every broker- and\ntopic-level config the source exposed), not deltas. compatibility.py filters\nagainst per-Kafka-version Apache defaults so values that match the default\ndo not produce evidence — only divergences from default are flagged.\n\nSource of truth for every threshold and rule below is one of these AWS\npublic documentation pages (cited inline at the relevant constant):\n- Express broker overview:\n https://docs.aws.amazon.com/msk/latest/developerguide/msk-broker-types-express.html\n- Express read/write broker and topic configurations:\n https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-express-read-write.html\n- Express read-only broker configurations:\n https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-express-read-only.html\n- Express broker quotas:\n https://docs.aws.amazon.com/msk/latest/developerguide/limits.html#msk-express-quota\n- Express broker best practices:\n https://docs.aws.amazon.com/msk/latest/developerguide/bestpractices-express.html\n\nApache Kafka defaults sourced from:\n- https://kafka.apache.org/documentation/#brokerconfigs\n- https://kafka.apache.org/documentation/#topicconfigs\n\nUsage:\n compatibility.py \u003ccluster-config.json> [--out-dir \u003cdir>]\n\nOutput filename uses the source's `cluster_name` field:\n compatibility.\u003ccluster_name>.json\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport datetime as _dt\nimport json\nimport re\nimport sys\nfrom pathlib import Path\nfrom typing import Any, Iterable\n\n# ---------------------------------------------------------------------------\n# Verdict vocabulary and ordering\n# ---------------------------------------------------------------------------\n\nINFO = \"INFO\"\nADVISORY = \"ADVISORY\"\nACTION_REQUIRED = \"ACTION_REQUIRED\"\n\n_RANK = {INFO: 0, ADVISORY: 1, ACTION_REQUIRED: 2}\n\n\ndef worst(a: str, b: str) -> str:\n return a if _RANK[a] >= _RANK[b] else b\n\n\ndef roll_up(verdicts: Iterable[str]) -> str:\n out = INFO\n for v in verdicts:\n out = worst(out, v)\n return out\n\n\n# ---------------------------------------------------------------------------\n# Pillar 1 — Topology\n# Source: Express broker overview (3-AZ requirement; minimum 3 brokers;\n# KRaft from 3.9). The target broker count is determined by the sizing\n# workbook, not carried over from the source, so no per-cluster broker\n# ceiling is evaluated here.\n# ---------------------------------------------------------------------------\n\nEXPRESS_AZ_COUNT = 3 # Express broker overview (Express is 3-AZ only)\nEXPRESS_TARGET_MIN_BROKERS = 3 # Express broker overview (RF=3 across 3 AZs ⇒ ≥3)\n\n\ndef assess_topology(cfg: dict) -> tuple[str, list[dict]]:\n evidence: list[dict] = []\n verdict = INFO\n topology = cfg[\"topology\"]\n bc = int(topology[\"num_brokers\"])\n az = topology.get(\"num_azs\")\n coordination = cfg[\"kafka\"].get(\"coordination_mechanism\", \"Unknown\")\n ver = parse_version(cfg[\"kafka\"][\"version\"])\n if ver is None:\n evidence.append(\n _ev(\n \"VERSION_UNKNOWN\",\n ADVISORY,\n \"We couldn't determine your cluster's Apache Kafka version, so \"\n \"we can't check the KRaft-on-3.9 transition or the supported \"\n \"version set. Provide `kafka.version` for a complete \"\n \"assessment.\",\n )\n )\n verdict = worst(verdict, ADVISORY)\n ver = (0, 0) # neutral placeholder; skips the version-specific check below\n\n # AZ count\n if az is None:\n evidence.append(\n _ev(\n \"AZ_COUNT_UNKNOWN\",\n ADVISORY,\n \"We couldn't determine how many Availability Zones your cluster \"\n \"uses. MSK Express always deploys across 3 Availability Zones, \"\n \"so your MSK Express cluster will use 3 regardless.\",\n )\n )\n verdict = worst(verdict, ADVISORY)\n elif int(az) != EXPRESS_AZ_COUNT:\n evidence.append(\n _ev(\n \"AZ_COUNT_NOT_3\",\n ADVISORY,\n f\"Your cluster spans {az} Availability Zone(s). MSK Express \"\n f\"always deploys across {EXPRESS_AZ_COUNT} Availability Zones, \"\n f\"so your MSK Express cluster will use {EXPRESS_AZ_COUNT} \"\n \"regardless.\",\n observed=int(az),\n required=EXPRESS_AZ_COUNT,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n # Broker-count floor. Express requires at least 3 brokers (RF=3 across 3\n # AZs). The target broker count is sized by the workbook, not taken from\n # the source, so this is an informational note, not a pass/fail check.\n if bc \u003c EXPRESS_TARGET_MIN_BROKERS:\n evidence.append(\n _ev(\n \"BROKER_COUNT_LT_3\",\n ADVISORY,\n f\"Your cluster has {bc} broker(s). MSK Express uses a minimum \"\n f\"of {EXPRESS_TARGET_MIN_BROKERS} brokers across 3 Availability \"\n \"Zones. The sizing workbook helps you choose your MSK Express \"\n \"broker count based on your throughput, so it isn't carried \"\n \"over from your current cluster.\",\n observed=bc,\n required=EXPRESS_TARGET_MIN_BROKERS,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n # KRaft transition on 3.9 — flagged informationally.\n if ver[:2] == (3, 9) and coordination == \"ZooKeeper\":\n evidence.append(\n _ev(\n \"KRAFT_REQUIRED_FOR_VERSION\",\n ADVISORY,\n \"Your cluster runs Kafka 3.9 with ZooKeeper. On Kafka 3.9, MSK \"\n \"Express runs in KRaft mode and is provisioned that way \"\n \"automatically; no action needed. If any of your tooling talks \"\n \"to ZooKeeper directly, see the Apache Kafka KRaft migration \"\n \"documentation: \"\n \"https://kafka.apache.org/37/operations/kraft/#zookeeper-to-kraft-migration\",\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n return verdict, evidence\n\n\n# ---------------------------------------------------------------------------\n# Pillar 2 — Kafka version\n# Source: Express broker overview — \"Express brokers are supported on the\n# following Apache Kafka versions: 3.6, 3.8, and 3.9.\"\n# ---------------------------------------------------------------------------\n\nEXPRESS_SUPPORTED_VERSIONS: set[tuple[int, int]] = {(3, 6), (3, 8), (3, 9), (4, 2)}\n\n# Minimum source Apache Kafka version that MSK Replicator can replicate from\n# when migrating a self-managed source to Express. Sources older than this can\n# still re-create the cluster architecture, but data migration must use a\n# MirrorMaker 2 based solution instead.\n# Source: \"Migrate third-party and self-managed Apache Kafka clusters to Amazon\n# MSK Express brokers with Amazon MSK Replicator\" (version 2.8.1+).\nMSK_REPLICATOR_MIN_SOURCE_VERSION: tuple[int, int, int] = (2, 8, 1)\n\n\ndef assess_kafka_version(cfg: dict) -> tuple[str, list[dict]]:\n evidence: list[dict] = []\n verdict = INFO\n raw = cfg[\"kafka\"][\"version\"]\n supported_sorted = sorted(EXPRESS_SUPPORTED_VERSIONS)\n supported_str = [f\"{m}.{n}\" for m, n in supported_sorted]\n supported_version_str = \", \".join(supported_str)\n\n parsed = parse_version(raw)\n if parsed is None:\n evidence.append(\n _ev(\n \"VERSION_UNKNOWN\",\n ADVISORY,\n \"We couldn't determine your cluster's Apache Kafka version, so \"\n f\"we can't confirm it's in the set MSK Express supports ({supported_version_str}). \"\n \"Provide `kafka.version` for a complete assessment.\",\n supported=supported_str,\n )\n )\n return ADVISORY, evidence\n\n mm = parsed[:2]\n if mm in EXPRESS_SUPPORTED_VERSIONS:\n evidence.append(\n _ev(\n \"VERSION_SUPPORTED\",\n INFO,\n f\"Your cluster runs Apache Kafka {raw}, which MSK Express \"\n f\"supports ({supported_version_str}). Your workload will run on the \"\n \"same Kafka version after migrating.\",\n )\n )\n else:\n msg = (\n f\"MSK Express supports Apache Kafka {supported_version_str}. Your cluster \"\n f\"runs {raw}, so after migrating your workload will run on a new \"\n \"Kafka version. Confirm your client libraries and applications are \"\n \"compatible with the version you choose for Express — Kafka clients \"\n \"are generally compatible across minor versions, but we recommend \"\n \"validating in a test environment before migrating. See the Apache \"\n \"Kafka upgrade notes at https://kafka.apache.org/documentation/#upgrade \"\n \"for details.\"\n )\n if parsed \u003c MSK_REPLICATOR_MIN_SOURCE_VERSION:\n msg += (\n \" If intending to migrate your data along with your cluster, \"\n \"please note that MSK Replicator can only copy data from \"\n \"clusters running Apache Kafka 2.8.1 or later. Since your \"\n \"cluster is older, you would need to set up a MirrorMaker 2 \"\n \"based solution instead.\"\n )\n evidence.append(\n _ev(\n \"VERSION_NOT_IN_EXPRESS_SET\",\n ADVISORY,\n msg,\n observed=raw,\n supported=supported_str,\n )\n )\n verdict = ADVISORY\n\n return verdict, evidence\n\n\n# ---------------------------------------------------------------------------\n# Pillar 3 — Configs (broker- and topic-level)\n# Sources: Express read/write configurations page (R/W broker + topic\n# configs, bounded ranges), Express read-only configurations page (forced\n# read-only values).\n#\n# Discovery passes FULL config dumps. We compare each value to the Apache\n# Kafka default for the source's version; only divergences from default are\n# evaluated against Express's R/W, RO, range, and forced-value sets.\n# ---------------------------------------------------------------------------\n\n# Express R/W broker configs (source: Express read/write configurations page).\nEXPRESS_BROKER_RW: frozenset[str] = frozenset(\n {\n \"advertised.listeners\",\n \"allow.everyone.if.no.acl.found\",\n \"auto.create.topics.enable\",\n \"compression.type\",\n \"connections.max.idle.ms\",\n \"delete.topic.enable\",\n \"group.initial.rebalance.delay.ms\",\n \"group.max.session.timeout.ms\",\n \"leader.imbalance.per.broker.percentage\",\n \"log.cleaner.delete.retention.ms\",\n \"log.cleaner.max.compaction.lag.ms\",\n \"log.cleaner.min.compaction.lag.ms\",\n \"log.cleanup.policy\",\n \"log.message.timestamp.after.max.ms\",\n \"log.message.timestamp.before.max.ms\",\n \"log.message.timestamp.type\",\n \"log.retention.bytes\",\n \"log.retention.ms\",\n \"max.connection.creation.rate\",\n \"max.connections\",\n \"max.connections.per.ip\",\n \"max.connections.per.ip.overrides\",\n \"max.incremental.fetch.session.cache.slots\",\n \"message.max.bytes\",\n \"num.partitions\",\n \"offsets.retention.minutes\",\n \"producer.id.expiration.ms\",\n \"replica.fetch.max.bytes\",\n \"replica.selector.class\",\n \"socket.receive.buffer.bytes\",\n \"socket.request.max.bytes\",\n \"socket.send.buffer.bytes\",\n \"transaction.max.timeout.ms\",\n \"transactional.id.expiration.ms\",\n }\n)\n\n# Express read-only broker configs (source: Express read-only configurations page).\nEXPRESS_BROKER_RO: frozenset[str] = frozenset(\n {\n \"broker.id\",\n \"broker.rack\",\n \"default.replication.factor\",\n \"fetch.max.bytes\",\n \"group.max.size\",\n \"inter.broker.listener.name\",\n \"inter.broker.protocol.version\",\n \"listeners\",\n \"log.message.format.version\",\n \"min.insync.replicas\",\n \"num.io.threads\",\n \"num.network.threads\",\n \"replica.fetch.response.max.bytes\",\n \"request.timeout.ms\",\n \"transaction.state.log.min.isr\",\n \"transaction.state.log.replication.factor\",\n \"unclean.leader.election.enable\",\n }\n)\n\nONE_DAY_MS = 24 * 60 * 60 * 1000 # 86_400_000\n\n# Bounded ranges enforced by Express (source: Express read/write configurations page).\nEXPRESS_BROKER_RANGES: dict[str, tuple[int | None, int | None]] = {\n \"log.cleaner.max.compaction.lag.ms\": (ONE_DAY_MS, None),\n}\n\n# Forced values (source: Express read-only configurations page).\nEXPRESS_BROKER_FORCED: dict[str, Any] = {\n \"default.replication.factor\": 3,\n \"min.insync.replicas\": 2,\n \"transaction.state.log.min.isr\": 2,\n \"unclean.leader.election.enable\": \"false\",\n}\n\n# Express R/W topic configs (source: Express read/write configurations page,\n# topic-level table).\nEXPRESS_TOPIC_RW: frozenset[str] = frozenset(\n {\n \"cleanup.policy\",\n \"compression.type\",\n \"delete.retention.ms\",\n \"max.compaction.lag.ms\",\n \"max.message.bytes\",\n \"message.timestamp.after.max.ms\",\n \"message.timestamp.before.max.ms\",\n \"message.timestamp.type\",\n \"min.compaction.lag.ms\",\n \"retention.bytes\",\n \"retention.ms\",\n }\n)\n\nEXPRESS_TOPIC_RANGES: dict[str, tuple[int | None, int | None]] = {\n \"max.compaction.lag.ms\": (ONE_DAY_MS, None),\n}\n\n# Topic-level forced values (Express read/write configurations page,\n# topic-level intro paragraph).\nTOPIC_FORCED: dict[str, Any] = {\n \"min.insync.replicas\": 2,\n \"unclean.leader.election.enable\": \"false\",\n}\n\nEXPRESS_TOPIC_FORCED_RF = 3\n\n# ---------------------------------------------------------------------------\n# Apache Kafka defaults per supported version.\n#\n# Source: https://kafka.apache.org/documentation/#brokerconfigs and\n# https://kafka.apache.org/documentation/#topicconfigs for each version.\n#\n# Only configs we actually check (above sets) need defaults here. If the\n# source's value matches the default for its version, we emit no evidence.\n# ---------------------------------------------------------------------------\n\n# Defaults shared across 3.6/3.8/3.9 for the configs we evaluate.\n_BROKER_DEFAULTS_COMMON: dict[str, Any] = {\n # R/W broker configs we check ranges/forced values for, plus high-traffic\n # configs that show up in dumps. Strings are normalized in _is_default.\n \"auto.create.topics.enable\": \"true\",\n \"compression.type\": \"producer\",\n \"connections.max.idle.ms\": 600000,\n \"delete.topic.enable\": \"true\",\n \"group.initial.rebalance.delay.ms\": 3000,\n \"group.max.session.timeout.ms\": 1800000,\n \"leader.imbalance.per.broker.percentage\": 10,\n \"log.cleaner.delete.retention.ms\": 86400000,\n \"log.cleaner.max.compaction.lag.ms\": 9223372036854775807,\n \"log.cleaner.min.compaction.lag.ms\": 0,\n \"log.cleanup.policy\": \"delete\",\n \"log.message.timestamp.after.max.ms\": 9223372036854775807,\n \"log.message.timestamp.before.max.ms\": 9223372036854775807,\n \"log.message.timestamp.type\": \"CreateTime\",\n \"log.retention.bytes\": -1,\n \"log.retention.ms\": -1,\n \"max.connection.creation.rate\": 2147483647,\n \"max.connections\": 2147483647,\n \"max.connections.per.ip\": 2147483647,\n \"max.incremental.fetch.session.cache.slots\": 1000,\n \"message.max.bytes\": 1048588,\n \"num.partitions\": 1,\n \"offsets.retention.minutes\": 10080,\n \"producer.id.expiration.ms\": 86400000,\n \"replica.fetch.max.bytes\": 1048576,\n \"socket.receive.buffer.bytes\": 102400,\n \"socket.request.max.bytes\": 104857600,\n \"socket.send.buffer.bytes\": 102400,\n \"transaction.max.timeout.ms\": 900000,\n \"transactional.id.expiration.ms\": 604800000,\n # Forced-value configs we check\n \"default.replication.factor\": 1,\n \"min.insync.replicas\": 1,\n \"transaction.state.log.min.isr\": 2,\n \"unclean.leader.election.enable\": \"false\",\n \"num.io.threads\": 8,\n \"num.network.threads\": 3,\n \"allow.everyone.if.no.acl.found\": \"false\",\n}\n\nBROKER_DEFAULTS_BY_VERSION: dict[tuple[int, int], dict[str, Any]] = {\n (3, 6): dict(_BROKER_DEFAULTS_COMMON),\n (3, 8): dict(_BROKER_DEFAULTS_COMMON),\n (3, 9): dict(_BROKER_DEFAULTS_COMMON),\n}\n\n_TOPIC_DEFAULTS_COMMON: dict[str, Any] = {\n \"cleanup.policy\": \"delete\",\n \"compression.type\": \"producer\",\n \"delete.retention.ms\": 86400000,\n \"max.compaction.lag.ms\": 9223372036854775807,\n \"max.message.bytes\": 1048588,\n \"message.timestamp.after.max.ms\": 9223372036854775807,\n \"message.timestamp.before.max.ms\": 9223372036854775807,\n \"message.timestamp.type\": \"CreateTime\",\n \"min.compaction.lag.ms\": 0,\n \"retention.bytes\": -1,\n \"retention.ms\": 604800000,\n \"min.insync.replicas\": 1,\n \"unclean.leader.election.enable\": \"false\",\n}\n\nTOPIC_DEFAULTS_BY_VERSION: dict[tuple[int, int], dict[str, Any]] = {\n (3, 6): dict(_TOPIC_DEFAULTS_COMMON),\n (3, 8): dict(_TOPIC_DEFAULTS_COMMON),\n (3, 9): dict(_TOPIC_DEFAULTS_COMMON),\n}\n\n\ndef _coerce_int(value: Any) -> int | None:\n \"\"\"Return int(value) or None if not coercible.\"\"\"\n if isinstance(value, bool):\n return None\n if isinstance(value, int):\n return value\n if isinstance(value, str):\n try:\n return int(value)\n except ValueError:\n return None\n return None\n\n\ndef _normalize_value(value: Any) -> str:\n \"\"\"Normalize a config value to a comparable string.\"\"\"\n if isinstance(value, bool):\n return \"true\" if value else \"false\"\n return str(value).strip().lower()\n\n\ndef _is_default(key: str, value: Any, defaults: dict[str, Any]) -> bool:\n \"\"\"Return True if value matches the Apache Kafka default for key.\"\"\"\n if key not in defaults:\n # No known default — cannot tell, treat as not-default so the rule\n # still fires.\n return False\n return _normalize_value(value) == _normalize_value(defaults[key])\n\n\ndef _check_range(\n key: str,\n value: Any,\n bounds: tuple[int | None, int | None],\n subject: str,\n) -> dict | None:\n \"\"\"Returns evidence-style dict if out-of-range, else None.\n\n `subject` names whose setting it is (\"Your cluster\" for broker configs,\n f\"Topic {name!r}\" for topic configs) so the message reads in the customer's\n voice.\n \"\"\"\n lo, hi = bounds\n lo_s = lo if lo is not None else \"INT_MIN\"\n hi_s = hi if hi is not None else \"INT_MAX\"\n iv = _coerce_int(value)\n if iv is None:\n return {\n \"config_key\": key,\n \"detail\": (\n f\"MSK Express expects {key} to be an integer within \"\n f\"[{lo_s}, {hi_s}]. {subject} sets it to {value!r}, which is \"\n \"not a valid integer, so this configuration can't be migrated \"\n \"as-is.\"\n ),\n \"observed\": value,\n \"limit\": [lo, hi],\n }\n if (lo is not None and iv \u003c lo) or (hi is not None and iv > hi):\n return {\n \"config_key\": key,\n \"detail\": (\n f\"MSK Express accepts {key} only within [{lo_s}, {hi_s}]. \"\n f\"{subject} sets it to {value!r}, which is outside that range, \"\n \"so this configuration can't be migrated as-is.\"\n ),\n \"observed\": iv,\n \"limit\": [lo, hi],\n }\n return None\n\n\ndef _ev(\n code: str,\n severity: str,\n detail: str,\n **extra: Any,\n) -> dict:\n \"\"\"Build an evidence dict tagged with its own severity.\n\n severity is per-finding; pillar verdict is the worst across findings, but\n summary buckets each finding individually.\n \"\"\"\n out: dict[str, Any] = {\"code\": code, \"severity\": severity, \"detail\": detail}\n out.update(extra)\n return out\n\n\ndef assess_configs(cfg: dict) -> tuple[str, list[dict]]:\n \"\"\"Pillar 3: broker- and topic-level config compatibility.\n\n Reads broker_configs (full dump) and topics[].configs (full dump), filters\n against per-version Apache Kafka defaults, then evaluates non-default\n values against Express's R/W, RO, range, and forced-value sets.\n \"\"\"\n evidence: list[dict] = []\n verdict = INFO\n\n parsed = parse_version(cfg[\"kafka\"][\"version\"])\n if parsed is None:\n evidence.append(\n _ev(\n \"VERSION_UNKNOWN\",\n ADVISORY,\n \"We couldn't determine your cluster's Apache Kafka version, so \"\n \"config defaults are compared against a generic baseline \"\n \"instead of your source version's actual defaults. Some \"\n \"config divergences may be under- or over-reported as a \"\n \"result. Provide `kafka.version` for a complete assessment.\",\n )\n )\n verdict = worst(verdict, ADVISORY)\n broker_defaults = _BROKER_DEFAULTS_COMMON\n topic_defaults = _TOPIC_DEFAULTS_COMMON\n else:\n ver: tuple[int, int] = (parsed[0], parsed[1])\n broker_defaults = BROKER_DEFAULTS_BY_VERSION.get(ver, _BROKER_DEFAULTS_COMMON)\n topic_defaults = TOPIC_DEFAULTS_BY_VERSION.get(ver, _TOPIC_DEFAULTS_COMMON)\n\n # 3a. Broker-level configs.\n for key, val in (cfg.get(\"broker_configs\") or {}).items():\n # Skip values that match the Apache default for the source version.\n if _is_default(key, val, broker_defaults):\n continue\n\n # Range checks first — they decide ACTION_REQUIRED.\n if key in EXPRESS_BROKER_RANGES:\n problem = _check_range(key, val, EXPRESS_BROKER_RANGES[key], \"Your cluster\")\n if problem is not None:\n evidence.append(\n _ev(\n \"BROKER_CONFIG_OUT_OF_RANGE\",\n ACTION_REQUIRED,\n problem[\"detail\"],\n config_key=problem[\"config_key\"],\n observed=problem[\"observed\"],\n limit=problem[\"limit\"],\n )\n )\n verdict = worst(verdict, ACTION_REQUIRED)\n continue\n\n if key in EXPRESS_BROKER_FORCED:\n forced = EXPRESS_BROKER_FORCED[key]\n if _normalize_value(val) != _normalize_value(forced):\n evidence.append(\n _ev(\n \"BROKER_CONFIG_FORCED_VALUE\",\n ADVISORY,\n f\"MSK Express manages {key} for you and sets it to \"\n f\"{forced!r} to keep your cluster highly available and \"\n f\"durable; your current value of {val!r} won't apply. We \"\n f\"recommend confirming your workload behaves as expected \"\n f\"with {key}={forced!r} in a test environment to ensure \"\n \"a smooth migration.\",\n config_key=key,\n observed=val,\n enforced=forced,\n )\n )\n verdict = worst(verdict, ADVISORY)\n elif key in EXPRESS_BROKER_RO:\n evidence.append(\n _ev(\n \"BROKER_CONFIG_READ_ONLY\",\n ADVISORY,\n f\"MSK Express manages {key} for you so you don't have to \"\n \"tune it (for thread-related settings it's sized \"\n \"automatically from the broker instance type); your current \"\n \"value won't apply. No action needed.\",\n config_key=key,\n observed=val,\n )\n )\n verdict = worst(verdict, ADVISORY)\n elif key not in EXPRESS_BROKER_RW:\n evidence.append(\n _ev(\n \"BROKER_CONFIG_NOT_EXPOSED\",\n ADVISORY,\n f\"{key} isn't a configurable property on MSK Express — \"\n \"Express manages it internally, and the behavior may differ \"\n f\"from your current value of {val!r}. We recommend \"\n \"validating in a test environment to ensure a smooth \"\n \"migration.\",\n config_key=key,\n observed=val,\n )\n )\n verdict = worst(verdict, ADVISORY)\n # else: in EXPRESS_BROKER_RW and within range — INFO, no evidence.\n\n # 3b. Per-topic configs.\n for topic in cfg.get(\"topics\", []):\n name = topic[\"name\"]\n rf = int(topic[\"replication_factor\"])\n cfgs = topic.get(\"configs\") or {}\n\n if rf != EXPRESS_TOPIC_FORCED_RF:\n evidence.append(\n _ev(\n \"TOPIC_RF_NOT_3\",\n ADVISORY,\n f\"Topic {name!r} uses replication factor {rf}. MSK Express \"\n \"creates every topic with replication factor 3 for \"\n \"durability, so on MSK Express this topic will use 3. No \"\n \"action needed.\",\n topic=name,\n observed=rf,\n enforced=EXPRESS_TOPIC_FORCED_RF,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n for key, val in cfgs.items():\n # Skip default-matching values.\n if _is_default(key, val, topic_defaults):\n continue\n\n if key in EXPRESS_TOPIC_RANGES:\n problem = _check_range(key, val, EXPRESS_TOPIC_RANGES[key], f\"Topic {name!r}\")\n if problem is not None:\n evidence.append(\n _ev(\n \"TOPIC_CONFIG_OUT_OF_RANGE\",\n ACTION_REQUIRED,\n problem[\"detail\"],\n topic=name,\n config_key=problem[\"config_key\"],\n observed=problem[\"observed\"],\n limit=problem[\"limit\"],\n )\n )\n verdict = worst(verdict, ACTION_REQUIRED)\n continue\n\n if key in TOPIC_FORCED:\n forced = TOPIC_FORCED[key]\n if _normalize_value(val) != _normalize_value(forced):\n evidence.append(\n _ev(\n \"TOPIC_CONFIG_FORCED_VALUE\",\n ADVISORY,\n f\"Topic {name!r} sets {key}={val!r}. MSK Express \"\n f\"fixes this at {forced!r} for every topic to keep \"\n \"your data durable, so your current value won't \"\n \"apply. We recommend confirming this topic behaves \"\n f\"as expected with {key}={forced!r} in a test \"\n \"environment to ensure a smooth migration.\",\n topic=name,\n config_key=key,\n observed=val,\n enforced=forced,\n )\n )\n verdict = worst(verdict, ADVISORY)\n elif key not in EXPRESS_TOPIC_RW:\n evidence.append(\n _ev(\n \"TOPIC_CONFIG_NOT_EXPOSED\",\n ADVISORY,\n f\"Topic {name!r} sets {key}={val!r}, which isn't a \"\n \"configurable topic property on MSK Express — Express \"\n \"uses the broker default instead. We recommend \"\n \"validating that the default works for this topic in a \"\n \"test environment to ensure a smooth migration.\",\n topic=name,\n config_key=key,\n observed=val,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n return verdict, evidence\n\n\n# ---------------------------------------------------------------------------\n# Pillar 4 — Auth\n# Sources: Express read-only configurations page (REPLICATION_SECURE listener\n# implies TLS), Express broker quotas page (IAM vs non-IAM connection limits).\n# ---------------------------------------------------------------------------\n\n# Closed enums for the security block. validate_input enforces membership;\n# the auth pillar reads these directly without normalization beyond strict\n# string comparison.\nENCRYPTION_VALUES: frozenset[str] = frozenset({\"TLS\", \"PLAINTEXT\", \"TLS_PLAINTEXT\", \"UNKNOWN\"})\nAUTHENTICATION_VALUES: frozenset[str] = frozenset(\n {\n \"UNAUTHENTICATED\",\n \"TLS\",\n \"SASL_SCRAM\",\n \"SASL_IAM\",\n \"SASL_OAUTHBEARER\",\n # Catch-all for mechanisms that don't fit any of the above (e.g.\n # SASL/GSSAPI/Kerberos, SASL/PLAIN, custom callback handlers).\n # These are not supported on MSK Express; the auth pillar emits\n # ACTION_REQUIRED.\n \"OTHER\",\n # The mechanism could not be determined; the auth pillar emits\n # an ADVISORY asking the customer to verify it.\n \"UNKNOWN\",\n }\n)\n# Authentication mechanisms that resolve to an IAM principal and therefore\n# share the per-broker IAM connection cap.\nIAM_AUTHENTICATION_VALUES: frozenset[str] = frozenset({\"SASL_IAM\", \"SASL_OAUTHBEARER\"})\n# Mechanisms MSK Express accepts that require TLS in transit (IAM, SASL/SCRAM,\n# TLS; OAUTHBEARER is an IAM token transport). Plaintext is only possible with\n# unauthenticated access, so a non-TLS source matters only for these.\nTLS_REQUIRED_AUTH_VALUES: frozenset[str] = frozenset({\"TLS\", \"SASL_SCRAM\", \"SASL_IAM\"})\n\n\ndef assess_auth(cfg: dict) -> tuple[str, list[dict]]:\n evidence: list[dict] = []\n verdict = INFO\n sec = cfg.get(\"security\") or {}\n # Treat a missing or blank field as UNKNOWN rather than a hard failure, so\n # an absent security block yields a calm advisory instead of a false\n # \"not TLS\" finding.\n encryption = sec.get(\"encryption_in_transit\") or \"UNKNOWN\"\n authentication = sec.get(\"authentication\") or \"UNKNOWN\"\n\n # --- Authentication mechanism ---\n # MSK Express supports unauthenticated, TLS, SASL/SCRAM, and IAM. Those\n # carry over as-is (INFO, no evidence). SASL_OAUTHBEARER from a\n # self-managed source is a custom OAuth provider (not the AWS IAM path) —\n # MSK Express does not accept non-AWS OAUTHBEARER tokens. OTHER is also\n # unsupported. UNKNOWN cannot be confirmed (ADVISORY).\n if authentication == \"SASL_OAUTHBEARER\":\n evidence.append(\n _ev(\n \"AUTH_OAUTHBEARER_NOT_SUPPORTED\",\n ACTION_REQUIRED,\n \"Your cluster uses SASL/OAUTHBEARER. MSK Express accepts \"\n \"OAUTHBEARER only as a transport for AWS IAM tokens (using the \"\n \"AWS MSK IAM signer libraries), so a custom OAuth identity \"\n \"provider isn't supported. Move your clients to IAM, \"\n \"SASL/SCRAM, or TLS to ensure a smooth migration.\",\n observed=authentication,\n )\n )\n verdict = worst(verdict, ACTION_REQUIRED)\n elif authentication == \"OTHER\":\n evidence.append(\n _ev(\n \"AUTH_MECHANISM_NOT_SUPPORTED\",\n ACTION_REQUIRED,\n \"Your cluster uses an authentication mechanism MSK Express \"\n \"doesn't support (for example SASL/GSSAPI/Kerberos, SASL/PLAIN, \"\n \"or a custom callback handler). MSK Express accepts \"\n \"unauthenticated, TLS, SASL/SCRAM, and IAM — move your clients \"\n \"to one of these to ensure a smooth migration.\",\n observed=authentication,\n )\n )\n verdict = worst(verdict, ACTION_REQUIRED)\n elif authentication == \"UNKNOWN\":\n evidence.append(\n _ev(\n \"AUTH_UNKNOWN\",\n ADVISORY,\n \"We couldn't determine your cluster's authentication \"\n \"mechanism. Confirm it's one MSK Express supports — one of \"\n \"unauthenticated, TLS, SASL/SCRAM, or IAM — to ensure a smooth \"\n \"migration. MSK Express doesn't support SASL/GSSAPI/Kerberos, \"\n \"SASL/PLAIN, or custom mechanisms.\",\n observed=authentication,\n )\n )\n verdict = worst(verdict, ADVISORY)\n # UNAUTHENTICATED, TLS, SASL_SCRAM, SASL_IAM -> INFO (no evidence).\n\n # --- Encryption in transit (coupled to the auth mechanism) ---\n # Express requires TLS for every authenticated mechanism; plaintext is only\n # possible with unauthenticated access. So a non-TLS source matters only\n # when an authenticated mechanism is in use.\n if encryption == \"UNKNOWN\":\n evidence.append(\n _ev(\n \"ENCRYPTION_UNKNOWN\",\n ADVISORY,\n \"We couldn't determine your cluster's encryption in transit. \"\n \"MSK Express requires TLS for authenticated clients (TLS, \"\n \"SASL/SCRAM, or IAM) and supports PLAINTEXT only for \"\n \"unauthenticated clients. Please confirm your clients can \"\n \"connect over TLS, if they are authenticated, to ensure a \"\n \"smooth migration.\",\n observed=encryption,\n )\n )\n verdict = worst(verdict, ADVISORY)\n elif encryption != \"TLS\" and authentication in TLS_REQUIRED_AUTH_VALUES:\n evidence.append(\n _ev(\n \"ENCRYPTION_NOT_TLS\",\n ACTION_REQUIRED,\n f\"Your cluster uses {encryption!r} for encryption in transit, \"\n \"but MSK Express requires TLS for authenticated clients (TLS, \"\n \"SASL/SCRAM, IAM). Update your clients to connect over TLS to \"\n \"ensure a smooth migration.\",\n observed=encryption,\n )\n )\n verdict = worst(verdict, ACTION_REQUIRED)\n # encryption == \"TLS\" -> INFO. Non-TLS with UNAUTHENTICATED -> INFO\n # (Express permits unauthenticated plaintext). Non-TLS with OTHER is\n # subsumed by the ACTION_REQUIRED above.\n\n return verdict, evidence\n\n\n# ---------------------------------------------------------------------------\n# Pillar 5 — Quotas\n# Source: Express broker quotas page (per-broker max-quota throughput at\n# m7g.16xlarge, partition\n# cap, IAM connection limits, per-partition throughput).\n# ---------------------------------------------------------------------------\n\nEXPRESS_MAX_INGRESS_MAX_QUOTA_PER_BROKER_MBPS = 750 # m7g.16xlarge max\nEXPRESS_MAX_EGRESS_MAX_QUOTA_PER_BROKER_MBPS = 1875 # m7g.16xlarge max\nEXPRESS_MAX_PARTITIONS_PER_BROKER = 32_000\nEXPRESS_MAX_IAM_CONNS_PER_BROKER = 3_000\nEXPRESS_PARTITION_THROUGHPUT_MBPS = 15\n\n\ndef assess_quotas(cfg: dict) -> tuple[str, list[dict]]:\n evidence: list[dict] = []\n verdict = INFO\n metrics = cfg.get(\"metrics\")\n\n if metrics is None:\n evidence.append(\n _ev(\n \"METRICS_MISSING\",\n ADVISORY,\n \"We don't have utilization metrics for your cluster, so we \"\n \"can't check your peak throughput, partitions, and connections \"\n \"against the MSK Express per-broker limits. The sizing estimate \"\n \"falls back to topology only. We recommend editing the sizing \"\n \"sheet manually once these metrics are available and reviewing \"\n \"the Express broker quotas to confirm your workload fits.\",\n )\n )\n return ADVISORY, evidence\n\n pi = metrics.get(\"peak_bytes_in_per_broker_mbps\")\n if pi is not None and pi > EXPRESS_MAX_INGRESS_MAX_QUOTA_PER_BROKER_MBPS:\n evidence.append(\n _ev(\n \"INGRESS_OVER_MAX_BROKER\",\n ADVISORY,\n f\"Your peak ingress of {pi} MBps per broker is more than any \"\n \"single MSK Express broker can absorb (the maximum is \"\n f\"{EXPRESS_MAX_INGRESS_MAX_QUOTA_PER_BROKER_MBPS} MBps, on \"\n \"m7g.16xlarge). Spread the load across more brokers so each \"\n \"carries less ingress. Note that the sizing workbook accounts \"\n \"for this already and can help you select the right MSK Express \"\n \"broker instance type.\",\n observed=pi,\n limit=EXPRESS_MAX_INGRESS_MAX_QUOTA_PER_BROKER_MBPS,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n po = metrics.get(\"peak_bytes_out_per_broker_mbps\")\n if po is not None and po > EXPRESS_MAX_EGRESS_MAX_QUOTA_PER_BROKER_MBPS:\n evidence.append(\n _ev(\n \"EGRESS_OVER_MAX_BROKER\",\n ADVISORY,\n f\"Your peak egress of {po} MBps per broker is more than any \"\n \"single MSK Express broker can absorb (the maximum is \"\n f\"{EXPRESS_MAX_EGRESS_MAX_QUOTA_PER_BROKER_MBPS} MBps, on \"\n \"m7g.16xlarge). Spread the load across more brokers so each \"\n \"carries less egress. Note that the sizing workbook accounts \"\n \"for this already and can help you select the right MSK Express \"\n \"broker instance type.\",\n observed=po,\n limit=EXPRESS_MAX_EGRESS_MAX_QUOTA_PER_BROKER_MBPS,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n pp = metrics.get(\"peak_partitions_per_broker\")\n if pp is not None and pp > EXPRESS_MAX_PARTITIONS_PER_BROKER:\n evidence.append(\n _ev(\n \"PARTITIONS_OVER_MAX_BROKER\",\n ADVISORY,\n f\"Your peak of {pp} partitions per broker is above the MSK \"\n f\"Express maximum ({EXPRESS_MAX_PARTITIONS_PER_BROKER}, on \"\n \"m7g.16xlarge). Adding brokers spreads the partitions so each \"\n \"one carries fewer. Note that the sizing workbook accounts for \"\n \"this already and can help you select the right MSK Express \"\n \"broker instance type.\",\n observed=pp,\n limit=EXPRESS_MAX_PARTITIONS_PER_BROKER,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n pc = metrics.get(\"peak_connections_per_broker\")\n authentication = (cfg.get(\"security\") or {}).get(\"authentication\")\n if (\n pc is not None\n and authentication in IAM_AUTHENTICATION_VALUES\n and pc > EXPRESS_MAX_IAM_CONNS_PER_BROKER\n ):\n evidence.append(\n _ev(\n \"CONNECTIONS_OVER_IAM_LIMIT\",\n ADVISORY,\n f\"Your peak of {pc} connections per broker is above the MSK \"\n f\"Express IAM limit of {EXPRESS_MAX_IAM_CONNS_PER_BROKER} per \"\n \"broker. You can spread the connections across more brokers, or \"\n \"use a non-IAM authentication mechanism, which MSK Express \"\n \"doesn't cap (monitor CPU and memory instead).\",\n observed=pc,\n limit=EXPRESS_MAX_IAM_CONNS_PER_BROKER,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n # Per-partition throughput approximation.\n bc = int(cfg[\"topology\"][\"num_brokers\"])\n total_partitions = sum(int(t[\"num_partitions\"]) for t in cfg.get(\"topics\", []))\n if pi is not None and total_partitions > 0:\n per_part_mbps = (pi * bc) / total_partitions\n if per_part_mbps > EXPRESS_PARTITION_THROUGHPUT_MBPS:\n evidence.append(\n _ev(\n \"PARTITION_THROUGHPUT_OVER_LIMIT\",\n ADVISORY,\n f\"Your average per-partition throughput of \"\n f\"{per_part_mbps:.1f} MB/s is above the MSK Express limit \"\n f\"of {EXPRESS_PARTITION_THROUGHPUT_MBPS} MB/s, where Express \"\n \"begins to throttle. Individual hot partitions can throttle \"\n \"your clients. We recommend spreading the busiest topics \"\n \"across more partitions before migrating.\",\n observed=round(per_part_mbps, 1),\n limit=EXPRESS_PARTITION_THROUGHPUT_MBPS,\n )\n )\n verdict = worst(verdict, ADVISORY)\n\n return verdict, evidence\n\n\n# ---------------------------------------------------------------------------\n# Helpers\n# ---------------------------------------------------------------------------\n\n_VERSION_RE = re.compile(r\"^\\s*(\\d+)\\.(\\d+)(?:\\.(\\d+))?\")\n\n\ndef parse_version(raw: str | None) -> tuple[int, ...] | None:\n \"\"\"Parse '3.6', '3.6.0', '3.6.1' → (3, 6, ...).\n\n Returns None when raw is None — the discovery contract allows\n `kafka.version` to be null when it could not be determined (see\n references/discovery.md). Callers must handle a None return the same\n way they already handle other unknown discovery fields (e.g.\n `topology.num_azs`): emit an ADVISORY and skip version-specific checks\n rather than raising.\n \"\"\"\n if raw is None:\n return None\n m = _VERSION_RE.match(raw)\n if not m:\n raise ValueError(f\"Unparseable Kafka version: {raw!r}\")\n parts = [int(g) for g in m.groups() if g is not None]\n return tuple(parts)\n\n\ndef _utcnow_iso() -> str:\n return _dt.datetime.now(_dt.timezone.utc).strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n\n\n# ---------------------------------------------------------------------------\n# Input validation — discovery contract\n# ---------------------------------------------------------------------------\n\nREQUIRED_TOP_LEVEL = (\"cluster_name\", \"kafka\", \"topology\", \"topics\")\nREQUIRED_KAFKA_FIELDS = (\"version\",)\nREQUIRED_TOPOLOGY_FIELDS = (\"num_brokers\",)\n\n\ndef validate_input(cfg: dict) -> None:\n \"\"\"Raise ValueError when the discovery contract lacks required fields or\n carries unknown values for closed enums (security.encryption_in_transit,\n security.authentication).\"\"\"\n missing = [k for k in REQUIRED_TOP_LEVEL if k not in cfg]\n if missing:\n raise ValueError(f\"discovery contract missing top-level fields: {missing}\")\n\n kafka_missing = [k for k in REQUIRED_KAFKA_FIELDS if k not in cfg[\"kafka\"]]\n if kafka_missing:\n raise ValueError(f\"kafka block missing required fields: {kafka_missing}\")\n\n topo_missing = [k for k in REQUIRED_TOPOLOGY_FIELDS if k not in cfg[\"topology\"]]\n if topo_missing:\n raise ValueError(f\"topology block missing required fields: {topo_missing}\")\n\n if not isinstance(cfg[\"topics\"], list):\n raise ValueError(\"topics must be a list\")\n\n # Strict enum validation on the security block. Both fields are required\n # if the block is present; if security is absent altogether the auth\n # pillar emits no evidence (treated as \"unknown / not provided\"), but\n # any value supplied must be from the closed enum.\n sec = cfg.get(\"security\")\n if sec is not None:\n encryption = sec.get(\"encryption_in_transit\")\n if encryption is not None and encryption not in ENCRYPTION_VALUES:\n raise ValueError(\n f\"security.encryption_in_transit={encryption!r} is not in the \"\n f\"allowed enum {sorted(ENCRYPTION_VALUES)}\"\n )\n authentication = sec.get(\"authentication\")\n if authentication is not None and authentication not in AUTHENTICATION_VALUES:\n raise ValueError(\n f\"security.authentication={authentication!r} is not in the \"\n f\"allowed enum {sorted(AUTHENTICATION_VALUES)}\"\n )\n\n # The target block carries sizing cost assumptions, not compatibility inputs.\n # No pillar reads it, but validate it here so a malformed discovery contract\n # fails before Assessment reaches the sizing step.\n target = cfg.get(\"target\")\n if target is not None:\n rack = target.get(\"rack_affined_consumers\")\n if rack is not None and not isinstance(rack, bool):\n raise ValueError(\n f\"target.rack_affined_consumers must be true, false, or null; got {rack!r}\"\n )\n discount = target.get(\"pricing_discount_pct\")\n if discount is not None:\n if isinstance(discount, bool) or not isinstance(discount, (int, float)):\n raise ValueError(\n f\"target.pricing_discount_pct must be a number or null; got {discount!r}\"\n )\n if not 0 \u003c= discount \u003c 100:\n raise ValueError(\n f\"target.pricing_discount_pct must be >= 0 and \u003c 100; got {discount!r}\"\n )\n\n\n# ---------------------------------------------------------------------------\n# Top-level orchestration\n# ---------------------------------------------------------------------------\n\nPILLARS = (\n (\"topology\", assess_topology),\n (\"kafka_version\", assess_kafka_version),\n (\"configs\", assess_configs),\n (\"auth\", assess_auth),\n (\"quotas\", assess_quotas),\n)\n\n\ndef assess(cfg: dict) -> dict:\n \"\"\"Run all five pillars; return the compatibility document.\"\"\"\n validate_input(cfg)\n pillars: dict[str, dict] = {}\n for name, fn in PILLARS:\n verdict, evidence = fn(cfg)\n pillars[name] = {\"verdict\": verdict, \"evidence\": evidence}\n\n overall = roll_up(p[\"verdict\"] for p in pillars.values())\n\n # Bucket each finding by ITS OWN severity, not the pillar's verdict.\n summary: dict[str, list[str]] = {\n \"action_required_codes\": [],\n \"advisory_codes\": [],\n \"info_codes\": [],\n }\n bucket_by_severity = {\n ACTION_REQUIRED: \"action_required_codes\",\n ADVISORY: \"advisory_codes\",\n INFO: \"info_codes\",\n }\n for p in pillars.values():\n for ev in p[\"evidence\"]:\n sev = ev.get(\"severity\", p[\"verdict\"]) # fallback for safety\n summary[bucket_by_severity[sev]].append(ev[\"code\"])\n\n return {\n \"cluster_name\": cfg[\"cluster_name\"],\n \"assessed_at\": _utcnow_iso(),\n \"overall\": overall,\n \"pillars\": pillars,\n \"summary\": summary,\n }\n\n\n# ---------------------------------------------------------------------------\n# CLI\n# ---------------------------------------------------------------------------\n\n\ndef _parse_args(argv: list[str]) -> argparse.Namespace:\n p = argparse.ArgumentParser(\n description=\"MSK Express compatibility assessor (file-only, deterministic).\"\n )\n p.add_argument(\n \"cluster_config\",\n type=Path,\n help=\"Path to migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json (discovery output).\",\n )\n p.add_argument(\n \"--out-dir\",\n type=Path,\n default=Path.cwd(),\n help=\"Directory to write compatibility.\u003ccluster_name>.json (default: cwd).\",\n )\n return p.parse_args(argv)\n\n\ndef main(argv: list[str] | None = None) -> int:\n args = _parse_args(sys.argv[1:] if argv is None else argv)\n cfg = json.loads(args.cluster_config.read_text())\n doc = assess(cfg)\n out_path = args.out_dir / f\"compatibility.{doc['cluster_name']}.json\"\n out_path.write_text(json.dumps(doc, indent=2, sort_keys=False) + \"\\n\")\n print(out_path, file=sys.stderr)\n print(f\"overall: {doc['overall']}\", file=sys.stderr)\n return 0\n\n\nif __name__ == \"__main__\":\n raise SystemExit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":47785,"content_sha256":"554d060a28292c3d79f09fe3729776b4b4435384434a09f5f8c2507bbca64920"},{"filename":"scripts/simulation_load_test_config.py","content":"#!/usr/bin/env python3\n# /// script\n# requires-python = \">=3.11\"\n# dependencies = []\n# ///\n\"\"\"\nDeterministic compute + guardrail engine for the MSK Express load-test simulation.\n\nThree actions:\n compute — given cluster sizing (instance_type, broker_count), emit all derived\n values: cluster capacity, auto-sized client fleet, and per-test\n default parameters + bounds. Feeds CFN parameters and test templates.\n render — given the same sizing, write a local, ready-to-deploy copy of the\n CloudFormation template with the customer's sizing AND the computed\n fleet baked in as parameter Defaults. The source template stays static;\n this emits a derived artifact to a working-directory path so the customer\n deploys from a local file (never from the skill source). The Kafka client\n is installed post-deploy via SSM, so nothing client-related is baked here.\n validate — given a test type and a proposed config, return PASS/REJECT plus\n inline WARNINGS. The agent loops back on REJECT, surfaces WARNINGS\n as one-sentence notes and proceeds.\n\nRun with:\n uv run \u003cscript> compute --instance-type express.m7g.4xlarge --broker-count 12\n uv run \u003cscript> render --instance-type express.m7g.4xlarge --broker-count 12 \\\n --kafka-version 3.9.x.kraft \\\n --output migrate-to-msk-skill-artifacts/simulation/simulation-stack.yaml\n uv run \u003cscript> validate --test e2e_latency --config '{\"throughput_mbps\":10,...}'\n uv run \u003cscript> validate --test broker_restart --config '{...}' \\\n --instance-type express.m7g.4xlarge --broker-count 12\n\"\"\"\nimport argparse\nimport json\nimport math\nimport sys\nfrom pathlib import Path\n\n# The static source template lives alongside this script in the skill package.\n# It is resolved internally and never surfaced to the customer — render() emits a\n# derived local copy, and all customer-facing commands point at that copy instead.\nSOURCE_TEMPLATE = Path(__file__).resolve().parent.parent / \"assets\" / \"simulation-stack.yaml\"\n\n# Express per-broker INGRESS throttle limits (MB/s), from AWS Express docs:\n# (sustained, max_quota)\n# sustained = recommended threshold (no performance degradation up to here);\n# max_quota = hard ceiling (the cluster throttles read/write traffic beyond it).\n# Full table including egress lives in references/simulation.md.\nPER_BROKER_INGRESS = {\n \"express.m7g.large\": (15.6, 23.4),\n \"express.m7g.xlarge\": (31.2, 46.8),\n \"express.m7g.2xlarge\": (62.5, 93.7),\n \"express.m7g.4xlarge\": (124.9, 187.5),\n \"express.m7g.8xlarge\": (250.0, 375.0),\n \"express.m7g.12xlarge\": (375.0, 562.5),\n \"express.m7g.16xlarge\": (500.0, 750.0),\n}\n\n# The only valid Express version + metadata-mode combinations, as the exact CFN\n# KafkaVersion strings the template's AllowedValues accepts. On Express, 3.9 is\n# KRaft-only (no ZooKeeper variant) and KRaft is unavailable below 3.9, so these\n# three are the whole set. Validated here so a freeform/unsupported version is\n# rejected locally instead of failing ~40 min into create-stack.\nSUPPORTED_KAFKA_VERSIONS = (\"3.6.0\", \"3.8.x\", \"3.9.x.kraft\")\n\n# Fixed client fleet instance type -- one size for all clusters, only count varies.\n# c5.2xlarge: 10 Gbps sustained baseline (no burst credits), $0.34/hr.\n# 400 MB/s is a conservative per-node Kafka producer throughput target accounting\n# for TLS overhead, acks=all latency, and batching inefficiency.\nFLEET_INSTANCE_TYPE = \"c5.2xlarge\"\nFLEET_NODE_THROUGHPUT_MBPS = 400\n\n\ndef compute(instance_type, broker_count, kafka_version=None):\n if instance_type not in PER_BROKER_INGRESS:\n raise ValueError(\n f\"unknown instance_type {instance_type!r}; valid: {list(PER_BROKER_INGRESS)}\"\n )\n if broker_count \u003c 3 or broker_count % 3 != 0:\n raise ValueError(\"broker_count must be a multiple of 3 and >= 3\")\n # Only validate the version when one is supplied; sizing math doesn't need it,\n # but rejecting an unsupported value here keeps a freeform answer from reaching\n # create-stack. 3.9 ZooKeeper (e.g. \"3.9.x\") is intentionally not in the set.\n if kafka_version is not None and kafka_version not in SUPPORTED_KAFKA_VERSIONS:\n raise ValueError(\n f\"unsupported kafka_version {kafka_version!r}; valid: \"\n f\"{list(SUPPORTED_KAFKA_VERSIONS)} (3.9 is KRaft-only on Express)\"\n )\n\n per_broker_sustained, per_broker_max = PER_BROKER_INGRESS[instance_type]\n cluster_sustained_ingress = round(per_broker_sustained * broker_count, 1)\n cluster_max_ingress = round(per_broker_max * broker_count, 1)\n fleet_bandwidth = FLEET_NODE_THROUGHPUT_MBPS\n # Provision 1.5x the cluster's MAXIMUM ingress quota so the fleet can drive the\n # cluster all the way to its throttle ceiling with headroom and never be the\n # bottleneck (even when deliberately probing past the limit).\n fleet_producer_count = max(2, math.ceil(cluster_max_ingress * 1.5 / fleet_bandwidth))\n fleet_consumer_count = fleet_producer_count\n fleet_max_mbps = fleet_producer_count * fleet_bandwidth\n default_target = round(cluster_sustained_ingress * 0.8)\n\n return {\n \"instance_type\": instance_type,\n \"broker_count\": broker_count,\n \"supported_kafka_versions\": list(SUPPORTED_KAFKA_VERSIONS),\n \"cluster_sustained_ingress_mbps\": cluster_sustained_ingress,\n \"cluster_max_ingress_mbps\": cluster_max_ingress,\n \"fleet_instance_type\": FLEET_INSTANCE_TYPE,\n \"fleet_instance_bandwidth_mbps\": fleet_bandwidth,\n \"fleet_producer_count\": fleet_producer_count,\n \"fleet_consumer_count\": fleet_consumer_count,\n \"fleet_max_mbps\": fleet_max_mbps,\n \"default_target_throughput_mbps\": default_target,\n \"tests\": {\n \"e2e_latency\": {\n \"throughput_mbps\": {\"default\": 10, \"min\": 1, \"max\": fleet_max_mbps},\n \"duration_minutes\": {\"default\": 10, \"min\": 5, \"max\": 120},\n \"num_producers\": {\"default\": 1, \"min\": 1, \"max\": fleet_producer_count},\n \"num_consumers\": {\"default\": 1, \"min\": 1, \"max\": fleet_consumer_count},\n },\n \"broker_restart\": {\n \"target_throughput_mbps\": {\n \"default\": default_target,\n \"min\": 1,\n \"max\": fleet_max_mbps,\n },\n \"duration_minutes\": {\"default\": 15, \"min\": 10, \"max\": 120},\n \"reboot_at_minute\": {\"default\": 5, \"min\": 2, \"max\": None}, # max = duration-2\n \"num_producers\": {\n \"default\": fleet_producer_count,\n \"min\": 1,\n \"max\": fleet_producer_count,\n },\n },\n },\n }\n\n\ndef _set_param_default(template_text, param_name, value):\n \"\"\"Set the `Default:` of a top-level CloudFormation parameter in-place.\n\n The template is static and authored with each parameter at 2-space indent and\n its properties at 4-space indent. We locate the parameter block and rewrite its\n existing `Default:` line (every baked parameter already declares one). Raises if\n the parameter or its Default line is missing, so a template drift fails loudly\n here rather than silently producing an unsized stack.\n \"\"\"\n lines = template_text.splitlines()\n out = []\n i = 0\n n = len(lines)\n replaced = False\n header = f\" {param_name}:\"\n while i \u003c n:\n line = lines[i]\n out.append(line)\n if line == header:\n # Walk this parameter's property lines (indented deeper than 2 spaces).\n i += 1\n block_done = False\n while i \u003c n:\n prop = lines[i]\n # End of block: a line that isn't blank and is indented \u003c= 2 spaces.\n if prop.strip() and (len(prop) - len(prop.lstrip(\" \"))) \u003c= 2:\n break\n stripped = prop.lstrip(\" \")\n if stripped.startswith(\"Default:\") and not block_done:\n indent = \" \" * (len(prop) - len(stripped))\n out.append(f\"{indent}Default: {value}\")\n replaced = True\n block_done = True\n else:\n out.append(prop)\n i += 1\n continue\n i += 1\n if not replaced:\n raise ValueError(\n f\"could not set Default for parameter {param_name!r}; the source template \"\n \"may have drifted (expected a 2-space-indented block with a Default line)\"\n )\n # Preserve a trailing newline if the original had one.\n text = \"\\n\".join(out)\n if template_text.endswith(\"\\n\"):\n text += \"\\n\"\n return text\n\n\ndef render(instance_type, broker_count, kafka_version, output_path):\n \"\"\"Write a local, ready-to-deploy copy of the template with sizing baked in.\n\n Reads the static source template internally, bakes the customer's sizing and the\n computed fleet counts in as parameter Defaults, and writes the result to\n output_path. The customer deploys from that local file; the skill source path is\n never exposed.\n\n The output path is resolved to an ABSOLUTE path (relative paths resolve against\n the current working directory) and the resolved location is returned, so the\n caller never has to guess where the file landed. As a safety rail, the resolved\n path must NOT be inside the skill package -- that both keeps the skill source\n clean and prevents the customer-facing artifact from ever pointing back at the\n install directory.\n \"\"\"\n caps = compute(instance_type, broker_count, kafka_version)\n\n # Resolve to absolute up front so the returned path is unambiguous regardless of\n # the directory the script was launched from.\n out = Path(output_path).expanduser().resolve()\n skill_root = SOURCE_TEMPLATE.parent.parent # .../skills/migrate-to-msk\n if out == skill_root or skill_root in out.parents:\n raise ValueError(\n f\"refusing to write the rendered template inside the skill package \"\n f\"({skill_root}); choose an output path under the customer's working \"\n f\"directory instead, e.g. \"\n f\"$(pwd)/migrate-to-msk-skill-artifacts/simulation/simulation-stack.yaml\"\n )\n\n template_text = SOURCE_TEMPLATE.read_text()\n\n baked = {\n \"InstanceType\": instance_type,\n \"BrokerCount\": broker_count,\n \"KafkaVersion\": kafka_version,\n \"ClientInstanceType\": caps[\"fleet_instance_type\"],\n \"ProducerCount\": caps[\"fleet_producer_count\"],\n \"ConsumerCount\": caps[\"fleet_consumer_count\"],\n }\n for name, value in baked.items():\n template_text = _set_param_default(template_text, name, value)\n\n out.parent.mkdir(parents=True, exist_ok=True)\n out.write_text(template_text)\n\n return {\n \"output_path\": str(out),\n \"deploy_command\": (\n f\"aws cloudformation create-stack --stack-name msk-express-simulation \"\n f\"--template-body file://{out} --capabilities CAPABILITY_IAM \"\n f\"--region \u003cregion>\"\n ),\n \"baked_parameters\": baked,\n \"fleet_producer_count\": caps[\"fleet_producer_count\"],\n \"fleet_consumer_count\": caps[\"fleet_consumer_count\"],\n \"fleet_instance_type\": caps[\"fleet_instance_type\"],\n }\n\n\ndef _check_range(name, val, lo, hi, errors):\n if val \u003c lo or (hi is not None and val > hi):\n errors.append(f\"{name}={val} is outside the allowed range [{lo}, {hi}]\")\n\n\ndef _throughput_warnings(name, tput, sustained, max_quota, warnings):\n \"\"\"Tiered throughput warning. Returns True if a degradation/throttle warning fired.\n > max_quota -> cluster THROTTLES (hard ceiling; cannot be sustained).\n > sustained -> performance DEGRADATION (elevated latency), but no throttling yet.\"\"\"\n if tput > max_quota:\n warnings.append(\n f\"{name}={tput} MB/s exceeds the cluster's maximum ingress quota \"\n f\"({max_quota} MB/s) — MSK Express throttles read/write traffic beyond this, \"\n f\"so the cluster cannot actually sustain that rate.\"\n )\n return True\n if tput > sustained:\n warnings.append(\n f\"{name}={tput} MB/s is above the cluster's sustained/recommended ingress \"\n f\"({sustained} MB/s); it can run but expect performance degradation \"\n f\"(elevated latency), with no throttling until the {max_quota} MB/s ceiling.\"\n )\n return True\n return False\n\n\ndef validate(test, config, caps):\n \"\"\"caps = output of compute(). Returns {decision, errors, warnings}.\"\"\"\n errors: list[str] = []\n warnings: list[str] = []\n fleet_max = caps[\"fleet_max_mbps\"]\n sustained = caps[\"cluster_sustained_ingress_mbps\"]\n max_quota = caps[\"cluster_max_ingress_mbps\"]\n fleet_bw = caps[\"fleet_instance_bandwidth_mbps\"]\n fleet_prod = caps[\"fleet_producer_count\"]\n\n if test == \"e2e_latency\":\n tput = config[\"throughput_mbps\"]\n dur = config[\"duration_minutes\"]\n prod = config[\"num_producers\"]\n cons = config[\"num_consumers\"]\n _check_range(\"throughput_mbps\", tput, 1, fleet_max, errors)\n _check_range(\"duration_minutes\", dur, 5, 120, errors)\n _check_range(\"num_producers\", prod, 1, fleet_prod, errors)\n _check_range(\"num_consumers\", cons, 1, caps[\"fleet_consumer_count\"], errors)\n if not _throughput_warnings(\"throughput_mbps\", tput, sustained, max_quota, warnings):\n if tput > sustained * 0.6:\n warnings.append(\n f\"throughput_mbps={tput} is above 60% of sustained ingress; high load \"\n f\"can inflate latency. For a clean signal, consider \u003c= {round(sustained * 0.5)} MB/s.\"\n )\n if prod \u003c math.ceil(tput / fleet_bw):\n warnings.append(\n f\"num_producers={prod} may not sustain {tput} MB/s; \"\n f\"consider >= {math.ceil(tput / fleet_bw)} producers.\"\n )\n\n elif test == \"broker_restart\":\n tput = config[\"target_throughput_mbps\"]\n dur = config[\"duration_minutes\"]\n reboot = config[\"reboot_at_minute\"]\n prod = config[\"num_producers\"]\n _check_range(\"target_throughput_mbps\", tput, 1, fleet_max, errors)\n _check_range(\"duration_minutes\", dur, 10, 120, errors)\n _check_range(\"reboot_at_minute\", reboot, 2, dur - 2, errors)\n _check_range(\"num_producers\", prod, 1, fleet_prod, errors)\n if not _throughput_warnings(\"target_throughput_mbps\", tput, sustained, max_quota, warnings):\n if tput > sustained * 0.6:\n warnings.append(\n f\"target_throughput_mbps={tput} is above 60% of sustained ingress; \"\n \"broker-CPU headroom during the restart is limited, so recovery may take longer.\"\n )\n if prod \u003c math.ceil(tput / fleet_bw):\n warnings.append(\n f\"num_producers={prod} may not sustain {tput} MB/s; \"\n f\"consider >= {math.ceil(tput / fleet_bw)} producers.\"\n )\n else:\n errors.append(f\"unknown test {test!r}; valid: e2e_latency, broker_restart\")\n\n return {\n \"decision\": \"REJECT\" if errors else \"PASS\",\n \"errors\": errors,\n \"warnings\": warnings,\n }\n\n\ndef parse_args():\n p = argparse.ArgumentParser(\n description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter\n )\n sub = p.add_subparsers(dest=\"action\", required=True)\n\n c = sub.add_parser(\"compute\", help=\"Emit derived sizing + per-test defaults/bounds.\")\n c.add_argument(\"--instance-type\", default=\"express.m7g.4xlarge\")\n c.add_argument(\"--broker-count\", type=int, default=12)\n c.add_argument(\n \"--kafka-version\",\n default=None,\n help=\"Optional. If given, rejected unless one of: \" + \", \".join(SUPPORTED_KAFKA_VERSIONS),\n )\n\n r = sub.add_parser(\n \"render\",\n help=\"Write a local ready-to-deploy template with sizing baked in as Defaults.\",\n )\n r.add_argument(\"--instance-type\", required=True)\n r.add_argument(\"--broker-count\", type=int, required=True)\n r.add_argument(\n \"--kafka-version\",\n required=True,\n help=\"Required. One of: \" + \", \".join(SUPPORTED_KAFKA_VERSIONS),\n )\n r.add_argument(\n \"--output\",\n required=True,\n help=\"Path for the filled template. Resolved to an absolute path \"\n \"(relative paths resolve against the current directory); pass an absolute \"\n \"path under the customer's working directory, e.g. \"\n \"$(pwd)/migrate-to-msk-skill-artifacts/simulation/simulation-stack.yaml. \"\n \"Must not be inside the skill package.\",\n )\n\n v = sub.add_parser(\"validate\", help=\"Validate a proposed test config.\")\n v.add_argument(\"--test\", required=True, choices=[\"e2e_latency\", \"broker_restart\"])\n v.add_argument(\"--config\", required=True, help=\"JSON object of test parameters.\")\n v.add_argument(\"--instance-type\", default=\"express.m7g.4xlarge\")\n v.add_argument(\"--broker-count\", type=int, default=12)\n return p.parse_args()\n\n\ndef main():\n args = parse_args()\n try:\n if args.action == \"render\":\n result = render(args.instance_type, args.broker_count, args.kafka_version, args.output)\n print(json.dumps(result, indent=2))\n return\n # validate has no --kafka-version arg; getattr keeps the shared compute() call safe.\n caps = compute(args.instance_type, args.broker_count, getattr(args, \"kafka_version\", None))\n if args.action == \"compute\":\n print(json.dumps(caps, indent=2))\n else:\n result = validate(args.test, json.loads(args.config), caps)\n print(json.dumps(result, indent=2))\n if result[\"decision\"] == \"REJECT\":\n sys.exit(1)\n except (ValueError, KeyError, json.JSONDecodeError, OSError) as e:\n print(json.dumps({\"error\": str(e)}), file=sys.stderr)\n sys.exit(2)\n\n\nif __name__ == \"__main__\":\n main()\n","content_type":"text/x-python; charset=utf-8","language":"python","size":18113,"content_sha256":"e8cf51c9f1ef743e8e76973b9bdb7f31635ad0c977377088a53ebf4cdcfacb6d"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Migrating to MSK Express","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Overview","type":"text"}]},{"type":"paragraph","content":[{"text":"This skill helps customers migrate self-managed Apache Kafka workloads to Amazon MSK Express. It provides three phases — ","type":"text"},{"text":"Discovery","type":"text","marks":[{"type":"strong"}]},{"text":", ","type":"text"},{"text":"Assessment","type":"text","marks":[{"type":"strong"}]},{"text":", and an optional ","type":"text"},{"text":"Simulation","type":"text","marks":[{"type":"strong"}]},{"text":" — that can be run end-to-end or individually depending on the customer's needs.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Scope","type":"text"}]},{"type":"paragraph","content":[{"text":"This skill covers migrations from ","type":"text"},{"text":"self-managed Apache Kafka","type":"text","marks":[{"type":"strong"}]},{"text":" (on-premises, EC2, Docker, Kubernetes, or other non-MSK deployments) to MSK Express. Migrations from ","type":"text"},{"text":"MSK Standard (Provisioned) to MSK Express","type":"text","marks":[{"type":"strong"}]},{"text":" are out of scope.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Prerequisites","type":"text"}]},{"type":"paragraph","content":[{"text":"The AWS MCP server is recommended for documentation lookups and informational questions, but is not required. The assessment scripts are pure file processors with no AWS API calls.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Intent Routing","type":"text"}]},{"type":"paragraph","content":[{"text":"Route the customer's request based on their intent:","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"1. Open/exploratory question (\"How do I migrate to MSK?\")","type":"text"}]},{"type":"paragraph","content":[{"text":"Explain what this skill offers:","type":"text"}]},{"type":"blockquote","content":[{"type":"paragraph","content":[{"text":"This skill helps you migrate to MSK Express in three phases:","type":"text"}]},{"type":"paragraph","content":[{"text":"Phase 1 — Discovery:","type":"text","marks":[{"type":"strong"}]},{"text":" Inventory your source Kafka cluster — brokers, topics, partition counts, configs, authentication, and workload metrics — plus two target decisions that drive cost: consumer rack affinity and any negotiated AWS pricing. I can discover this from IaC files (Terraform, CDK, Docker Compose, Kubernetes manifests), provide commands for you to run on your cluster, or you can provide the information manually. Output: ","type":"text"},{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"paragraph","content":[{"text":"Phase 2 — Assessment:","type":"text","marks":[{"type":"strong"}]},{"text":" Validate your cluster against MSK Express across 5 compatibility pillars (topology, Kafka version, configs, auth, quotas) and produce a target Express specification using the managing-amazon-msk Skill's pricing logic. I'll flag what Express will refuse vs what Express will silently convert. Outputs: ","type":"text"},{"text":"compatibility.\u003ccluster_name>.json","type":"text","marks":[{"type":"code_inline"}]},{"text":", the pricing results in Markdown ","type":"text"},{"text":"msk_sizing_pricing.md","type":"text","marks":[{"type":"code_inline"}]},{"text":", and ","type":"text"},{"text":"msk-sizing-inputs.\u003ccluster_name>.json","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"paragraph","content":[{"text":"Phase 3 — Simulation:","type":"text","marks":[{"type":"strong"}]},{"text":" Spin up an MSK Express cluster with load-testing infrastructure to see how Express performs on your own workload, then run a vended test (End-to-End Latency or Broker Restart Under Load) and review the results on a CloudWatch dashboard.","type":"text"}]},{"type":"paragraph","content":[{"text":"Data replication:","type":"text","marks":[{"type":"strong"}]},{"text":" For migrating data to your Express cluster, you can use MSK Replicator. I can provide guidance on setup and configuration.","type":"text"}]},{"type":"paragraph","content":[{"text":"Where would you like to start? I can begin with discovery if you point me to your infrastructure code or describe your cluster, or jump to assessment if you already have a ","type":"text"},{"text":"cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" file, or go straight to simulation if you already know your target Express configuration.","type":"text"}]}]},{"type":"paragraph","content":[{"text":"Guardrails for this overview response:","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"This response is an overview and a routing question only. Do NOT begin, simulate, or pre-empt any phase.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Do NOT produce or estimate assessment output here — no verdicts, pillar findings, compatibility conclusions, broker counts, instance recommendations, or cost figures. Those values exist only after you run the Phase 2 scripts against a real ","type":"text"},{"text":"cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Do NOT open, read, or summarize the internals of ","type":"text"},{"text":"compatibility.py","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"simulation_load_test_config.py","type":"text","marks":[{"type":"code_inline"}]},{"text":", or the reference files to explain how a phase works. Describe the phases at the level shown above; do not walk the customer through the implementation.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"When the customer chooses a phase, run that phase's scripts or flow to produce real results. Always operate the skill to answer — never answer from having read its source. For the exact commands, see \"Running the assessment\" in ","type":"text"},{"text":"references/assessment-compatibility.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/assessment-compatibility.md","title":null}}]},{"text":" for Phase 2, and ","type":"text"},{"text":"references/simulation.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/simulation.md","title":null}}]},{"text":" for Phase 3.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"2. Discovery intent (DEFAULT when IaC files are provided)","type":"text"}]},{"type":"paragraph","content":[{"text":"If the customer provides a directory path, IaC files, or says \"here's our infra\" — this is discovery intent. Run ONLY Phase 1 (Discovery). Do NOT run assessment, do NOT suggest migration steps, do NOT mention blockers or compatibility. Produce the ","type":"text"},{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" file and stop.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"3. Assessment intent","type":"text"}]},{"type":"paragraph","content":[{"text":"Customer explicitly asks to assess or has a ","type":"text"},{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" file already produced. Run Phase 2 (Assessment) only.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"4. Simulation intent","type":"text"}]},{"type":"paragraph","content":[{"text":"Customer wants to test MSK Express with their workload. They can provide cluster sizing directly (instance type, broker count, Kafka version) or reference an earlier assessment. Proceed directly to ","type":"text"},{"text":"Phase 3 — Simulation","type":"text","marks":[{"type":"link","attrs":{"href":"#phase-3--simulation-optional","title":null}}]},{"text":". An assessment is helpful but not required — the simulation asks for sizing inputs directly.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"5. Informational questions","type":"text"}]},{"type":"paragraph","content":[{"text":"Customer asks about Express capabilities, constraints, configuration differences, authentication support, pricing, or compaction behavior without providing cluster-specific data. Use AWS documentation tools (","type":"text"},{"text":"aws___search_documentation","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"aws___read_documentation","type":"text","marks":[{"type":"code_inline"}]},{"text":") if available to look up the answer from MSK Express documentation. If MCP tools are not available, reference the ","type":"text"},{"text":"MSK Express documentation","type":"text","marks":[{"type":"link","attrs":{"href":"https://docs.aws.amazon.com/msk/latest/developerguide/msk-broker-types-express.html","title":null}}]},{"text":" and answer based on knowledge of AWS MSK.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"6. Migration strategy questions","type":"text"}]},{"type":"paragraph","content":[{"text":"Customer asks about MSK Replicator compatibility, version upgrade paths, MirrorMaker 2, or migration strategies. MSK Replicator is the native AWS-supported solution for data replication and works for both MSK-to-MSK and non-MSK-to-MSK migrations. Use AWS documentation tools (","type":"text"},{"text":"aws___search_documentation","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"aws___read_documentation","type":"text","marks":[{"type":"code_inline"}]},{"text":") if available to retrieve current requirements and supported configurations. If MCP tools are not available, reference the ","type":"text"},{"text":"MSK Replicator documentation","type":"text","marks":[{"type":"link","attrs":{"href":"https://docs.aws.amazon.com/msk/latest/developerguide/msk-replicator.html","title":null}}]},{"text":" and answer based on knowledge of AWS MSK.","type":"text"}]},{"type":"hr","attrs":{"markup":"---"}},{"type":"heading","attrs":{"level":2},"content":[{"text":"Phase 1 — Discovery","type":"text"}]},{"type":"paragraph","content":[{"text":"Purpose:","type":"text","marks":[{"type":"strong"}]},{"text":" Inventory the source cluster to build a migration profile.","type":"text"}]},{"type":"paragraph","content":[{"text":"Input:","type":"text","marks":[{"type":"strong"}]},{"text":" One of:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"A directory path containing IaC files (CDK, CloudFormation, Docker Compose, Kubernetes manifests, Terraform)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Output from Kafka CLI commands the customer runs on their cluster","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Manual information provided by the customer in conversation","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Output:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" — saved to the working directory.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Discovery rules","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Before doing ANYTHING else in discovery, you MUST read ","type":"text"},{"text":"references/discovery.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/discovery.md","title":null}}]},{"text":" in full. It defines the input methods, the REQUIRED response template, the forbidden content, and the ","type":"text"},{"text":"cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" schema. Do NOT respond until you have read it, and follow its template EXACTLY.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ALWAYS save ","type":"text"},{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the working directory.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Do NOT proceed to Phase 2 without explicit customer confirmation.","type":"text"}]}]}]},{"type":"hr","attrs":{"markup":"---"}},{"type":"heading","attrs":{"level":2},"content":[{"text":"Phase 2 — Assessment","type":"text"}]},{"type":"paragraph","content":[{"text":"Purpose:","type":"text","marks":[{"type":"strong"}]},{"text":" Assess the cluster against MSK Express requirements and produce a target Express specification (instance type, broker count, monthly cost projection).","type":"text"}]},{"type":"paragraph","content":[{"text":"Input:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" from Phase 1.","type":"text"}]},{"type":"paragraph","content":[{"text":"Outputs:","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/compatibility.\u003ccluster_name>.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" — five-pillar verdict.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/msk_sizing_pricing.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — the managing-amazon-msk Skill's pricing report with broker count and cost recommendations.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"migrate-to-msk-skill-artifacts/\u003ccluster_name>/msk-sizing-inputs.\u003ccluster_name>.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" — a record of the six input values for sizing logic.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Assessment has two independent halves; run them in either order, and a failure in one does not block the other:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Compatibility","type":"text","marks":[{"type":"strong"}]},{"text":" — ","type":"text"},{"text":"scripts/compatibility.py","type":"text","marks":[{"type":"code_inline"}]},{"text":", a pure file processor (no live AWS API calls) run via ","type":"text"},{"text":"uv run","type":"text","marks":[{"type":"code_inline"}]},{"text":" with PEP 723 inline dependencies. It validates the source across five pillars — topology, Kafka version, configs, auth, and quotas — and emits one verdict per pillar (","type":"text"},{"text":"INFO","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"ADVISORY","type":"text","marks":[{"type":"code_inline"}]},{"text":", or ","type":"text"},{"text":"ACTION_REQUIRED","type":"text","marks":[{"type":"code_inline"}]},{"text":", worst-of for the overall). Use those three strings verbatim.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Sizing","type":"text","marks":[{"type":"strong"}]},{"text":" — not a script in this skill. Load the managing-amazon-msk Skill and run its ","type":"text"},{"text":"scripts/msk_sizing.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" with the workload inputs derived from ","type":"text"},{"text":"cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":", passing ","type":"text"},{"text":"--broker-classes express","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"--region","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the target Region. The script prices from the live AWS Pricing API when credentials are available and from an embedded ","type":"text"},{"text":"us-east-1","type":"text","marks":[{"type":"code_inline"}]},{"text":" snapshot when they are not.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Assessment rules","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Read ","type":"text"},{"text":"references/assessment-compatibility.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/assessment-compatibility.md","title":null}}]},{"text":" before responding. It carries the invocation commands, the per-pillar thresholds and evidence codes, the verdict definitions, the full forbidden-behavior list, and the ","type":"text"},{"text":"required response template","type":"text","marks":[{"type":"strong"}]},{"text":" covering both artifacts. Do not freestyle the post-script summary.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Read ","type":"text"},{"text":"references/assessment-sizing.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/assessment-sizing.md","title":null}}]},{"text":" before running sizing. It carries the input derivations (several are not one-to-one — getting them wrong silently produces a wrong broker count), the ","type":"text"},{"text":"target","type":"text","marks":[{"type":"code_inline"}]},{"text":"-block flags for rack affinity and negotiated discounts, the Region and pricing-source handling, the Express-only presentation rule, and the source-footprint comparison.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Surface any ","type":"text"},{"text":"ACTION_REQUIRED","type":"text","marks":[{"type":"code_inline"}]},{"text":" evidence to the user for awareness, but do not gate further phases on it. Express may still accept the workload with mitigations.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Do NOT pivot back into discovery.","type":"text","marks":[{"type":"strong"}]},{"text":" Assessment operates on the existing ","type":"text"},{"text":"cluster-config.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" as-is. Partial data is fine — the scripts emit ADVISORY evidence (","type":"text"},{"text":"METRICS_MISSING","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"AZ_COUNT_UNKNOWN","type":"text","marks":[{"type":"code_inline"}]},{"text":", etc.) for missing fields; surface those findings and stop. Do not propose Kafka CLI commands, IaC walks, scripts, or questionnaires to fill the gaps.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Report broker counts and costs only as read verbatim from the sizing script output. Never round, re-derive, or estimate them. Report the ","type":"text"},{"text":"pricing source","type":"text","marks":[{"type":"code_inline"}]},{"text":" the script used alongside the costs; when it is ","type":"text"},{"text":"static","type":"text","marks":[{"type":"code_inline"}]},{"text":", the rates are an embedded snapshot, not confirmed current pricing.","type":"text"}]}]}]},{"type":"hr","attrs":{"markup":"---"}},{"type":"heading","attrs":{"level":2},"content":[{"text":"Phase 3 — Simulation (optional)","type":"text"}]},{"type":"paragraph","content":[{"text":"Deploy a temporary, isolated MSK Express cluster and client fleet in the customer's account so they can see how Express performs on their own workload, then run one of two vended tests (End-to-End Latency, Broker Restart Under Load) and hand over a CloudWatch dashboard. Follow the 12-step conversational flow and all deploy, sizing, and guardrail details in ","type":"text"},{"text":"references/simulation.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/simulation.md","title":null}}]},{"text":"; the deterministic artifacts it drives are ","type":"text"},{"text":"scripts/simulation_load_test_config.py","type":"text","marks":[{"type":"link","attrs":{"href":"scripts/simulation_load_test_config.py","title":null}}]},{"text":" and the static ","type":"text"},{"text":"assets/simulation-stack.yaml","type":"text","marks":[{"type":"link","attrs":{"href":"assets/simulation-stack.yaml","title":null}}]},{"text":".","type":"text"}]},{"type":"hr","attrs":{"markup":"---"}},{"type":"heading","attrs":{"level":2},"content":[{"text":"Execution model","type":"text"}]},{"type":"paragraph","content":[{"text":"Scripts run on the customer's local machine via ","type":"text"},{"text":"uv run","type":"text","marks":[{"type":"code_inline"}]},{"text":". They declare their own dependencies (PEP 723) and are pure file processors — no AWS API calls, no network access, and no third-party dependencies (standard library only).","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Security Considerations","type":"text"}]},{"type":"paragraph","content":[{"text":"Apply these controls at every phase. For additional detail, see ","type":"text"},{"text":"MSK Security best practices","type":"text","marks":[{"type":"link","attrs":{"href":"https://docs.aws.amazon.com/msk/latest/developerguide/security.html","title":null}}]},{"text":" and ","type":"text"},{"text":"MSK IAM access control","type":"text","marks":[{"type":"link","attrs":{"href":"https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html","title":null}}]},{"text":".","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Encryption in transit (mandatory).","type":"text","marks":[{"type":"strong"}]},{"text":" Enforce TLS for client-broker traffic on the MSK Express target (","type":"text"},{"text":"EncryptionInTransit.ClientBroker = TLS","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Encryption at rest (mandatory).","type":"text","marks":[{"type":"strong"}]},{"text":" Provision the target cluster with a customer-managed KMS key (or AWS-managed if your compliance posture allows).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Authentication — prefer IAM over long-lived credentials.","type":"text","marks":[{"type":"strong"}]},{"text":" Configure the MSK Express target with IAM authentication as the sole client auth method. This gives ephemeral, role-based credentials with full CloudTrail coverage.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Credential storage — use AWS Secrets Manager.","type":"text","marks":[{"type":"strong"}]},{"text":" Store SASL/SCRAM and TLS credentials for source cluster access in Secrets Manager. Never pass passwords as CLI arguments.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Network isolation.","type":"text","marks":[{"type":"strong"}]},{"text":" Deploy MSK clusters in private subnets. Use security groups scoped to specific CIDR ranges or security group references. Do NOT use 0.0.0.0/0 ingress rules.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"CloudTrail logging and CloudWatch alarms.","type":"text","marks":[{"type":"strong"}]},{"text":" Ensure CloudTrail is enabled in the target account and covers ","type":"text"},{"text":"kafka.amazonaws.com","type":"text","marks":[{"type":"code_inline"}]},{"text":" API calls. Configure alarms:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ClientAuthenticationFailure","type":"text","marks":[{"type":"code_inline"}]},{"text":" — surge indicates credential problems or attack","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ConnectionCloseCount","type":"text","marks":[{"type":"code_inline"}]},{"text":" — abnormal spike may indicate connection-flooding","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"CloudTrail metric filters for denied ","type":"text"},{"text":"kafka-cluster:*","type":"text","marks":[{"type":"code_inline"}]},{"text":" actions","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Connection-rate alarms approaching the 100 conn/sec/broker IAM limit","type":"text"}]}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Sensitive data handling.","type":"text","marks":[{"type":"strong"}]},{"text":" Discovery and assessment outputs contain broker addresses, auth hints, and broker config values. Treat these as sensitive — do not paste into public channels or ticketing systems without redaction.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Troubleshooting","type":"text"}]},{"type":"paragraph","content":[{"text":"Per-pillar findings, including the source topology and out-of-range config cases, are explained in ","type":"text"},{"text":"references/assessment-compatibility.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/assessment-compatibility.md","title":null}}]},{"text":".","type":"text"}]}]},"metadata":{"date":"2026-11-02","name":"migrate-to-msk","author":"@skillopedia","source":{"stars":3376,"repo_name":"agent-toolkit-for-aws","origin_url":"https://github.com/aws/agent-toolkit-for-aws/blob/HEAD/skills/specialized-skills/analytics-skills/migrate-to-msk/SKILL.md","repo_owner":"aws","body_sha256":"146ca7b67afcee3fa64581db6a6e324cc7e133af39f21aa88f25d10c27d30671","cluster_key":"65909f0b70d00b1316e3cae6ef24b4ffa127c792aed12631a0043dab3c2e8897","clean_bundle":{"format":"clean-skill-bundle-v1","source":"aws/agent-toolkit-for-aws/skills/specialized-skills/analytics-skills/migrate-to-msk/SKILL.md","text_bytes":208418,"source_commit":"1761e96813ebcbd43ed81d19e837219033f0afe5","skills_sh_rank":4382,"attachment_count":7,"text_attachments":7,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/specialized-skills/analytics-skills/migrate-to-msk/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"data-analytics","category_label":"Data"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"data-analytics","import_tag":"skills-sh-top20k-2026-11","description":"Helps migrate self-managed Apache Kafka workloads to Amazon MSK Express. Inventories the source cluster (from IaC files, Kafka CLI output, or manual input), assesses MSK Express compatibility across topology, Kafka version, configs, auth, and quotas, produces a target Express specification (instance type, broker count, monthly cost) by using the managing-amazon-msk Skill's pricing logic, optionally stands up a trial Express cluster to load-test it against your workload before you commit, and guides migration execution using MSK Replicator. Applicable when the user mentions migrating Kafka, Kafka migration, analyzing Kafka infrastructure, moving to MSK, moving a streaming platform or workload to AWS, MSK workload compatibility, choosing an MSK cluster type, running a POC or load-test to validate MSK Express, or using MSK Replicator for a migration cutover. Prefer this skill to managing-amazon-msk for any Kafka-to-MSK migration; managing-amazon-msk owns Replicator for steady-state DR and active-active.\n"}},"renderedAt":1793666313287}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.