AWSTemplateFormatVersion: '2010-09-09' Parameters: MSKSourceKafkaVersion: Type: String Default: 2.3.1 Description: The Apache Kafka version for the source Amazon MSK cluster. AllowedValues: - 2.2.1 - 2.3.1 - 2.4.1.1 - 2.5.1 - 2.6.0 - 2.6.1 - 2.7.0 MSKDestinationKafkaVersion: Type: String Default: 2.4.1.1 Description: The Apache Kafka version for the destination Amazon MSK cluster. AllowedValues: - 2.2.1 - 2.3.1 - 2.4.1.1 - 2.5.1 - 2.6.0 - 2.6.1 - 2.7.0 TLSMutualAuthenticationSourceMSKCluster: Type: String Default: false Description: Whether TLS Mutual Auth should be enabled for the Amazon MSK cluster1. AllowedValues: - true - false TLSMutualAuthenticationDestinationMSKCluster: Type: String Default: false Description: Whether TLS Mutual Auth should be enabled for the Amazon MSK cluster2. AllowedValues: - true - false PCAARNSourceMSKCluster: Type: String AllowedPattern: 'arn:aws:acm-pca:[us\-east\-1|us\-east\-2|eu\-west\-1]{9}:\d{12}:certificate-authority\/[\da-f]{8}-[\da-f]{4}-[\da-f]{4}-[\da-f]{4}-[\da-f]{12}|^$' ConstraintDescription: Not a Valid ACM PCA ARN Description: Provide the ARN for an ACM PCA in your account for the source MSK Cluster PCAARNDestinationMSKCluster: Type: String AllowedPattern: 'arn:aws:acm-pca:[us\-east\-1|us\-east\-2|eu\-west\-1]{9}:\d{12}:certificate-authority\/[\da-f]{8}-[\da-f]{4}-[\da-f]{4}-[\da-f]{4}-[\da-f]{12}|^$' ConstraintDescription: Not a Valid ACM PCA ARN Description: Provide the ARN for an ACM PCA in your account for the destination MSK Cluster VPCStack: Description: The name of the VPC stack Type: String BastionStack: Description: The name of the Bastion/Kafka client instance stack Type: String Conditions: MTLSMSKCluster1: !Equals [ !Ref TLSMutualAuthenticationSourceMSKCluster, true ] noMTLSMSKCluster1: !Equals [ !Ref TLSMutualAuthenticationSourceMSKCluster, false ] MTLSMSKCluster2: !Equals [ !Ref TLSMutualAuthenticationDestinationMSKCluster, true ] noMTLSMSKCluster2: !Equals [ !Ref TLSMutualAuthenticationDestinationMSKCluster, false ] Resources: MSKSecurityGroup: Type: AWS::EC2::SecurityGroup Properties: GroupDescription: MSK Security Group VpcId: Fn::ImportValue: !Sub "${VPCStack}-VPCID" SecurityGroupIngress: - IpProtocol: tcp FromPort: 2181 ToPort: 2181 SourceSecurityGroupId: Fn::ImportValue: !Sub "${BastionStack}-KafkaClientEC2InstanceSecurityGroupId" - IpProtocol: tcp FromPort: 9094 ToPort: 9094 SourceSecurityGroupId: Fn::ImportValue: !Sub "${BastionStack}-KafkaClientEC2InstanceSecurityGroupId" - IpProtocol: tcp FromPort: 9092 ToPort: 9092 SourceSecurityGroupId: Fn::ImportValue: !Sub "${BastionStack}-KafkaClientEC2InstanceSecurityGroupId" MSKSourceClusterMTLS: Type: AWS::MSK::Cluster Condition: MTLSMSKCluster1 Properties: BrokerNodeGroupInfo: ClientSubnets: - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKOne" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKTwo" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKThree" InstanceType: kafka.m5.large SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId] StorageInfo: EBSStorageInfo: VolumeSize: 1000 ClusterName: !Join - '-' - - 'MSKCluster1' - !Ref 'AWS::StackName' EncryptionInfo: EncryptionInTransit: ClientBroker: TLS_PLAINTEXT InCluster: true ClientAuthentication: Tls: CertificateAuthorityArnList: - !Ref PCAARNSourceMSKCluster EnhancedMonitoring: DEFAULT KafkaVersion: !Ref MSKSourceKafkaVersion NumberOfBrokerNodes: 3 MSKSourceClusterNoMTLS: Type: AWS::MSK::Cluster Condition: noMTLSMSKCluster1 Properties: BrokerNodeGroupInfo: ClientSubnets: - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKOne" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKTwo" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKThree" InstanceType: kafka.m5.large SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId] StorageInfo: EBSStorageInfo: VolumeSize: 1000 ClusterName: !Join - '-' - - 'MSKCluster1' - !Ref 'AWS::StackName' EncryptionInfo: EncryptionInTransit: ClientBroker: TLS_PLAINTEXT InCluster: true EnhancedMonitoring: DEFAULT KafkaVersion: !Ref MSKSourceKafkaVersion NumberOfBrokerNodes: 3 MSKDestinationClusterMTLS: Type: AWS::MSK::Cluster Condition: MTLSMSKCluster2 Properties: BrokerNodeGroupInfo: ClientSubnets: - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKOne" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKTwo" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKThree" InstanceType: kafka.m5.large SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId] StorageInfo: EBSStorageInfo: VolumeSize: 1000 ClusterName: !Join - '-' - - 'MSKCluster2' - !Ref 'AWS::StackName' EncryptionInfo: EncryptionInTransit: ClientBroker: TLS_PLAINTEXT InCluster: true ClientAuthentication: Tls: CertificateAuthorityArnList: - !Ref PCAARNDestinationMSKCluster EnhancedMonitoring: DEFAULT KafkaVersion: !Ref MSKDestinationKafkaVersion NumberOfBrokerNodes: 3 MSKDestinationClusterNoMTLS: Type: AWS::MSK::Cluster Condition: noMTLSMSKCluster2 Properties: BrokerNodeGroupInfo: ClientSubnets: - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKOne" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKTwo" - Fn::ImportValue: !Sub "${VPCStack}-PrivateSubnetMSKThree" InstanceType: kafka.m5.large SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId] StorageInfo: EBSStorageInfo: VolumeSize: 1000 ClusterName: !Join - '-' - - 'MSKCluster2' - !Ref 'AWS::StackName' EncryptionInfo: EncryptionInTransit: ClientBroker: TLS_PLAINTEXT InCluster: true EnhancedMonitoring: DEFAULT KafkaVersion: !Ref MSKDestinationKafkaVersion NumberOfBrokerNodes: 3 Outputs: MSKSourceClusterArn: Description: The Arn for the Source MSK cluster Value: !If [MTLSMSKCluster1, !Ref 'MSKSourceClusterMTLS', !Ref 'MSKSourceClusterNoMTLS'] MSKDestinationClusterArn: Description: The Arn for the Destination MSK cluster Value: !If [MTLSMSKCluster2, !Ref 'MSKDestinationClusterMTLS', !Ref 'MSKDestinationClusterNoMTLS'] MSKSecurityGroupID: Description: The ID of the security group created for the MSK clusters Value: !GetAtt MSKSecurityGroup.GroupId SSHKafkaClientEC2Instance1: Description: SSH command for Kafka the EC2 instance1 Value: Fn::ImportValue: !Sub "${BastionStack}-SSHKafkaClientEC2Instance1" SSHKafkaClientEC2Instance2: Description: SSH command for Kafka the EC2 instance2 Value: Fn::ImportValue: !Sub "${BastionStack}-SSHKafkaClientEC2Instance2" KafkaClientEC2InstanceSecurityGroupId: Description: The security group id for the EC2 instance Value: Fn::ImportValue: !Sub "${BastionStack}-KafkaClientEC2InstanceSecurityGroupId" MSKClustersSecurityGroupId: Description: The security group id for the Amazon MSK clusters Value: !GetAtt MSKSecurityGroup.GroupId SchemaRegistryKafkaClientEC2Instance1Url: Description: The url for the Schema Registry Value: Fn::ImportValue: !Sub "${BastionStack}-SchemaRegistryKafkaClientEC2Instance1Url" SchemaRegistryKafkaClientEC2Instance2Url: Description: The url for the Schema Registry Value: Fn::ImportValue: !Sub "${BastionStack}-SchemaRegistryKafkaClientEC2Instance2Url"