Made updates to CloudFormation template, SAM template file and AvroProducerHandler.java file to take care of correctly passing environment variables and also not have the user type in VPC, Subnets and Security Group

Author: indranil-banerjee-aws

msk-lambda-schema-avro-java-sam/MSKAndKafkaClientEC2.yaml

@@ -33,7 +33,11 @@ Parameters:
     Default: https://github.com/aws-samples/serverless-patterns/
   ContactSchemaName:
     Type: String
-    Default: ContactSchema
+    Default: ContactSchema   
+  GlueSchemaRegistryForMSKName:
+    Type: String
+    Default: GlueSchemaRegistryForMSK
+
 Conditions:
   CreateProvisionedCluster: !Equals 
     - !Ref EnvType
@@ -271,13 +275,21 @@ Resources:
       FromPort: 22
       ToPort: 22
       SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
-      
+
+  MSKGlueRegistry:
+    Type: AWS::Glue::Registry
+    Properties:
+      Name: !Ref GlueSchemaRegistryForMSKName
+      Description: "Registry for storing schemas related to MSK"
+
   ContactSchema:
     Type: AWS::Glue::Schema
     Properties:
       Name: !Ref ContactSchemaName
       Compatibility: BACKWARD
       DataFormat: AVRO
+      Registry: 
+        Arn: !GetAtt MSKGlueRegistry.Arn
       SchemaDefinition: |
         {
           "type": "record",
@@ -543,11 +555,26 @@ Resources:
             cd ./serverless-patterns/msk-lambda-schema-avro-java-sam
             cp template_original.yaml template.yaml
             sudo chown -R ec2-user .
+            
+            GLUE_SCHEMA_REGISTRY_NAME=${glue_registry_name}
+            CONTACT_SCHEMA=${contact_schema_name}
+            VPC_ID=${vpcid}
+            LAMBDA_SECURITY_GROUP_ID={securitygroup}
+            PRIVATE_SUBNET_1={privatesubnetone}
+            PRIVATE_SUBNET_2={privatesubnettwo}
+            PRIVATE_SUBNET_3={privatesubnetthree}
+            SUBNET_IDS="$PRIVATE_SUBNET_1,$PRIVATE_SUBNET_2,$PRIVATE_SUBNET_3"
+            
             source /home/ec2-user/.bash_profile
             sed -i "s/CLUSTER_NAME/$CLUSTER_NAME/g" template.yaml
             sed -i "s/CLUSTER_ID/$CLUSTER_ID/g" template.yaml
             sed -i "s/KAFKA_TOPIC/$KAFKA_TOPIC/g" template.yaml
             sed -i "s/JAVA_VERSION/$JAVA_VERSION/g" template.yaml
+            sed -i "s/GLUE_SCHEMA_REGISTRY_NAME/$GLUE_SCHEMA_REGISTRY_NAME/g" template.yaml
+            sed -i "s/CONTACT_SCHEMA/$CONTACT_SCHEMA/g" template.yaml
+            sed -i "s/VPC_ID/$VPC_ID/g" template.yaml
+            sed -i "s/LAMBDA_SECURITY_GROUP_ID/$LAMBDA_SECURITY_GROUP_ID/g" template.yaml
+            sed -i "s/SUBNET_IDS/$SUBNET_IDS/g" template.yaml
             
             # Get IP CIDR range for EC2 Instance Connect
             cd /home/ec2-user
@@ -570,6 +597,14 @@ Resources:
             serverless_land_github_location: !Ref ServerlessLandGithubLocation
             aws_region: !Ref 'AWS::Region'
             java_version: !Ref JavaVersion
+            vpcid: !Ref VPCId
+            privatesubnetone: !Ref PrivateSubnetMSKOne
+            privatesubnettwo: !Ref PrivateSubnetMSKTwo
+            privatesubnetthree: !Ref PrivateSubnetMSKThree
+            securitygroup: !Ref SecurityGroupId
+            glue_registry_name: !Ref GlueSchemaRegistryForMSKName
+            contact_schema_name: !Ref ContactSchemaName
+            
 
   KafkaClientEC2InstanceServerless:
     Condition: CreateServerlessCluster
@@ -816,11 +851,27 @@ Resources:
             cd ./serverless-patterns/msk-lambda-schema-avro-java-sam
             cp template_original.yaml template.yaml
             sudo chown -R ec2-user .
+            
+            GLUE_SCHEMA_REGISTRY_NAME=${glue_registry_name}
+            CONTACT_SCHEMA=${contact_schema_name}
+            VPC_ID=${vpcid}
+            LAMBDA_SECURITY_GROUP_ID={securitygroup}
+            PRIVATE_SUBNET_1={privatesubnetone}
+            PRIVATE_SUBNET_2={privatesubnettwo}
+            PRIVATE_SUBNET_3={privatesubnetthree}
+            SUBNET_IDS="$PRIVATE_SUBNET_1,$PRIVATE_SUBNET_2,$PRIVATE_SUBNET_3"
+            
+            
             source /home/ec2-user/.bash_profile
             sed -i "s/CLUSTER_NAME/$CLUSTER_NAME/g" template.yaml
             sed -i "s/CLUSTER_ID/$CLUSTER_ID/g" template.yaml
             sed -i "s/KAFKA_TOPIC/$KAFKA_TOPIC/g" template.yaml
             sed -i "s/JAVA_VERSION/$JAVA_VERSION/g" template.yaml
+            sed -i "s/GLUE_SCHEMA_REGISTRY_NAME/$GLUE_SCHEMA_REGISTRY_NAME/g" template.yaml
+            sed -i "s/CONTACT_SCHEMA/$CONTACT_SCHEMA/g" template.yaml
+            sed -i "s/VPC_ID/$VPC_ID/g" template.yaml
+            sed -i "s/LAMBDA_SECURITY_GROUP_ID/$LAMBDA_SECURITY_GROUP_ID/g" template.yaml
+            sed -i "s/SUBNET_IDS/$SUBNET_IDS/g" template.yaml
             
             # Get IP CIDR range for EC2 Instance Connect
             cd /home/ec2-user
@@ -843,6 +894,15 @@ Resources:
             serverless_land_github_location: !Ref ServerlessLandGithubLocation
             aws_region: !Ref 'AWS::Region'
             java_version: !Ref JavaVersion
+            vpcid: !Ref VPCId
+            privatesubnetone: !Ref PrivateSubnetMSKOne
+            privatesubnettwo: !Ref PrivateSubnetMSKTwo
+            privatesubnetthree: !Ref PrivateSubnetMSKThree
+            securitygroup: !Ref SecurityGroupId
+            glue_registry_name: !Ref GlueSchemaRegistryForMSKName
+            contact_schema_name: !Ref ContactSchemaName
+            
+            
 
   EC2InstanceEndpoint:
     Type: AWS::EC2::InstanceConnectEndpoint
@@ -1215,6 +1275,9 @@ Resources:
           SecurityGroups:
               - !GetAtt MSKSecurityGroup.GroupId
 
+
+
+
 Outputs:
   VPCId: 
     Description: The ID of the VPC created

msk-lambda-schema-avro-java-sam/kafka_event_producer_function/src/main/java/com/amazonaws/services/lambda/samples/events/msk/AvroProducerHandler.java

@@ -32,7 +32,7 @@ public String handleRequest(Map<String, Object> event, Context context) {
         try {
             // Get environment variables
             String mskClusterArn = System.getenv("MSK_CLUSTER_ARN");
-            String kafkaTopic = System.getenv("KAFKA_TOPIC");
+            String kafkaTopic = System.getenv("MSK_TOPIC");
             String schemaName = System.getenv("CONTACT_SCHEMA_NAME");
             String region = System.getenv("AWS_REGION");
             String registryName = System.getenv("REGISTRY_NAME") != null ? 

msk-lambda-schema-avro-java-sam/template_original.yaml

@@ -125,7 +125,8 @@ Resources:
       Environment:
         Variables:
           MSK_CLUSTER_ARN: !Join [ '', ["arn:", "aws:", "kafka:", !Ref "AWS::Region" , ":" ,!Ref "AWS::AccountId", ":", "cluster/", !Ref MSKClusterName, "/" , !Ref MSKClusterId] ]
-          KAFKA_TOPIC: !Ref MSKTopic
+          MSK_TOPIC: !Ref MSKTopic
+          REGISTRY_NAME: !Ref GlueSchemaRegistryName
           CONTACT_SCHEMA_NAME: !Ref ContactSchemaName
           JAVA_TOOL_OPTIONS: -XX:+TieredCompilation -XX:TieredStopAtLevel=1
       Policies:
@@ -191,10 +192,14 @@ Parameters:
     Type: String
     Description: Enter the name of the MSK Topic
     Default: KAFKA_TOPIC
+  GlueSchemaRegistryName:
+    Type: String
+    Description: Enter the name of the Glue Schema Registry
+    Default: GLUE_SCHEMA_REGISTRY_NAME
   ContactSchemaName:
     Type: String
     Description: Enter the name of the Contact Schema
-    Default: ContactSchema
+    Default: CONTACT_SCHEMA
   VpcId:
     Type: String
     Description: Enter the VPC ID where the MSK cluster is deployed
@@ -206,7 +211,7 @@ Parameters:
   SecurityGroupIds:
     Type: CommaDelimitedList
     Description: Enter the security group IDs that allow access to the MSK cluster (comma-separated)
-    Default: SECURITY_GROUP_IDS
+    Default: LAMBDA_SECURITY_GROUP_ID
 
 Outputs:
   MSKConsumerLambdaFunction: