AWSTemplateFormatVersion: '2010-09-09'
Description: >
  **WARNING** This template creates Amazon
  MSK cluster and related resources. You
  will be billed for the AWS resources used
  if you create a stack from this template.
Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups:
      - Label:
          default: "Amazon MSK cluster configuration parameters"
        Parameters:
          - MSKKafkaVersion
          - ClusterConfigARN
          - ClusterConfigRevisionNumber
      - Label:
          default: "Client CloudFormation stack name"
        Parameters:
          - ClientStackName
    ParameterLabels:
      MSKKafkaVersion:
        default: "MSK version, using only 2.7.1"
      ClusterConfigARN:
        default: "MSK Cluster configuration Arn"
      ClusterConfigRevisionNumber:
        default: "MSK Cluster configuration revision number"
      ClientStackName:
        default: "Client stack name that you created earlier"
Parameters:
  MSKKafkaVersion:
    Type: String
    Default: 2.7.1
    AllowedValues:
      - 2.7.1
  ClusterConfigARN:
    Type: String
  ClusterConfigRevisionNumber:
    Type: Number
    Default: 1
  ClientStackName:
    Type: String
    Default: "MSKClientStack"
Resources:
  MSKSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: MSK Security Group
      VpcId:
        Fn::ImportValue:
          !Sub "${ClientStackName}-VPCId"
      SecurityGroupIngress:
      - Description: allows 2181 (MSK Zookeeper) access from security group associated with Kafka clients EC2 instances
        IpProtocol: tcp
        FromPort: 2181
        ToPort: 2181
        SourceSecurityGroupId:
          Fn::ImportValue:
            !Sub "${ClientStackName}-KafkaClientEC2InstanceSecurityGroupId"
      - Description: allows 9092 (MSK plaintext) access from security group associated with Kafka clients EC2 instances
        IpProtocol: tcp
        FromPort: 9092
        ToPort: 9092
        SourceSecurityGroupId:
          Fn::ImportValue:
            !Sub "${ClientStackName}-KafkaClientEC2InstanceSecurityGroupId"
      SecurityGroupEgress:
      - Description: all egress allowed
        IpProtocol: -1
        CidrIp: 0.0.0.0/0
  MSKCluster:
    Type: AWS::MSK::Cluster
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
          - Fn::ImportValue:
              !Sub "${ClientStackName}-PrivateSubnetMSKOne"
          - Fn::ImportValue:
              !Sub "${ClientStackName}-PrivateSubnetMSKTwo"
          - Fn::ImportValue:
              !Sub "${ClientStackName}-PrivateSubnetMSKThree"
        InstanceType: kafka.m5.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 100
      ClusterName: !Ref 'AWS::StackName'
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS_PLAINTEXT
          InCluster: true
      ConfigurationInfo:
        Arn: !Ref ClusterConfigARN
        Revision: !Ref ClusterConfigRevisionNumber
      EnhancedMonitoring: PER_TOPIC_PER_BROKER
      KafkaVersion: !Ref MSKKafkaVersion
      NumberOfBrokerNodes: 3
Outputs:
  MSKClusterArn:
    Description: MSK Cluster Arn
    Value: !Ref MSKCluster
    Export:
      Name: !Sub "${AWS::StackName}-ClusterArn"