AWS MSK Setup

app.cfn.yml

  AWSTemplateFormatVersion: 2010-09-09
  Parameters:
    KeyName:
      Description: Name of an existing EC2 KeyPair to enable SSH access to the instance
      Type: 'AWS::EC2::KeyPair::KeyName'
      ConstraintDescription: Can contain only ASCII characters.
    VPC:
      Description: ID of the VPC for MSK cluster deployment  
      Type: 'AWS::EC2::VPC::Id'
    PublicSubnetOne:
      Description: Public subnet of the given VPC for MSK cluster deployment  
      Type: 'AWS::EC2::Subnet::Id'
    KafkaClientInstanceSecurityGroup:
      Description: ID of the VPC for MSK cluster deployment  
      Type: 'AWS::EC2::SecurityGroup::Id'
    AMI:
      Description: AMI ID for the EC2 instance  
      Type: String # Try ami-08f3064d8481f3782 https://cloud-images.ubuntu.com/locator/ec2/
  Resources:
    
    KafkaClientEC2Instance:
      Type: 'AWS::EC2::Instance'
      Properties:
        InstanceType: t3.small
        KeyName: !Ref KeyName
        IamInstanceProfile: !Ref EC2InstanceProfile
        AvailabilityZone: !Select 
          - 0
          - !GetAZs 
            Ref: 'AWS::Region'
        SubnetId: !Ref PublicSubnetOne
        SecurityGroupIds:
          - !Ref KafkaClientInstanceSecurityGroup
        ImageId: !Ref AMI
        Tags:
          - Key: Name
            Value: KafkaClientInstance
        UserData: !Base64 >
          #!/bin/bash

          yum update -y 

          yum install python3.7 -y

          yum install java-1.8.0-openjdk-devel -y

          yum erase awscli -y

          cd /home/ec2-user

          echo "export PATH=.local/bin:$PATH" >> .bash_profile

          mkdir kafka

          mkdir mm

          cd kafka

          wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz

          tar -xzf kafka_2.12-2.2.1.tgz

          cd /home/ec2-user

          wget https://bootstrap.pypa.io/get-pip.py

          su -c "python3.7 get-pip.py --user" -s /bin/sh ec2-user

          su -c "/home/ec2-user/.local/bin/pip3 install boto3 --user" -s /bin/sh
          ec2-user

          su -c "/home/ec2-user/.local/bin/pip3 install awscli --user" -s /bin/sh
          ec2-user

          chown -R ec2-user ./kafka

          chgrp -R ec2-user ./kafka

          chown -R ec2-user ./mm

          chgrp -R ec2-user ./mm
    EC2Role:
      Type: 'AWS::IAM::Role'
      Properties:
        AssumeRolePolicyDocument:
          Version: 2012-10-17
          Statement:
            - Sid: ''
              Effect: Allow
              Principal:
                Service: ec2.amazonaws.com
              Action: 'sts:AssumeRole'
        Path: /
        ManagedPolicyArns:
          - 'arn:aws:iam::aws:policy/AmazonMSKFullAccess'
          - 'arn:aws:iam::aws:policy/AWSCloudFormationReadOnlyAccess'
    EC2InstanceProfile:
      Type: 'AWS::IAM::InstanceProfile'
      Properties:
        InstanceProfileName: EC2MSKCFProfile
        Roles:
          - !Ref EC2Role
  Outputs:
    KafkaClientEC2InstancePublicDNS:
      Description: The Public DNS for the MirrorMaker EC2 instance
      Value: !GetAtt 
        - KafkaClientEC2Instance
        - PublicDnsName

mks.cfn.yml

AWSTemplateFormatVersion: 2010-09-09
Parameters:
  VPC:
    Description: ID of the VPC for MSK cluster deployment  
    Type: 'AWS::EC2::VPC::Id'
  PrivateSubnetOne:
    Description: Private subnet of the given VPC for MSK cluster deployment
    Type: 'AWS::EC2::Subnet::Id'
  PrivateSubnetTwo:
    Description: Private subnet of the given VPC for MSK cluster deployment
    Type: 'AWS::EC2::Subnet::Id'
