AWSTemplateFormatVersion: '2010-09-09'
Mappings:
  SubnetConfig:
      VPC:
        CIDR: '10.0.0.0/16'
      PublicOne:
        CIDR: '10.0.0.0/24'
      PublicTwo:
        CIDR: '10.0.1.0/24'
      PrivateSubnetMSKOne:
        CIDR: '10.0.2.0/24'
      PrivateSubnetMSKTwo:
        CIDR: '10.0.3.0/24'
      PrivateSubnetMSKThree:
        CIDR: '10.0.4.0/24'
  RegionAMI:
      us-east-1:
        HVM64: ami-02e136e904f3da870
      us-east-2:
        HVM64: ami-074cce78125f09d61
      us-west-2:
        HVM64: ami-013a129d325529d4d
      eu-west-1:
        HVM64: ami-03ab7423a204da002
Resources:
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      EnableDnsSupport: true
      EnableDnsHostnames: true
      CidrBlock: !FindInMap ['SubnetConfig', 'VPC', 'CIDR']
      Tags:
        - Key: 'Name'
          Value: 'MMVPC'

  PublicSubnetOne:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 0
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PublicOne', 'CIDR']
      MapPublicIpOnLaunch: true
      Tags:
        - Key: 'Name'
          Value: 'PublicSubnetOne'
  PublicSubnetTwo:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 1
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PublicTwo', 'CIDR']
      MapPublicIpOnLaunch: true
      Tags:
        - Key: 'Name'
          Value: 'PublicSubnetTwo'
  PrivateSubnetMSKOne:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 0
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PrivateSubnetMSKOne', 'CIDR']
      MapPublicIpOnLaunch: false
      Tags:
        - Key: 'Name'
          Value: 'PrivateSubnetMSKOne'
  PrivateSubnetMSKTwo:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 1
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PrivateSubnetMSKTwo', 'CIDR']
      MapPublicIpOnLaunch: false
      Tags:
        - Key: 'Name'
          Value: 'PrivateSubnetMSKTwo'
  PrivateSubnetMSKThree:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 2
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PrivateSubnetMSKThree', 'CIDR']
      MapPublicIpOnLaunch: false
      Tags:
        - Key: 'Name'
          Value: 'PrivateSubnetMSKThree'

  InternetGateway:
    Type: AWS::EC2::InternetGateway    
  GatewayAttachement:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      VpcId: !Ref 'VPC'
      InternetGatewayId: !Ref 'InternetGateway'
  NatGateway1EIP: 
    Type: AWS::EC2::EIP
    DependsOn: GatewayAttachement
    Properties:
      Domain: vpc
  NATGateway:
    Type: AWS::EC2::NatGateway
    Properties: 
      AllocationId: 
        Fn::GetAtt:
         - NatGateway1EIP
         - AllocationId
      SubnetId: !Ref PublicSubnetOne
  PublicRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref 'VPC'
  PublicRoute:
    Type: AWS::EC2::Route
    DependsOn: GatewayAttachement
    Properties:
      RouteTableId: !Ref 'PublicRouteTable'
      DestinationCidrBlock: '0.0.0.0/0'
      GatewayId: !Ref 'InternetGateway'
  PublicSubnetOneRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PublicSubnetOne
      RouteTableId: !Ref PublicRouteTable
  PublicSubnetTwoRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PublicSubnetTwo
      RouteTableId: !Ref PublicRouteTable
  PrivateRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref 'VPC'
  PrivateInternetRoute:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId:
        Ref: PrivateRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId:
        Ref: NATGateway
  PrivateSubnetMSKOneRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnetMSKOne
  PrivateSubnetMSKTwoRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnetMSKTwo
  PrivateSubnetMSKThreeRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnetMSKThree

  KafkaClientInstanceSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Enable SSH access via port 22 and 8081
      VpcId: !Ref 'VPC'
      
  MSKSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: MSK Security Group
      VpcId: !Ref 'VPC'
      SecurityGroupIngress:
      - IpProtocol: tcp
        FromPort: 2181
        ToPort: 2181
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 9098
        ToPort: 9098
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 9094
        ToPort: 9094
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 9092
        ToPort: 9092
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
  MSKSelfIngressAllowRule:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn: MSKSecurityGroup
    Properties:
      GroupId: !Ref MSKSecurityGroup
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref MSKSecurityGroup
  KafkaClientEC2Instance:
    Type: AWS::EC2::Instance
    Properties:
      InstanceType: m5.large
      IamInstanceProfile: !Ref EC2InstanceProfile
      AvailabilityZone: 
        Fn::Select:
         - 0
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      SubnetId: !Ref 'PublicSubnetOne'
      SecurityGroupIds: [!GetAtt KafkaClientInstanceSecurityGroup.GroupId]
      ImageId: !FindInMap ['RegionAMI', !Ref 'AWS::Region', 'HVM64']
      Tags:
        - Key: 'Name'
          Value: 'KafkaClientInstance'
      BlockDeviceMappings:
        - DeviceName: /dev/xvda
          Ebs:
            VolumeSize: 20
            VolumeType: gp2
            DeleteOnTermination: true
      UserData: 
        Fn::Base64: 
          !Sub |
          #!/bin/bash
          set -e
          yum update -y
          yum install java-1.8.0-openjdk-devel -y
          yum install git -y
          
          cd /home/ec2-user
          su -c "mkdir -p confluent kafka" -s /bin/sh ec2-user
          cd confluent
          su -c "wget http://packages.confluent.io/archive/7.5/confluent-community-7.5.0.tar.gz" -s /bin/sh ec2-user
          su -c "tar -xzf confluent-community-7.5.0.tar.gz --strip 1" -s /bin/sh ec2-user

          cd /home/ec2-user
          cd kafka
          su -c "wget https://archive.apache.org/dist/kafka/3.7.0/kafka_2.13-3.7.0.tgz" -s /bin/sh ec2-user
          su -c "tar -xzf kafka_2.13-3.7.0.tgz --strip 1" -s /bin/sh ec2-user

          # get iam auth jar
          cd libs
          wget https://github.com/aws/aws-msk-iam-auth/releases/download/v1.1.1/aws-msk-iam-auth-1.1.1-all.jar

          echo -n "
          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" > /home/ec2-user/kafka/bin/client.properties

          echo -n "installed blog version1" > /home/ec2-user/readme.txt
          

  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/AWSCloudFormationReadOnlyAccess
        - arn:aws:iam::aws:policy/AmazonRDSFullAccess
        - arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
      Policies:
        - PolicyName: MSKConfigurationAccess
          PolicyDocument: !Sub '{
              "Version": "2012-10-17",
              "Statement": [
                  {
                      "Sid": "VisualEditor0",
                      "Effect": "Allow",
                      "Action": "kafka:CreateConfiguration",
                      "Resource": "*"
                  }
              ]
          }'
        - PolicyName: MSKConnectAuthentication
          PolicyDocument: !Sub '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "kafka-cluster:*Topic*",
                        "kafka-cluster:Connect",
                        "kafka-cluster:AlterCluster",
                        "kafka-cluster:DescribeCluster",
                        "kafka-cluster:DescribeClusterDynamicConfiguration"
                    ],
                    "Resource": [
                        "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:cluster/MSKCluster-${AWS::StackName}/*"
                    ]
                },
                {
                    "Effect": "Allow",
                    "Action": [
                        "kafka-cluster:*Topic*",
                        "kafka-cluster:WriteData",
                        "kafka-cluster:ReadData"
                    ],
                    "Resource": [
                        "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:topic/MSKCluster-${AWS::StackName}/*"
                    ]
                },
                {
                    "Effect": "Allow",
                    "Action": [
                        "kafka-cluster:AlterGroup",
                        "kafka-cluster:DescribeGroup"
                    ],
                    "Resource": [
                        "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:group/MSKCluster-${AWS::StackName}/*"
                    ]
                }
            ]
        }'
  EC2InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties: 
      InstanceProfileName: !Join
                            - '-'
                            - - 'EC2MSKCFProfile'
                              - !Ref 'AWS::StackName'
      Roles:
        - !Ref EC2Role
  MSKCluster:
    Type: AWS::MSK::Cluster
    Properties: 
      BrokerNodeGroupInfo: 
        ClientSubnets: [!Ref 'PrivateSubnetMSKOne', !Ref 'PrivateSubnetMSKTwo', !Ref 'PrivateSubnetMSKThree']
        InstanceType: kafka.m5.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
        StorageInfo: 
          EBSStorageInfo:
            VolumeSize: 1000
      ClusterName: !Join
                    - '-'
                    - - 'MSKCluster'
                      - !Ref 'AWS::StackName'
      EncryptionInfo: 
        EncryptionInTransit: 
          ClientBroker: TLS_PLAINTEXT
          InCluster: true
      EnhancedMonitoring: DEFAULT
      KafkaVersion: 3.5.1
      NumberOfBrokerNodes: 3
      ClientAuthentication:
        Sasl:
          Iam: 
            Enabled: True
        Unauthenticated:
          Enabled: True


  



  

