Migrate To Msk

AWSOfficialPublished by AWS

Skill
Workflow
Cloud and DevOps
Apache-2.0
Low risk
Has scripts

From the author

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.

— the description by AWS

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

This skill provides a comprehensive, three-phase framework for migrating self-managed Apache Kafka clusters to Amazon MSK Express. It guides users through inventory discovery, compatibility assessment, and performance simulation to ensure a smooth transition. By analyzing existing infrastructure, the skill identifies potential configuration gaps, calculates target sizing, and provides cost projections, while also offering support for data replication using MSK Replicator.

What it does

Inventories source Kafka clusters from IaC files, CLI output, or manual input.
Assesses compatibility across topology, Kafka versions, configurations, auth, and quotas.
Generates target MSK Express specifications, including broker counts and cost estimates.
Facilitates optional load-testing via temporary MSK Express trial clusters.
Provides guidance on using MSK Replicator for migration cutovers.
Produces standardized artifacts for cluster configuration and compatibility findings.

When to use it

When planning a migration from self-managed Kafka to Amazon MSK Express.
When you need to assess the compatibility of your current Kafka setup with MSK Express.
When you require an accurate sizing and cost projection for an MSK Express target.
When you want to validate performance through a proof-of-concept load test before migrating.
When you need a structured approach to inventorying your existing streaming infrastructure.

Advantages

Reduces migration risk through automated compatibility and topology assessment.
Provides data-driven cost and sizing recommendations based on actual workload metrics.
Supports informed decision-making with optional pre-migration performance simulation.
Standardizes the migration process across discovery, assessment, and execution phases.
Integrates with existing IaC workflows for efficient inventory collection.

Usage examples

“I need to migrate my on-premises Kafka cluster to MSK Express. Where do I start?”: The agent will explain the three-phase migration process (Discovery, Assessment, Simulation) and ask for your infrastructure details or IaC files to begin the discovery phase.
“Here is my Terraform directory for our Kafka cluster. Can you assess if it is compatible with MSK Express?”: The agent will perform the discovery phase to generate a cluster configuration and then proceed to the assessment phase to validate compatibility and provide sizing recommendations.
“I have a cluster-config.json file. How do I get a cost estimate for moving to MSK Express?”: The agent will run the assessment phase using your configuration file to produce a compatibility report and a pricing summary for the recommended MSK Express setup.
“How do I test MSK Express performance with my current workload?”: The agent will guide you through the simulation phase, helping you deploy a temporary MSK Express cluster to run load tests and review performance metrics.

Install

Download the skill as a ZIP:
https://mp-staging.autogineer.com/agent-skills/aws/migrate-to-msk/download

Unzip it into the cross-client skills folder of your project:

mkdir -p .agents/skills && unzip migrate-to-msk.zip -d .agents/skills

Documentation by AWS

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 informational questions, but is not required. The assessment scripts are pure file processors with no AWS API calls.

Intent Routing

Route the customer's request based on their intent:

1. Open/exploratory question ("How do I migrate to MSK?")

Explain what this skill offers:

This skill helps you migrate to MSK Express in three phases:
Phase 1 — Discovery: 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: migrate-to-msk-skill-artifacts/<cluster_name>/cluster-config.json.
Phase 2 — Assessment: 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: compatibility.<cluster_name>.json, the pricing results in Markdown msk_sizing_pricing.md, and msk-sizing-inputs.<cluster_name>.json.
Phase 3 — Simulation: 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.
Data replication: For migrating data to your Express cluster, you can use MSK Replicator. I can provide guidance on setup and configuration.
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 cluster-config.json file, or go straight to simulation if you already know your target Express configuration.

Guardrails for this overview response:

This response is an overview and a routing question only. Do NOT begin, simulate, or pre-empt any phase.
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 cluster-config.json.
Do NOT open, read, or summarize the internals of compatibility.py, simulation_load_test_config.py, 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.
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 references/assessment-compatibility.md for Phase 2, and references/simulation.md for Phase 3.

2. Discovery intent (DEFAULT when IaC files are provided)

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 migrate-to-msk-skill-artifacts/<cluster_name>/cluster-config.json file and stop.

3. Assessment intent

Customer explicitly asks to assess or has a migrate-to-msk-skill-artifacts/<cluster_name>/cluster-config.json file already produced. Run Phase 2 (Assessment) only.

4. Simulation intent

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 Phase 3 — Simulation. An assessment is helpful but not required — the simulation asks for sizing inputs directly.

5. Informational questions

Customer asks about Express capabilities, constraints, configuration differences, authentication support, pricing, or compaction behavior without providing cluster-specific data. Use AWS documentation tools (aws___search_documentation, aws___read_documentation) if available to look up the answer from MSK Express documentation. If MCP tools are not available, reference the MSK Express documentation and answer based on knowledge of AWS MSK.

6. Migration strategy questions

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 (aws___search_documentation, aws___read_documentation) if available to retrieve current requirements and supported configurations. If MCP tools are not available, reference the MSK Replicator documentation and answer based on knowledge of AWS MSK.

Phase 1 — Discovery

Purpose: Inventory the source cluster to build a migration profile.

Input: One of:

A directory path containing IaC files (CDK, CloudFormation, Docker Compose, Kubernetes manifests, Terraform)
Output from Kafka CLI commands the customer runs on their cluster
Manual information provided by the customer in conversation

Output: migrate-to-msk-skill-artifacts/<cluster_name>/cluster-config.json — saved to the working directory.

Discovery rules

Before doing ANYTHING else in discovery, you MUST read references/discovery.md in full. It defines the input methods, the REQUIRED response template, the forbidden content, and the cluster-config.json schema. Do NOT respond until you have read it, and follow its template EXACTLY.
ALWAYS save migrate-to-msk-skill-artifacts/<cluster_name>/cluster-config.json in the working directory.
Do NOT proceed to Phase 2 without explicit customer confirmation.

Phase 2 — Assessment

Purpose: Assess the cluster against MSK Express requirements and produce a target Express specification (instance type, broker count, monthly cost projection).

Input: migrate-to-msk-skill-artifacts/<cluster_name>/cluster-config.json from Phase 1.

Outputs:

migrate-to-msk-skill-artifacts/<cluster_name>/compatibility.<cluster_name>.json — five-pillar verdict.
migrate-to-msk-skill-artifacts/<cluster_name>/msk_sizing_pricing.md — the managing-amazon-msk Skill's pricing report with broker count and cost recommendations.
migrate-to-msk-skill-artifacts/<cluster_name>/msk-sizing-inputs.<cluster_name>.json — a record of the six input values for sizing logic.

Assessment has two independent halves; run them in either order, and a failure in one does not block the other:

Compatibility — scripts/compatibility.py, a pure file processor (no live AWS API calls) run via uv run 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 (INFO, ADVISORY, or ACTION_REQUIRED, worst-of for the overall). Use those three strings verbatim.
Sizing — not a script in this skill. Load the managing-amazon-msk Skill and run its scripts/msk_sizing.py with the workload inputs derived from cluster-config.json, passing --broker-classes express and --region for the target Region. The script prices from the live AWS Pricing API when credentials are available and from an embedded us-east-1 snapshot when they are not.

Assessment rules

Read references/assessment-compatibility.md before responding. It carries the invocation commands, the per-pillar thresholds and evidence codes, the verdict definitions, the full forbidden-behavior list, and the required response template covering both artifacts. Do not freestyle the post-script summary.
Read references/assessment-sizing.md before running sizing. It carries the input derivations (several are not one-to-one — getting them wrong silently produces a wrong broker count), the target-block flags for rack affinity and negotiated discounts, the Region and pricing-source handling, the Express-only presentation rule, and the source-footprint comparison.
Surface any ACTION_REQUIRED evidence to the user for awareness, but do not gate further phases on it. Express may still accept the workload with mitigations.
Do NOT pivot back into discovery. Assessment operates on the existing cluster-config.json as-is. Partial data is fine — the scripts emit ADVISORY evidence (METRICS_MISSING, AZ_COUNT_UNKNOWN, etc.) for missing fields; surface those findings and stop. Do not propose Kafka CLI commands, IaC walks, scripts, or questionnaires to fill the gaps.
Report broker counts and costs only as read verbatim from the sizing script output. Never round, re-derive, or estimate them. Report the pricing source the script used alongside the costs; when it is static, the rates are an embedded snapshot, not confirmed current pricing.

Phase 3 — Simulation (optional)

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 references/simulation.md; the deterministic artifacts it drives are scripts/simulation_load_test_config.py and the static assets/simulation-stack.yaml.

Execution model

Scripts run on the customer's local machine via uv run. 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).

Security Considerations

Apply these controls at every phase. For additional detail, see MSK Security best practices and MSK IAM access control.

Encryption in transit (mandatory). Enforce TLS for client-broker traffic on the MSK Express target (EncryptionInTransit.ClientBroker = TLS).
Encryption at rest (mandatory). Provision the target cluster with a customer-managed KMS key (or AWS-managed if your compliance posture allows).
Authentication — prefer IAM over long-lived credentials. Configure the MSK Express target with IAM authentication as the sole client auth method. This gives ephemeral, role-based credentials with full CloudTrail coverage.
Credential storage — use AWS Secrets Manager. Store SASL/SCRAM and TLS credentials for source cluster access in Secrets Manager. Never pass passwords as CLI arguments.
Network isolation. 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.
CloudTrail logging and CloudWatch alarms. Ensure CloudTrail is enabled in the target account and covers kafka.amazonaws.com API calls. Configure alarms:
- ClientAuthenticationFailure — surge indicates credential problems or attack
- ConnectionCloseCount — abnormal spike may indicate connection-flooding
- CloudTrail metric filters for denied kafka-cluster:* actions
- Connection-rate alarms approaching the 100 conn/sec/broker IAM limit
Sensitive data handling. 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.

Troubleshooting

Per-pillar findings, including the source topology and out-of-range config cases, are explained in references/assessment-compatibility.md.

Files

ATTRIBUTION.mdAdded by AutoGineer908 B
LICENSELicence file added by AutoGineer9.9 KiB
NOTICELicence file added by AutoGineer67 B
SKILL.md14.0 KiB
assets/simulation-stack.yaml47.9 KiB
references/assessment-compatibility.md23.1 KiB
references/assessment-sizing.md14.8 KiB
references/discovery.md14.8 KiB
references/simulation.md38.6 KiB
scripts/compatibility.pyScript46.7 KiB
scripts/simulation_load_test_config.pyScript17.7 KiB

assets/simulation-stack.yaml

AWSTemplateFormatVersion: "2010-09-09"
Description: >
  MSK Express load-test simulation. Deploys a VPC, an MSK Express cluster (IAM auth,
  TLS in transit), an auto-sized EC2 client fleet, SSM documents for the two vended
  tests (End-to-End Latency, Broker Restart Under Load), and a CloudWatch dashboard.
  Cluster sizing and fleet counts come from scripts/simulation_load_test_config.py — do NOT edit
  this template by hand. To resize, tear down and redeploy with new parameters.

Parameters:
  InstanceType:
    Type: String
    Default: express.m7g.4xlarge
    AllowedValues: [express.m7g.large, express.m7g.xlarge, express.m7g.2xlarge, express.m7g.4xlarge, express.m7g.8xlarge, express.m7g.12xlarge, express.m7g.16xlarge]
    Description: Express broker instance type.
  BrokerCount:
    Type: Number
    Default: 3
    Description: Number of brokers (multiple of 3).
  KafkaVersion:
    Type: String
    Default: 3.9.x.kraft
    # Express supports Kafka 3.6 / 3.8 (ZooKeeper) and 3.9 (KRaft, .kraft suffix).
    AllowedValues: [3.6.0, 3.8.x, 3.9.x.kraft]
    Description: >
      Express Kafka version + metadata mode. 3.9.x.kraft = KRaft (3.9 only, KRaft
      mandatory). 3.6.0 / 3.8.x = ZooKeeper. 3.9 ZooKeeper is not supported on Express.
  TopicName:
    Type: String
    Default: load-test
  TopicPartitions:
    Type: Number
    Default: 48
  ClientInstanceType:
    Type: String
    Default: c5.2xlarge
    Description: EC2 instance type for the producer/consumer fleet (from simulation_load_test_config.py).
  ProducerCount:
    Type: Number
    Default: 3
    Description: Producer fleet size (from simulation_load_test_config.py fleet_producer_count).
  ConsumerCount:
    Type: Number
    Default: 3
    Description: Consumer fleet size (from simulation_load_test_config.py fleet_consumer_count).
  LatestAl2023Ami:
    Type: AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>
    Default: /aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64

Resources:
  # ---------------- Networking ----------------
  Vpc:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.0.0.0/16
      EnableDnsSupport: true
      EnableDnsHostnames: true
      Tags: [{Key: Name, Value: !Sub "${AWS::StackName}-vpc"}]
  Igw:
    Type: AWS::EC2::InternetGateway
  IgwAttach:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties: {VpcId: !Ref Vpc, InternetGatewayId: !Ref Igw}
  PublicSubnet:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: 10.0.0.0/24
      AvailabilityZone: !Select [0, !GetAZs ""]
      # No public IP auto-assignment: only the NAT Gateway lives here and it gets an
      # explicit EIP (see Eip/NatGateway below); no other resource is launched in this subnet.
  PrivateSubnet1:
    Type: AWS::EC2::Subnet
    Properties: {VpcId: !Ref Vpc, CidrBlock: 10.0.1.0/24, AvailabilityZone: !Select [0, !GetAZs ""], MapPublicIpOnLaunch: false}
  PrivateSubnet2:
    Type: AWS::EC2::Subnet
    Properties: {VpcId: !Ref Vpc, CidrBlock: 10.0.2.0/24, AvailabilityZone: !Select [1, !GetAZs ""], MapPublicIpOnLaunch: false}
  PrivateSubnet3:
    Type: AWS::EC2::Subnet
    Properties: {VpcId: !Ref Vpc, CidrBlock: 10.0.3.0/24, AvailabilityZone: !Select [2, !GetAZs ""], MapPublicIpOnLaunch: false}
  Eip:
    Type: AWS::EC2::EIP
    Properties: {Domain: vpc}
  NatGateway:
    Type: AWS::EC2::NatGateway
    DependsOn: IgwAttach
    Properties: {AllocationId: !GetAtt Eip.AllocationId, SubnetId: !Ref PublicSubnet}
  PublicRt:
    Type: AWS::EC2::RouteTable
    Properties: {VpcId: !Ref Vpc}
  # Risk acceptance: a 0.0.0.0/0 route to the IGW in the public subnet is the standard,
  # required AWS pattern that lets the NAT Gateway below reach the internet on behalf of the
  # private subnets. No workload is deployed in PublicSubnet; restricting this route would
  # break NAT egress for the whole stack. Accepted as-is.
  PublicRoute:
    Type: AWS::EC2::Route
    DependsOn: IgwAttach
    Properties: {RouteTableId: !Ref PublicRt, DestinationCidrBlock: 0.0.0.0/0, GatewayId: !Ref Igw}
  PublicAssoc:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties: {RouteTableId: !Ref PublicRt, SubnetId: !Ref PublicSubnet}
  PrivateRt:
    Type: AWS::EC2::RouteTable
    Properties: {VpcId: !Ref Vpc}
  PrivateRoute:
    Type: AWS::EC2::Route
    Properties: {RouteTableId: !Ref PrivateRt, DestinationCidrBlock: 0.0.0.0/0, NatGatewayId: !Ref NatGateway}
  PrivateAssoc1:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties: {RouteTableId: !Ref PrivateRt, SubnetId: !Ref PrivateSubnet1}
  PrivateAssoc2:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties: {RouteTableId: !Ref PrivateRt, SubnetId: !Ref PrivateSubnet2}
  PrivateAssoc3:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties: {RouteTableId: !Ref PrivateRt, SubnetId: !Ref PrivateSubnet3}

  # ---------------- Security groups ----------------
  ClientSg:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: MSK client fleet
      VpcId: !Ref Vpc
      SecurityGroupEgress:
        - {IpProtocol: tcp, FromPort: 9098, ToPort: 9098, DestinationSecurityGroupId: !Ref MskSg, Description: MSK IAM auth}
        # Risk acceptance: egress to 0.0.0.0/0 on 443 is required because SSM, CloudWatch,
        # STS, and the Kafka control-plane API do not publish fixed IP ranges reachable via
        # prefix lists in all Regions. Scoping this to VPC interface endpoints per service is
        # the full remediation but is out of scope for this ephemeral load-test stack; accepted
        # as-is for this template.
        - {IpProtocol: tcp, FromPort: 443, ToPort: 443, CidrIp: 0.0.0.0/0, Description: "HTTPS to AWS APIs (SSM, CloudWatch, STS, Kafka)"}
  MskSg:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: MSK Express brokers
      VpcId: !Ref Vpc
  MskIngress9098:
    Type: AWS::EC2::SecurityGroupIngress
    Properties:
      GroupId: !Ref MskSg
      IpProtocol: tcp
      FromPort: 9098
      ToPort: 9098
      SourceSecurityGroupId: !Ref ClientSg
      Description: IAM auth bootstrap from client fleet

  # ---------------- MSK Express ----------------
  LogGroupKey:
    Type: AWS::KMS::Key
    Properties:
      Description: Encrypts MSK simulation CloudWatch log group
      EnableKeyRotation: true
      KeyPolicy:
        Version: "2012-10-17"
        Statement:
          - Sid: AllowAccountRoot
            Effect: Allow
            Principal: {AWS: !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:root"}
            # Full KMS admin access for account root is the standard AWS pattern for CMK key
            # policies. Production deployments should evaluate narrowing this to specific
            # kms:Create*/Describe*/Enable*/List*/Put*/Update*/Revoke*/Disable*/Delete*/
            # ScheduleKeyDeletion/CancelKeyDeletion actions for the intended admin scope.
            Action: kms:*
            Resource: "*"
          - Sid: AllowCloudWatchLogs
            Effect: Allow
            Principal: {Service: !Sub "logs.${AWS::Region}.amazonaws.com"}
            Action: [kms:Encrypt, kms:Decrypt, kms:GenerateDataKey*, kms:DescribeKey]
            Resource: "*"
            Condition:
              ArnLike:
                kms:EncryptionContext:aws:logs:arn: !Sub "arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:*"
  MskLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      RetentionInDays: 14
      KmsKeyId: !GetAtt LogGroupKey.Arn
  FlowLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      RetentionInDays: 14
      KmsKeyId: !GetAtt LogGroupKey.Arn
  FlowLogRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal: {Service: vpc-flow-logs.amazonaws.com}
            Action: sts:AssumeRole
            Condition:
              StringEquals: {"aws:SourceAccount": !Ref "AWS::AccountId"}
              ArnLike:
                "aws:SourceArn": !Sub "arn:${AWS::Partition}:ec2:${AWS::Region}:${AWS::AccountId}:vpc-flow-log/*"
  FlowLogDeliveryPolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      Roles: [!Ref FlowLogRole]
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action: [logs:CreateLogStream, logs:PutLogEvents, logs:DescribeLogStreams]
            # Log-stream actions require the log-group ARN with a :* suffix.
            Resource: !Sub "${FlowLogGroup.Arn}:*"
  VpcFlowLog:
    Type: AWS::EC2::FlowLog
    DependsOn: FlowLogDeliveryPolicy
    Properties:
      ResourceId: !Ref Vpc
      ResourceType: VPC
      TrafficType: ALL
      LogDestinationType: cloud-watch-logs
      LogGroupName: !Ref FlowLogGroup
      DeliverLogsPermissionArn: !GetAtt FlowLogRole.Arn
  MskCluster:
    Type: AWS::MSK::Cluster
    Properties:
      ClusterName: !Sub "${AWS::StackName}-express"
      KafkaVersion: !Ref KafkaVersion
      NumberOfBrokerNodes: !Ref BrokerCount
      EnhancedMonitoring: PER_BROKER
      BrokerNodeGroupInfo:
        InstanceType: !Ref InstanceType
        ClientSubnets: [!Ref PrivateSubnet1, !Ref PrivateSubnet2, !Ref PrivateSubnet3]
        SecurityGroups: [!Ref MskSg]
      ClientAuthentication:
        Sasl: {Iam: {Enabled: true}}
        Unauthenticated: {Enabled: false}
      EncryptionInfo:
        # Encryption at rest uses the default AWS-managed KMS key.
        EncryptionInTransit: {ClientBroker: TLS, InCluster: true}
      LoggingInfo:
        BrokerLogs:
          CloudWatchLogs: {Enabled: true, LogGroup: !Ref MskLogGroup}
      Tags:
        # Tag used to track MSK Express migration simulation usage; value is the skill version.
        MSKExpressMigrationSkill: "1"

  # ---------------- IAM ----------------
  InstanceRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal: {Service: ec2.amazonaws.com}
            Action: sts:AssumeRole
            Condition:
              StringEquals:
                aws:SourceAccount: !Ref AWS::AccountId
      ManagedPolicyArns: [!Sub "arn:${AWS::Partition}:iam::aws:policy/AmazonSSMManagedInstanceCore"]
  InstanceKafkaPolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      Roles: [!Ref InstanceRole]
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action: [kafka-cluster:Connect, kafka-cluster:DescribeCluster]
            Resource: !Ref MskCluster
          - Effect: Allow
            Action: [kafka-cluster:CreateTopic, kafka-cluster:DescribeTopic, kafka-cluster:WriteData, kafka-cluster:ReadData]
            Resource: !Sub "arn:${AWS::Partition}:kafka:${AWS::Region}:${AWS::AccountId}:topic/${AWS::StackName}-express/*"
          - Effect: Allow
            Action: [kafka-cluster:AlterGroup, kafka-cluster:DescribeGroup]
            Resource: !Sub "arn:${AWS::Partition}:kafka:${AWS::Region}:${AWS::AccountId}:group/${AWS::StackName}-express/*"
          - Effect: Allow
            Action: [kafka:GetBootstrapBrokers, kafka:DescribeCluster, kafka:DescribeClusterV2, kafka:ListNodes]
            Resource: !Ref MskCluster
          # Broker Restart Under Load test — scoped to this cluster only.
          - Effect: Allow
            Action: kafka:RebootBroker
            Resource: !Ref MskCluster
          # PutMetricData has no resource-level scoping; restrict by namespace.
          - Effect: Allow
            Action: cloudwatch:PutMetricData
            Resource: "*"
            Condition: {StringEquals: {"cloudwatch:namespace": MSKSimulation}}
  InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    DependsOn: InstanceKafkaPolicy
    Properties: {Roles: [!Ref InstanceRole]}

  # ---------------- Client fleet ----------------
  LaunchTemplate:
    Type: AWS::EC2::LaunchTemplate
    Properties:
      LaunchTemplateName: !Sub "${AWS::StackName}-client"
      LaunchTemplateData:
        ImageId: !Ref LatestAl2023Ami
        InstanceType: !Ref ClientInstanceType
        IamInstanceProfile: {Arn: !GetAtt InstanceProfile.Arn}
        SecurityGroupIds: [!Ref ClientSg]
        MetadataOptions: {HttpTokens: required, InstanceMetadataTags: enabled}
        Monitoring: {Enabled: true}
        UserData:
          Fn::Base64: !Sub |
            #!/bin/bash
            set -xe
            exec > /var/log/simulation-bootstrap.log 2>&1
            # Signal the matching Auto Scaling group based on the instance's simulation:role tag.
            TOK=$(curl -sX PUT http://169.254.169.254/latest/api/token -H "X-aws-ec2-metadata-token-ttl-seconds: 300")
            ROLE=$(curl -s -H "X-aws-ec2-metadata-token: $TOK" http://169.254.169.254/latest/meta-data/tags/instance/simulation:role)
            if [ "$ROLE" = "producer" ]; then RES=ProducerAsg; else RES=ConsumerAsg; fi
            # On failure, signal CloudFormation so the stack rolls back instead of completing half-built.
            trap 'cfn-signal -e 1 --stack ${AWS::StackName} --resource $RES --region ${AWS::Region}' ERR
            dnf install -y java-17-amazon-corretto-headless tar gzip aws-cfn-bootstrap
            cd /opt
            # Prepare the Kafka install location and client authentication config.
            # The Kafka client is installed after deployment via the InstallKafkaClient document.
            mkdir -p /opt/kafka/bin /opt/kafka/libs
            cat > /opt/kafka/client.properties <<'EOF'
            security.protocol=SASL_SSL
            sasl.mechanism=AWS_MSK_IAM
            sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
            sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
            EOF
            # Bootstrap succeeded; signal CloudFormation for this instance.
            cfn-signal -e 0 --stack ${AWS::StackName} --resource $RES --region ${AWS::Region}
  ProducerAsg:
    Type: AWS::AutoScaling::AutoScalingGroup
    # Wait for all producers to finish bootstrapping before the stack completes.
    CreationPolicy:
      ResourceSignal:
        Count: !Ref ProducerCount
        Timeout: PT15M
    Properties:
      VPCZoneIdentifier: [!Ref PrivateSubnet1, !Ref PrivateSubnet2, !Ref PrivateSubnet3]
      LaunchTemplate: {LaunchTemplateId: !Ref LaunchTemplate, Version: !GetAtt LaunchTemplate.LatestVersionNumber}
      MinSize: !Ref ProducerCount
      MaxSize: !Ref ProducerCount
      DesiredCapacity: !Ref ProducerCount
      AutoScalingGroupName: !Sub "${AWS::StackName}-producer"
      Tags:
        - {Key: "simulation:role", Value: producer, PropagateAtLaunch: true}
        - {Key: Name, Value: !Sub "${AWS::StackName}-producer", PropagateAtLaunch: true}
  ConsumerAsg:
    Type: AWS::AutoScaling::AutoScalingGroup
    CreationPolicy:
      ResourceSignal:
        Count: !Ref ConsumerCount
        Timeout: PT15M
    Properties:
      VPCZoneIdentifier: [!Ref PrivateSubnet1, !Ref PrivateSubnet2, !Ref PrivateSubnet3]
      LaunchTemplate: {LaunchTemplateId: !Ref LaunchTemplate, Version: !GetAtt LaunchTemplate.LatestVersionNumber}
      MinSize: !Ref ConsumerCount
      MaxSize: !Ref ConsumerCount
      DesiredCapacity: !Ref ConsumerCount
      AutoScalingGroupName: !Sub "${AWS::StackName}-consumer"
      Tags:
        - {Key: "simulation:role", Value: consumer, PropagateAtLaunch: true}
        - {Key: Name, Value: !Sub "${AWS::StackName}-consumer", PropagateAtLaunch: true}

  # ---------------- SSM documents (self-contained) ----------------
  CreateTopicsDoc:
    Type: AWS::SSM::Document
    Properties:
      DocumentType: Command
      Name: !Sub "${AWS::StackName}-CreateTopics"
      Content:
        schemaVersion: "2.2"
        description: >
          Create the per-test load topics (run once, after cluster is ACTIVE): one
          for the E2E Latency test, one for the Broker Restart test, plus a dedicated
          single-partition probe topic for the e2e-latency sampler. Per-test topics
          keep each test's traffic isolated, so an idle test's consumer group never
          accumulates phantom lag from the other test writing to a shared topic.
        parameters:
          topicName: {type: String, default: !Ref TopicName}
          partitions: {type: String, default: !Sub "${TopicPartitions}"}
        mainSteps:
          - action: aws:runShellScript
            name: createTopic
            inputs:
              runCommand:
                - 'if ! /opt/kafka/bin/kafka-topics.sh --version >/dev/null 2>&1; then echo "ERROR: Kafka client not installed/working on this instance. Install it first via the InstallKafkaClient SSM document, then retry." >&2; exit 1; fi'
                - !Sub 'BS=$(aws kafka get-bootstrap-brokers --cluster-arn ${MskCluster} --region ${AWS::Region} --query BootstrapBrokerStringSaslIam --output text)'
                # E2E Latency test load topic (consumer group simulation-e2e).
                - '/opt/kafka/bin/kafka-topics.sh --bootstrap-server $BS --command-config /opt/kafka/client.properties --create --if-not-exists --topic {{topicName}}-e2e --partitions {{partitions}} --replication-factor 3'
                # Dedicated single-partition probe topic for the e2e-latency sampler (kept off the bulk-load topic).
                - '/opt/kafka/bin/kafka-topics.sh --bootstrap-server $BS --command-config /opt/kafka/client.properties --create --if-not-exists --topic {{topicName}}-e2e-probe --partitions 1 --replication-factor 3'
                # Broker Restart test load topic (consumer group simulation-restart).
                - '/opt/kafka/bin/kafka-topics.sh --bootstrap-server $BS --command-config /opt/kafka/client.properties --create --if-not-exists --topic {{topicName}}-restart --partitions {{partitions}} --replication-factor 3'

  RunE2ELatencyDoc:
    Type: AWS::SSM::Document
    Properties:
      DocumentType: Command
      Name: !Sub "${AWS::StackName}-RunE2ELatency"
      Content:
        schemaVersion: "2.2"
        description: >
          End-to-end (produce->consume) latency under load. Producers drive the
          requested per-instance throughput; one producer samples round-trip
          latency and emits it as the MSKSimulation/E2ELatencyMs custom metric.
          Consumers drain the topic. Branches on the simulation:role tag.
        parameters:
          perInstanceThroughputMbps: {type: String, default: "10"}
          durationMinutes: {type: String, default: "10"}
          recordSizeBytes: {type: String, default: "1024"}
          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."}
          leadInstanceId: {type: String, default: "", description: "Instance id that samples e2e latency (the first targeted producer)."}
          topic: {type: String, default: !Sub "${TopicName}-e2e"}
        mainSteps:
          - action: aws:runShellScript
            name: run
            inputs:
              timeoutSeconds: "8400"
              runCommand:
                - !Sub |
                    set -e
                    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
                    REGION=${AWS::Region}
                    CL=${MskCluster}
                    BS=$(aws kafka get-bootstrap-brokers --cluster-arn $CL --region $REGION --query BootstrapBrokerStringSaslIam --output text)
                    TOK=$(curl -sX PUT http://169.254.169.254/latest/api/token -H "X-aws-ec2-metadata-token-ttl-seconds: 300")
                    IMDS="curl -s -H X-aws-ec2-metadata-token:$TOK http://169.254.169.254/latest/meta-data"
                    IID=$($IMDS/instance-id)
                    ROLE=$($IMDS/tags/instance/simulation:role)
                    DUR={{durationMinutes}}; SECS=$((DUR*60))
                    CAP={{maxDrainMinutes}}; CAPS=$((CAP*60))   # hard cap on the adaptive drain (see consumer branch)
                    TOPIC={{topic}}; RS={{recordSizeBytes}}
                    # Dedicated probe topic so the latency sampler reads back only its own messages.
                    PROBE_TOPIC=${!TOPIC}-probe
                    P=/opt/kafka/client.properties; K=/opt/kafka/bin
                - |
                    if [ "$ROLE" = "consumer" ]; then
                      # Background consumer; drains until caught up (or maxDrainMinutes) so consume-side metrics reflect the test.
                      GRP=simulation-e2e
                      # Reset the consumer group to the topic's latest so each run starts clean.
                      $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
                      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 &
                      CPID=$!
                      PRODEND=$(( $(date +%s) + SECS ))     # producers stop ~here
                      HARDEND=$(( PRODEND + CAPS ))          # absolute cap on the drain
                      while kill -0 $CPID 2>/dev/null && [ $(date +%s) -lt $HARDEND ]; do
                        sleep 15
                        if [ $(date +%s) -lt $PRODEND ]; then continue; fi   # don't test "caught up" until producers stop
                        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
                        echo "adaptive-drain: total_lag=$LAG"
                        if [ "$LAG" != "NA" ] && [ "$LAG" -le 1000 ]; then echo "adaptive-drain: caught up (lag<=1000), stopping"; break; fi
                      done
                      kill $CPID 2>/dev/null || true
                      exit 0
                    fi
                    # Producers generate the aggregate target load.
                    RECS=$(awk -v tput={{perInstanceThroughputMbps}} -v rs=$RS 'BEGIN{printf "%d", tput*1024*1024/rs}')
                    TOTAL=$(( RECS * SECS ))
                    nohup $K/kafka-producer-perf-test.sh --topic $TOPIC --num-records $TOTAL --record-size $RS --throughput $RECS \
                      --producer-props bootstrap.servers=$BS --producer.config $P >/tmp/perf.out 2>&1 &
                    # The lead producer samples end-to-end latency and publishes it as a custom metric.
                    if [ "$IID" = "{{leadInstanceId}}" ]; then
                      END=$(( $(date +%s) + SECS ))
                      while [ $(date +%s) -lt $END ]; do
                        MS=$($K/kafka-e2e-latency.sh $BS $PROBE_TOPIC 1000 1 $RS $P 2>/dev/null | grep -i 'Avg latency' | grep -oE '[0-9]+\.?[0-9]*' | head -1)
                        if echo "$MS" | grep -qE '^[0-9]+\.?[0-9]*$'; then
                          aws cloudwatch put-metric-data --region $REGION --namespace MSKSimulation \
                            --metric-name E2ELatencyMs --unit Milliseconds --value "$MS" --dimensions Test=E2ELatency,Cluster=$CL
                        fi
                        sleep 30
                      done
                    fi
                    wait

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

  InstallKafkaClientDoc:
    Type: AWS::SSM::Document
    Properties:
      DocumentType: Command
      Name: !Sub "${AWS::StackName}-InstallKafkaClient"
      Content:
        schemaVersion: "2.2"
        description: >
          Install the Kafka client and the aws-msk-iam-auth jar on the fleet. The
          Kafka client tarball URL is provided as a parameter at invocation
          (kafkaUrl); the aws-msk-iam-auth jar is installed from its AWS-owned source.
        parameters:
          kafkaUrl:
            type: String
            description: "URL to the Kafka client tarball to install."
          kafkaSha512:
            type: String
            description: "SHA-512 checksum of the Kafka client tarball; verified before install."
        mainSteps:
          - action: aws:runShellScript
            name: install
            inputs:
              runCommand:
                - |
                    set -xe
                    mkdir -p /opt/kafka/bin /opt/kafka/libs
                    cd /opt
                    # Require an HTTPS URL for the customer-provided client tarball before downloading.
                    KURL="{{kafkaUrl}}"
                    case "$KURL" in
                      https://*) : ;;
                      *) echo "ERROR: kafkaUrl must be an HTTPS URL (got: $KURL)." >&2; exit 1 ;;
                    esac
                    # Install the Kafka client from the provided URL.
                    curl -fsSL "$KURL" -o kafka.tgz
                    # Validate the checksum is a 128-char lowercase hex string before use
                    # (it is interpolated into a shell command, so reject metacharacters).
                    SHA="{{kafkaSha512}}"
                    if ! echo "$SHA" | grep -qxE '[0-9a-f]{128}'; then
                      echo "ERROR: kafkaSha512 must be a 128-char lowercase hex string." >&2; exit 1
                    fi
                    # Verify the tarball against the customer-provided SHA-512 before using it.
                    echo "$SHA  kafka.tgz" | sha512sum -c -
                    tar xzf kafka.tgz --strip-components=1 -C /opt/kafka
                    # Install the aws-msk-iam-auth jar from its AWS-owned source (required for MSK IAM auth).
                    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
                    # Verify the pinned jar against its known SHA-256.
                    echo "735dd058b81d088e767770d6cc4e7f1a1beea815ced18b66a23b35fd87651e77  /opt/kafka/libs/aws-msk-iam-auth.jar" | sha256sum -c -
                    cat > /opt/kafka/client.properties <<'EOF'
                    security.protocol=SASL_SSL
                    sasl.mechanism=AWS_MSK_IAM
                    sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
                    sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
                    EOF
                    /opt/kafka/bin/kafka-topics.sh --version

  ValidateClientDoc:
    Type: AWS::SSM::Document
    Properties:
      DocumentType: Command
      Name: !Sub "${AWS::StackName}-ValidateClient"
      Content:
        schemaVersion: "2.2"
        description: >
          Validate the installed Kafka client against the cluster: confirms the client
          runs, its version is >= the cluster's Kafka version, and it can authenticate
          (IAM over TLS) and list topics. Fails with a clear message if the client is
          missing or incompatible.
        mainSteps:
          - action: aws:runShellScript
            name: validate
            inputs:
              runCommand:
                - !Sub |
                    set -e
                    if ! /opt/kafka/bin/kafka-topics.sh --version >/dev/null 2>&1; then
                      echo "ERROR: Kafka client not installed/working. Run ${AWS::StackName}-InstallKafkaClient first." >&2; exit 1
                    fi
                    CLIENT_VER=$(/opt/kafka/bin/kafka-topics.sh --version 2>/dev/null | awk '{print $1}')
                    CLUSTER_VER=$(echo "${KafkaVersion}" | sed 's/\.kraft$//; s/\.x/.0/')
                    LOWER=$(printf '%s\n%s\n' "$CLIENT_VER" "$CLUSTER_VER" | sort -V | head -n1)
                    if [ "$CLIENT_VER" != "$CLUSTER_VER" ] && [ "$LOWER" = "$CLIENT_VER" ]; then
                      echo "ERROR: Kafka client version $CLIENT_VER is older than the cluster's $CLUSTER_VER; install a client >= the cluster version." >&2; exit 1
                    fi
                    BS=$(aws kafka get-bootstrap-brokers --cluster-arn ${MskCluster} --region ${AWS::Region} --query BootstrapBrokerStringSaslIam --output text)
                    if ! /opt/kafka/bin/kafka-topics.sh --bootstrap-server "$BS" --command-config /opt/kafka/client.properties --list >/dev/null 2>&1; then
                      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
                    fi
                    echo "Kafka client OK: version $CLIENT_VER >= cluster $CLUSTER_VER, authenticated and listed topics on the cluster."

  # ---------------- Dashboard ----------------
  Dashboard:
    Type: AWS::CloudWatch::Dashboard
    Properties:
      DashboardName: !Sub "${AWS::StackName}-loadtest"
      DashboardBody: !Sub |
        {
          "widgets": [
            {
              "type": "text",
              "x": 0,
              "y": 0,
              "width": 24,
              "height": 5,
              "properties": {
                "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."
              }
            },
            {
              "type": "metric",
              "x": 0,
              "y": 5,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "Fleet CPU % \u2014 client fleet (maxed = fleet is the bottleneck)",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "AVG(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\"${AWS::StackName}-producer\" MetricName=\"CPUUtilization\"', 'Average', 60))",
                      "label": "producers",
                      "id": "pcpu"
                    }
                  ],
                  [
                    {
                      "expression": "AVG(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\"${AWS::StackName}-consumer\" MetricName=\"CPUUtilization\"', 'Average', 60))",
                      "label": "consumers",
                      "id": "ccpu"
                    }
                  ]
                ]
              }
            },
            {
              "type": "metric",
              "x": 12,
              "y": 5,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "Fleet network bytes/min \u2014 producer egress / consumer ingress",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "SUM(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\"${AWS::StackName}-producer\" MetricName=\"NetworkOut\"', 'Average', 60))",
                      "label": "producers NetworkOut",
                      "id": "pno"
                    }
                  ],
                  [
                    {
                      "expression": "SUM(SEARCH('{AWS/EC2,AutoScalingGroupName} AutoScalingGroupName=\"${AWS::StackName}-consumer\" MetricName=\"NetworkIn\"', 'Average', 60))",
                      "label": "consumers NetworkIn",
                      "id": "cni"
                    }
                  ]
                ]
              }
            },
            {
              "type": "metric",
              "x": 0,
              "y": 11,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "End-to-end latency (ms) \u2014 produce->consume round trip (probe)",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "MAX(SEARCH('{MSKSimulation,Cluster,Test} Test=\"E2ELatency\" MetricName=\"E2ELatencyMs\"', 'Average', 60))",
                      "label": "E2ELatencyMs",
                      "id": "e2elat"
                    }
                  ]
                ]
              }
            },
            {
              "type": "metric",
              "x": 12,
              "y": 11,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "Produce latency (ms) during broker restart \u2014 watch spike & recovery",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "MAX(SEARCH('{MSKSimulation,Cluster,Test} Test=\"BrokerRestart\" MetricName=\"ProduceLatencyMs\"', 'Average', 60))",
                      "label": "ProduceLatencyMs",
                      "id": "prodlat"
                    }
                  ]
                ]
              }
            },
            {
              "type": "metric",
              "x": 0,
              "y": 17,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "Consumer lag \u2014 how far consumers trail (messages behind, per test)",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "MAX(SEARCH('{AWS/Kafka,\"Cluster Name\",\"Consumer Group\",\"Topic\"} \"Cluster Name\"=\"${AWS::StackName}-express\" \"Consumer Group\"=\"simulation-e2e\" MetricName=\"MaxOffsetLag\"', 'Average', 60))",
                      "label": "E2E lag - simulation-e2e (msgs behind)",
                      "id": "lage2e"
                    }
                  ],
                  [
                    {
                      "expression": "MAX(SEARCH('{AWS/Kafka,\"Cluster Name\",\"Consumer Group\",\"Topic\"} \"Cluster Name\"=\"${AWS::StackName}-express\" \"Consumer Group\"=\"simulation-restart\" MetricName=\"MaxOffsetLag\"', 'Average', 60))",
                      "label": "Broker Restart lag - simulation-restart (msgs behind)",
                      "id": "lagrestart"
                    }
                  ]
                ]
              }
            },
            {
              "type": "text",
              "x": 0,
              "y": 23,
              "width": 24,
              "height": 5,
              "properties": {
                "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."
              }
            },
            {
              "type": "metric",
              "x": 0,
              "y": 28,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "Bytes In/Out per sec \u2014 total write/read traffic (all brokers)",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "SUM(SEARCH('{AWS/Kafka,\"Broker ID\",\"Cluster Name\"} \"Cluster Name\"=\"${AWS::StackName}-express\" MetricName=\"BytesInPerSec\"', 'Average', 60))",
                      "label": "BytesInPerSec (cluster)",
                      "id": "bin"
                    }
                  ],
                  [
                    {
                      "expression": "SUM(SEARCH('{AWS/Kafka,\"Broker ID\",\"Cluster Name\"} \"Cluster Name\"=\"${AWS::StackName}-express\" MetricName=\"BytesOutPerSec\"', 'Average', 60))",
                      "label": "BytesOutPerSec (cluster)",
                      "id": "bout"
                    }
                  ]
                ]
              }
            },
            {
              "type": "metric",
              "x": 12,
              "y": 28,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "CPU user % \u2014 broker CPU on Kafka (keep < 60%)",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "AVG(SEARCH('{AWS/Kafka,\"Broker ID\",\"Cluster Name\"} \"Cluster Name\"=\"${AWS::StackName}-express\" MetricName=\"CpuUser\"', 'Average', 60))",
                      "label": "CpuUser (avg %)",
                      "id": "cpu"
                    }
                  ]
                ]
              }
            },
            {
              "type": "metric",
              "x": 0,
              "y": 34,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "Health \u2014 controller=1; under-replicated & offline=0",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    "AWS/Kafka",
                    "ActiveControllerCount",
                    "Cluster Name",
                    "${AWS::StackName}-express"
                  ],
                  [
                    "AWS/Kafka",
                    "UnderReplicatedPartitions",
                    "Cluster Name",
                    "${AWS::StackName}-express"
                  ],
                  [
                    "AWS/Kafka",
                    "OfflinePartitionsCount",
                    "Cluster Name",
                    "${AWS::StackName}-express"
                  ],
                  [
                    {
                      "expression": "SUM(SEARCH('{AWS/Kafka,\"Broker ID\",\"Cluster Name\"} \"Cluster Name\"=\"${AWS::StackName}-express\" MetricName=\"UnderMinIsrPartitionCount\"', 'Average', 60))",
                      "label": "UnderMinIsrPartitionCount (sum)",
                      "id": "uminisr"
                    }
                  ]
                ]
              }
            },
            {
              "type": "metric",
              "x": 12,
              "y": 34,
              "width": 12,
              "height": 6,
              "properties": {
                "title": "Replication bytes/sec \u2014 re-replication (spikes as a restarted broker catches up)",
                "region": "${AWS::Region}",
                "stat": "Average",
                "period": 60,
                "metrics": [
                  [
                    {
                      "expression": "SUM(SEARCH('{AWS/Kafka,\"Broker ID\",\"Cluster Name\"} \"Cluster Name\"=\"${AWS::StackName}-express\" MetricName=\"ReplicationBytesOutPerSec\"', 'Average', 60))",
                      "label": "ReplicationBytesOutPerSec (cluster)",
                      "id": "repout"
                    }
                  ]
                ]
              }
            }
          ]
        }