Resources:
  KafkaClientInstanceSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: Enable SSH, HTTP, HTTPs access
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 22
          ToPort: 22
        - IpProtocol: tcp
          FromPort: 433
          ToPort: 433
        - IpProtocol: tcp
          FromPort: 433
          ToPort: 433
  MSKSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: Enable SSH access via port 22
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 2181
          ToPort: 2181
          SourceSecurityGroupId: !GetAtt 
            - KafkaClientInstanceSecurityGroup
            - GroupId
        - IpProtocol: tcp
          FromPort: 9094
          ToPort: 9094
          SourceSecurityGroupId: !GetAtt 
            - KafkaClientInstanceSecurityGroup
            - GroupId
        - IpProtocol: tcp
          FromPort: 9092
          ToPort: 9092
          SourceSecurityGroupId: !GetAtt 
            - KafkaClientInstanceSecurityGroup
            - GroupId
  MSKCluster:
    Type: 'AWS::MSK::Cluster'
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
          - !Ref PrivateSubnetOne
          - !Ref PrivateSubnetTwo
        InstanceType: kafka.t3.small
        SecurityGroups:
          - !GetAtt 
            - MSKSecurityGroup
            - GroupId
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 2000
      ClusterName: MSKCluster
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS
          InCluster: true
      EnhancedMonitoring: PER_TOPIC_PER_BROKER
      KafkaVersion: 2.2.1
      NumberOfBrokerNodes: 2 # must be multiple of zones
Outputs:
  KafkaClientInstanceSecurityGroupID:
    Description: The ID of the security group created for the application
    Value: !GetAtt 
      - KafkaClientInstanceSecurityGroup
      - GroupId
  MSKSecurityGroupID:
    Description: The ID of the security group created for the MSK clusters
    Value: !GetAtt 
      - MSKSecurityGroup
      - GroupId
  MSKClusterArn:
    Description: The Arn for the MSKMMCluster1 MSK cluster
    Value: !Ref MSKCluster

vpc.cfn.yml

AWSTemplateFormatVersion: '2010-09-09'

Description:  This template deploys a VPC, with a pair of public and private subnets spread
  across two Availability Zones. It deploys an internet gateway, with a default
  route on the public subnets. It deploys a pair of NAT gateways (one in each AZ),
  and default routes for them in the private subnets.

Parameters:
  EnvironmentName:
    Description: An environment name that is prefixed to resource names
    Type: String

  VpcCIDR:
    Description: Please enter the IP range (CIDR notation) for this VPC
    Type: String
    Default: 10.192.0.0/16

  PublicSubnet1CIDR:
    Description: Please enter the IP range (CIDR notation) for the public subnet in the first Availability Zone
    Type: String
    Default: 10.192.10.0/24

  PublicSubnet2CIDR:
    Description: Please enter the IP range (CIDR notation) for the public subnet in the second Availability Zone
    Type: String
    Default: 10.192.11.0/24

  PrivateSubnet1CIDR:
    Description: Please enter the IP range (CIDR notation) for the private subnet in the first Availability Zone
    Type: String
    Default: 10.192.20.0/24

  PrivateSubnet2CIDR:
    Description: Please enter the IP range (CIDR notation) for the private subnet in the second Availability Zone
    Type: String
    Default: 10.192.21.0/24