Outputs:
  VPCId: 
    Description: The ID of the VPC created
    Value: !Ref 'VPC'
  PublicSubnetOne: 
    Description: The name of the public subnet created
    Value: !Ref 'PublicSubnetOne'
  PrivateSubnetMSKOne: 
    Description: The ID of private subnet one created
    Value: !Ref 'PrivateSubnetMSKOne'
  PrivateSubnetMSKTwo: 
    Description: The ID of private subnet two created
    Value: !Ref 'PrivateSubnetMSKTwo'
  PrivateSubnetMSKThree: 
    Description: The ID of private subnet three created
    Value: !Ref 'PrivateSubnetMSKThree'
  MSKSecurityGroupID: 
    Description: The ID of the security group created for the MSK clusters
    Value: !GetAtt MSKSecurityGroup.GroupId
  KafkaClientEC2InstancePublicDNS:
    Description: The Public DNS for the EC2 instance
    Value: !GetAtt KafkaClientEC2Instance.PublicDnsName
  KafkaClientEC2InstanceSsh:
    Description: The SSH for the EC2 instance
    Value: !Sub ssh -A ec2-user@${KafkaClientEC2Instance.PublicDnsName}
  MSKClusterArn:
    Description: The Arn for the MSKMMCluster1 MSK cluster
    Value: !Ref 'MSKCluster'
  KafkaClientEC2InstanceSecurityGroupId:
    Description: The security group id for the EC2 instance
    Value: !GetAtt KafkaClientInstanceSecurityGroup.GroupId