Outputs:
  ClusterArn:
    Value: !Ref MskCluster
  DashboardUrl:
    Value: !Sub "https://${AWS::Region}.console.aws.amazon.com/cloudwatch/home?region=${AWS::Region}#dashboards:name=${AWS::StackName}-loadtest"
  DashboardArn:
    Value: !Sub "arn:${AWS::Partition}:cloudwatch::${AWS::AccountId}:dashboard/${AWS::StackName}-loadtest"
  CreateTopicsCommand:
    Value: !Sub "aws ssm send-command --document-name ${AWS::StackName}-CreateTopics --targets Key=tag:simulation:role,Values=producer --max-concurrency 1 --region ${AWS::Region}"
  InstallKafkaClientCommand:
    Description: >
      Installs the Kafka client (from the kafkaUrl you pass) and the aws-msk-iam-auth
      jar (from its AWS-owned source) on the fleet.
    Value: !Sub "aws ssm send-command --document-name ${AWS::StackName}-InstallKafkaClient --targets Key=tag:simulation:role,Values=producer,consumer --parameters 'kafkaUrl=<kafka-client-url>,kafkaSha512=<sha512>' --region ${AWS::Region}"
  ValidateClientCommand:
    Description: >
      Validates the installed Kafka client against this cluster (version >= cluster,
      and live IAM/TLS auth + topic list). Run after install and before tests; re-run
      after any client fix.
    Value: !Sub "aws ssm send-command --document-name ${AWS::StackName}-ValidateClient --targets Key=tag:simulation:role,Values=producer,consumer --region ${AWS::Region}"
  FleetSummary:
    Value: !Sub "${ProducerCount} x ${ClientInstanceType} producers, ${ConsumerCount} consumers; cluster ${BrokerCount} x ${InstanceType}"

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs an MCP server, Needs Python

Security

What our scanner noticed· Low risk

Instructions that use the network
Scripts that your agent may run

This skill includes scripts that your agent may run. Read them before you use it.

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Published by

AWS

Publisher

AWS (Official)

Source

aws/agent-toolkit-for-aws · plugins/aws-data-analytics/skills/migrate-to-msk · commit 0a218f8

Licence

Apache License 2.0 (redistribution allowed)

LICENSE (added by AutoGineer from LICENSE)
NOTICE (added by AutoGineer from NOTICE)
ATTRIBUTION.md (written by AutoGineer)

Synced

2026-11-08

Also distributed in

aws/agent-toolkit-for-aws

AutoGineer is not affiliated with or endorsed by AWS.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download migrate-to-msk.zip