Resources:
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: !Ref VpcCIDR
      EnableDnsSupport: true
      EnableDnsHostnames: true
      Tags:
        - Key: Name
          Value: !Ref EnvironmentName

  InternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags:
        - Key: Name
          Value: !Ref EnvironmentName

  InternetGatewayAttachment:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      InternetGatewayId: !Ref InternetGateway
      VpcId: !Ref VPC

  PublicSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      AvailabilityZone: !Select [ 0, !GetAZs '' ]
      CidrBlock: !Ref PublicSubnet1CIDR
      MapPublicIpOnLaunch: true
      Tags:
        - Key: Name
          Value: !Sub ${EnvironmentName} Public Subnet (AZ1)

  PublicSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      AvailabilityZone: !Select [ 1, !GetAZs  '' ]
      CidrBlock: !Ref PublicSubnet2CIDR
      MapPublicIpOnLaunch: true
      Tags:
        - Key: Name
          Value: !Sub ${EnvironmentName} Public Subnet (AZ2)

  PrivateSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      AvailabilityZone: !Select [ 0, !GetAZs  '' ]
      CidrBlock: !Ref PrivateSubnet1CIDR
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Sub ${EnvironmentName} Private Subnet (AZ1)

  PrivateSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      AvailabilityZone: !Select [ 1, !GetAZs  '' ]
      CidrBlock: !Ref PrivateSubnet2CIDR
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Sub ${EnvironmentName} Private Subnet (AZ2)

  NatGateway1EIP:
    Type: AWS::EC2::EIP
    DependsOn: InternetGatewayAttachment
    Properties:
      Domain: vpc

  NatGateway2EIP:
    Type: AWS::EC2::EIP
    DependsOn: InternetGatewayAttachment
    Properties:
      Domain: vpc

  NatGateway1:
    Type: AWS::EC2::NatGateway
    Properties:
      AllocationId: !GetAtt NatGateway1EIP.AllocationId
      SubnetId: !Ref PublicSubnet1

  NatGateway2:
    Type: AWS::EC2::NatGateway
    Properties:
      AllocationId: !GetAtt NatGateway2EIP.AllocationId
      SubnetId: !Ref PublicSubnet2

  PublicRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub ${EnvironmentName} Public Routes

  DefaultPublicRoute:
    Type: AWS::EC2::Route
    DependsOn: InternetGatewayAttachment
    Properties:
      RouteTableId: !Ref PublicRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref InternetGateway

  PublicSubnet1RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PublicRouteTable
      SubnetId: !Ref PublicSubnet1

  PublicSubnet2RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PublicRouteTable
      SubnetId: !Ref PublicSubnet2


  PrivateRouteTable1:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub ${EnvironmentName} Private Routes (AZ1)

  DefaultPrivateRoute1:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId: !Ref PrivateRouteTable1
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId: !Ref NatGateway1

  PrivateSubnet1RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable1
      SubnetId: !Ref PrivateSubnet1

  PrivateRouteTable2:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: !Sub ${EnvironmentName} Private Routes (AZ2)

  DefaultPrivateRoute2:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId: !Ref PrivateRouteTable2
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId: !Ref NatGateway2

  PrivateSubnet2RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable2
      SubnetId: !Ref PrivateSubnet2

  NoIngressSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupName: "no-ingress-sg"
      GroupDescription: "Security group with no ingress rule"
      VpcId: !Ref VPC

Outputs:
  VPC:
    Description: A reference to the created VPC
    Value: !Ref VPC

  PublicSubnets:
    Description: A list of the public subnets
    Value: !Join [ ",", [ !Ref PublicSubnet1, !Ref PublicSubnet2 ]]

  PrivateSubnets:
    Description: A list of the private subnets
    Value: !Join [ ",", [ !Ref PrivateSubnet1, !Ref PrivateSubnet2 ]]

  PublicSubnet1:
    Description: A reference to the public subnet in the 1st Availability Zone
    Value: !Ref PublicSubnet1

  PublicSubnet2:
    Description: A reference to the public subnet in the 2nd Availability Zone
    Value: !Ref PublicSubnet2

  PrivateSubnet1:
    Description: A reference to the private subnet in the 1st Availability Zone
    Value: !Ref PrivateSubnet1

  PrivateSubnet2:
    Description: A reference to the private subnet in the 2nd Availability Zone
    Value: !Ref PrivateSubnet2

  NoIngressSecurityGroup:
    Description: Security group with no ingress rule
    Value: !Ref NoIngressSecurityGroup