AWS CloudFormation template for AWS MSK demo

msk-connect-demo-cfn-template.yml

AWSTemplateFormatVersion: '2010-09-09'
Parameters:
  DatabasePassword:
    AllowedPattern: '[a-zA-Z0-9]+'
    ConstraintDescription: must contain only alphanumeric characters. Must have length 8-41.
    Description: Database admin account password.
    MaxLength: '41'
    MinLength: '8'
    Default: 'S3cretPwd99'
    Type: String
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
  
  AuroraDBSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupName: AuroraDBSecurityGroup
      GroupDescription: Enable SSH access via port 22
      VpcId: !Ref VPC
      SecurityGroupIngress:
      - IpProtocol: tcp
        FromPort: 3306
        ToPort: 3306
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 3306
        ToPort: 3306
        SourceSecurityGroupId: !GetAtt MSKSecurityGroup.GroupId
      SecurityGroupEgress:
        - IpProtocol: -1
          CidrIp: 0.0.0.0/0
  
  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 python3.7 -y
          yum install java-1.8.0-openjdk-devel -y
          yum install nmap-ncat -y
          yum install git -y
          yum erase awscli -y
          yum install jq -y
          sudo yum install -y mysql-devel
          sudo yum -y install mysql
          amazon-linux-extras install docker -y
          service docker start
          usermod -a -G docker ec2-user

          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

          # install AWS CLI 2 - access with aws2
          curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
          unzip awscliv2.zip
          ./aws/install -b /usr/local/bin/aws2
          su -c "ln -s /usr/local/bin/aws2/aws ~/.local/bin/aws2" -s /bin/sh ec2-user

          su -c "mkdir -p /tmp/kafka" -s /bin/sh ec2-user
          su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/schema-registry.properties /tmp/kafka" -l ec2-user
          su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/producer.properties_msk /tmp/kafka" -l ec2-user
          su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/KafkaClickstreamClient-1.0-SNAPSHOT.jar/KafkaClickstreamClient-1.0-SNAPSHOT.jar /tmp/kafka" -l ec2-user

          # get Confluent community
          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/5.3/confluent-community-5.3.1-2.12.tar.gz" -s /bin/sh ec2-user
          su -c "tar -xzf confluent-community-5.3.1-2.12.tar.gz --strip 1" -s /bin/sh ec2-user

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

          echo -n "
          [Unit]
          Description=Confluent Schema Registry
          After=network.target

          [Service]
          Type=simple
          User=ec2-user
          ExecStart=/bin/sh -c '/home/ec2-user/confluent/bin/schema-registry-start /tmp/kafka/schema-registry.properties > /tmp/kafka/schema-registry.log 2>&1'
          ExecStop=/home/ec2-user/confluent/bin/schema-registry-stop
          Restart=on-abnormal

          [Install]
          WantedBy=multi-user.target" > /etc/systemd/system/confluent-schema-registry.service
          
          cd /usr/local/bin/aws2
          ./aws --region ${AWS::Region} cloudformation wait stack-create-complete --stack-name ${AWS::StackName}
          rds_db_name=salesdb
          # get rds endpoint
          RDS_AURORA_ENDPOINT=`./aws rds describe-db-instances --region ${AWS::Region} | jq -r '.DBInstances[] | select(.DBName == "salesdb") | .Endpoint.Address'`
          echo 'aurora_endpoint : '$RDS_AURORA_ENDPOINT

          cd /home/ec2-user
          # Load the data into Aurora
          su -c "mkdir -p scripts" -l ec2-user
          su -c "aws s3 cp s3://emr-workshops-us-west-2/glue_immersion_day/scripts/salesdb.sql /home/ec2-user/scripts/salesdb.sql" -l ec2-user
          su -c "mysql -f -u master -h $RDS_AURORA_ENDPOINT  --password=${DatabasePassword} < /home/ec2-user/scripts/salesdb.sql" -l ec2-user

  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: AllowS3Access
          PolicyDocument: !Sub '{
              "Version": "2012-10-17",
              "Statement": [
                  {
                      "Effect": "Allow",
                      "Action": [
                          "s3:ListBucket",
                          "s3:ListBuckets"
                      ],
                      "Resource": [
                        "${S3ConnectorPluginsBucket.Arn}"
                      ]
                  },
                  {
                      "Effect": "Allow",
                      "Action": [
                          "s3:PutObject",
                          "s3:GetObject",
                          "s3:DeleteObject"
                      ],
                      "Resource": [
                        "${S3ConnectorPluginsBucket.Arn}/*"
                      ]
                  },
                  {
                      "Effect": "Allow",
                      "Action": [
                          "s3:GetObject"
                      ],
                      "Resource": [
                          "arn:aws:s3:::aws-streaming-artifacts/*",
                          "arn:aws:s3:::emr-workshops-us-west-2/*"
                      ]
                  }
              ]
          }'
        - 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
  AuroraConnectorCWGroup:
    Type: AWS::Logs::LogGroup
    Properties: 
      LogGroupName: /msk-connect-demo-cwlog-group
      RetentionInDays: 3
    
  S3ConnectorPluginsBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub 'msk-lab-${AWS::AccountId}-plugins-bucket'

  AuroraConnectorIAMRole:
    Type: AWS::IAM::Role
    Properties: 
      AssumeRolePolicyDocument: 
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: kafkaconnect.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: "/"
      Policies:
        - PolicyName: CloudWatchLogsPolicy
          PolicyDocument: '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "logs:CreateLogGroup",
                        "logs:CreateLogStream",
                        "logs:PutLogEvents"
                    ],
                    "Resource": [
                        "arn:aws:logs:*:*:/msk-connect-demo-cwlog-group/*"
                    ]
                }
            ]
        }'
        - PolicyName: MSKConnectAuthentication
          PolicyDocument: !Sub '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "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}/*"
                    ]
                }
            ]
        }'
  DatagenConnectorIAMRole:
    Type: AWS::IAM::Role
    Properties: 
      AssumeRolePolicyDocument: 
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: kafkaconnect.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: "/"
      Policies:
        - PolicyName: CloudWatchLogsPolicy
          PolicyDocument: '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "logs:CreateLogGroup",
                        "logs:CreateLogStream",
                        "logs:PutLogEvents"
                    ],
                    "Resource": [
                        "arn:aws:logs:*:*:/msk-connect-demo-cwlog-group/*"
                    ]
                }
            ]
        }'
        - PolicyName: MSKConnectAuthentication
          PolicyDocument: !Sub '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "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}/*"
                    ]
                }
            ]
        }'
  DynamoDBConnectorIAMRole:
    Type: AWS::IAM::Role
    Properties: 
      AssumeRolePolicyDocument: 
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: kafkaconnect.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: "/"
      Policies:
        - PolicyName: CloudWatchLogsPolicy
          PolicyDocument: '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "logs:CreateLogGroup",
                        "logs:CreateLogStream",
                        "logs:PutLogEvents"
                    ],
                    "Resource": [
                        "arn:aws:logs:*:*:/msk-connect-demo-cwlog-group/*"
                    ]
                }
            ]
        }'
        - PolicyName: DynamoDBAccess
          PolicyDocument: '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "VisualEditor0",
                    "Effect": "Allow",
                    "Action": [
                        "dynamodb:ListContributorInsights",
                        "dynamodb:DescribeReservedCapacityOfferings",
                        "dynamodb:ListGlobalTables",
                        "dynamodb:ListTables",
                        "dynamodb:DescribeReservedCapacity",
                        "dynamodb:ListBackups",
                        "dynamodb:PurchaseReservedCapacityOfferings",
                        "dynamodb:DescribeLimits",
                        "dynamodb:ListExports",
                        "dynamodb:ListStreams"
                    ],
                    "Resource": "*"
                },
                {
                    "Sid": "VisualEditor1",
                    "Effect": "Allow",
                    "Action": "dynamodb:*",
                    "Resource": "arn:aws:dynamodb:*:568863012249:table/*"
                }
            ]
        }'
        - PolicyName: MSKConnectAuthentication
          PolicyDocument: !Sub '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "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}/*"
                    ]
                }
            ]
        }'
  
  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: 2.6.2
      NumberOfBrokerNodes: 3
      ClientAuthentication:
        Sasl:
          Iam: 
            Enabled: True
        Unauthenticated:
          Enabled: True
  DatabaseSubnetGroup:
    Type: AWS::RDS::DBSubnetGroup
    Properties:
      DBSubnetGroupDescription: CloudFormation managed DB subnet group.
      SubnetIds:
        - !Ref 'PublicSubnetOne'
        - !Ref 'PublicSubnetTwo'

  AuroraDBParameterGroup:
    Type: AWS::RDS::DBParameterGroup
    Properties:
      Description: MSK Worshop DB parameter group
      Family: aurora-mysql5.7
      Parameters:
        max_connections: 300

  AuroraDBClusterParameterGroup:
    Type: AWS::RDS::DBClusterParameterGroup
    Properties:
      Description: 'CloudFormation Sample Aurora Cluster Parameter Group'
      Family: aurora-mysql5.7
      Parameters:
        time_zone: US/Eastern
        binlog_format: ROW
        binlog_checksum: NONE

  AuroraCluster:
    Type: AWS::RDS::DBCluster
    DependsOn:
      - DatabaseSubnetGroup
    Properties:
      Engine: aurora-mysql
      MasterUsername: 'master'
      MasterUserPassword: !Ref DatabasePassword
      DatabaseName: 'salesdb'
      DBSubnetGroupName: !Ref DatabaseSubnetGroup
      DBClusterParameterGroupName: !Ref AuroraDBClusterParameterGroup
      VpcSecurityGroupIds:
        - !Ref 'AuroraDBSecurityGroup'

  AuroraDB:
    Type: AWS::RDS::DBInstance
    DependsOn: AuroraCluster
    Properties:
      Engine: aurora-mysql
      DBClusterIdentifier: !Ref AuroraCluster
      DBInstanceClass: db.r5.large
      DBSubnetGroupName: !Ref DatabaseSubnetGroup
      DBParameterGroupName: !Ref AuroraDBParameterGroup
      PubliclyAccessible: 'false'
      DBInstanceIdentifier: !Join [ '-', ['mask-lab', 'salesdb'] ]
      Tags:
        - Key: 'Name'
          Value: !Ref AWS::StackName

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}
  S3ConnectorPluginsBucketName:
    Description: The S3 Bucket for storing the MSK Connect Custom Plugins
    Value: !Join
      - ''
      - - 's3://'
        - !Ref S3ConnectorPluginsBucket
  SchemaRegistryPrivateDNS:
    Description: The Private DNS for the Schema Registry
    Value: !Sub http://${KafkaClientEC2Instance.PrivateDnsName}:8081
  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