Merge branch 'main' of https://github.com/vaibhavjainv/serverless-patterns

Author: indranil-banerjee-aws

msk-lambda-schema-avro-java-sam/README.md

@@ -160,6 +160,44 @@ The Key and Value are base64 encoded and have to be decoded. A message can also
 
 The code in this example prints out the fields in the Kafka message and also decrypts the key and the value and logs them in Cloudwatch logs.
 
+### Message Filtering with Event Source Mapping
+
+This sample application demonstrates how to use event source mapping filters with Amazon MSK and Lambda. The producer Lambda function generates contacts with zip codes that start with either "1000" or "2000" (with approximately 50% probability for each). However, the consumer Lambda function is configured to only process messages where the zip code starts with "1000".
+
+#### Filter Configuration
+
+The filter is configured in the SAM template using the `FilterCriteria` property of the MSK event source mapping:
+
+```yaml
+FilterCriteria:
+  Filters:
+    - Pattern: '{    "value": {        "zip": [ { "prefix": "1000" } ]      }}'
+```
+
+This filter pattern instructs the event source mapping to only send messages to the Lambda function if the message value contains a "zip" field that starts with "1000".
+
+#### Verifying the Filter Behavior
+
+To verify that the filter is working correctly, follow these steps:
+
+1. **Invoke the producer Lambda function** using one of the methods described above.
+
+2. **Check the producer function logs** in CloudWatch:
+   - Navigate to the CloudWatch Logs console
+   - Find the log group for the producer function (`/aws/lambda/msk-lambda-schema-avro-java-sam-LambdaMSKProducerJavaFunction-XXXXXXXXXXXX`)
+   - Open the most recent log stream
+   - Look for the "ZIP CODE DISTRIBUTION SUMMARY" section, which shows how many messages were generated with zip codes starting with "1000" and how many with "2000"
+   - You should see that the producer generated a mix of both zip code types
+
+3. **Check the consumer function logs** in CloudWatch:
+   - Navigate to the CloudWatch Logs console
+   - Find the log group for the consumer function (`/aws/lambda/msk-lambda-schema-avro-java-sam-LambdaMSKConsumerJavaFunction-XXXXXXXXXXXX`)
+   - Open the most recent log stream
+   - You should see that the consumer only processed messages with zip codes starting with "1000"
+   - Messages with zip codes starting with "2000" were filtered out by the event source mapping and never reached the Lambda function
+
+This demonstrates how event source mapping filters can be used to efficiently process only the messages that match specific criteria, reducing Lambda invocation costs and processing overhead.
+
 ## Cleanup
 
 You can first clean-up the Lambda function by running the sam delete command

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

@@ -41,6 +41,10 @@ public String handleRequest(KafkaEvent event, Context context) {
 
 		String response = new String("200 OK");
 		this.listOfMessages = new ArrayList<KafkaMessage>();
+		
+		// Counters for zip code patterns
+		int zip1000Count = 0;
+		int zip2000Count = 0;
 		//Incoming KafkaEvent object has a property called records that is a map
 		//Each key in the map is a combination of a topic and a partition
 		Map<String, List<KafkaEventRecord>> record=event.getRecords();
@@ -128,6 +132,33 @@ public String handleRequest(KafkaEvent event, Context context) {
 					    try {
 					        decodedValue = new String(decodedValueBytes);
 					        logger.log("Decoded value as string: " + (decodedValue.length() > 100 ? decodedValue.substring(0, 100) + "..." : decodedValue));
+					        
+					        // Add more detailed logging for AVRO messages
+					        logger.log("=== AVRO MESSAGE DETAILS ===");
+					        logger.log("Message appears to be AVRO-formatted. Attempting to extract fields:");
+					        
+					        // Try to extract some common fields from the AVRO binary data
+					        // This is a simple approach to show some readable content
+					        StringBuilder readableContent = new StringBuilder();
+					        for (int i = 0; i < decodedValueBytes.length; i++) {
+					            // Skip non-printable characters
+					            if (decodedValueBytes[i] >= 32 && decodedValueBytes[i] < 127) {
+					                readableContent.append((char)decodedValueBytes[i]);
+					            }
+					        }
+					        
+					        String readableString = readableContent.toString();
+					        logger.log("Readable content extracted from AVRO: " + readableString);
+					        
+					        // Check for zip code patterns
+					        if (readableString.contains("1000")) {
+					            logger.log("FOUND ZIP CODE STARTING WITH 1000");
+					        }
+					        if (readableString.contains("2000")) {
+					            logger.log("FOUND ZIP CODE STARTING WITH 2000");
+					        }
+					        
+					        logger.log("=== END AVRO MESSAGE DETAILS ===");
 					    } catch (Exception e) {
 					        logger.log("ERROR converting bytes to string: " + e.getMessage());
 					        decodedValue = "Error decoding: " + e.getMessage();
@@ -178,9 +209,39 @@ public String handleRequest(KafkaEvent event, Context context) {
 	            // as well as in Json format using gson.toJson function
 				logger.log("Received this message from Kafka - " + thisMessage.toString());
 				logger.log("Message in JSON format : " + gson.toJson(thisMessage));
+				
+				// Add a more readable summary of the message
+				logger.log("=== MESSAGE SUMMARY ===");
+				logger.log("Topic: " + thisMessage.getTopic());
+				logger.log("Partition: " + thisMessage.getPartition());
+				logger.log("Offset: " + thisMessage.getOffset());
+				logger.log("Key: " + thisMessage.getDecodedKey());
+				
+				// Check for zip code patterns in the decoded value
+				String decodedValueStr = thisMessage.getDecodedValue();
+				if (decodedValueStr != null) {
+				    if (decodedValueStr.contains("1000")) {
+				        logger.log("ZIP CODE: Found 1000 pattern in message");
+				        zip1000Count++;
+				    }
+				    if (decodedValueStr.contains("2000")) {
+				        logger.log("ZIP CODE: Found 2000 pattern in message");
+				        zip2000Count++;
+				    }
+				}
+				
+				logger.log("=== END MESSAGE SUMMARY ===");
 			}
 		}
 		logger.log("All Messages in this batch = " + gson.toJson(listOfMessages));
+		
+		// Log summary of zip code distribution
+		logger.log("========== ZIP CODE DISTRIBUTION SUMMARY ==========");
+		logger.log("Messages with zip code containing 1000: " + zip1000Count);
+		logger.log("Messages with zip code containing 2000: " + zip2000Count);
+		logger.log("Other messages: " + (listOfMessages.size() - zip1000Count - zip2000Count));
+		logger.log("====================================================");
+		
 		logger.log("========== LAMBDA FUNCTION COMPLETED ==========");
 		return response;
 	}

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

@@ -29,6 +29,11 @@ public String handleRequest(Map<String, Object> event, Context context) {
         LambdaLogger logger = context.getLogger();
         logger.log("Received event: " + gson.toJson(event));
 
+        // Initialize counters for zip code distribution
+        int messageCount = 10;
+        int zip1000Count = 0;
+        int zip2000Count = 0;
+
         try {
             // Get environment variables
             String mskClusterArn = System.getenv("MSK_CLUSTER_ARN");
@@ -42,6 +47,9 @@ public String handleRequest(Map<String, Object> event, Context context) {
                 throw new RuntimeException("Required environment variables not set: MSK_CLUSTER_ARN, KAFKA_TOPIC, CONTACT_SCHEMA_NAME");
             }
 
+            // Log that we're generating zip codes with different prefixes
+            logger.log("Generating contacts with zip codes starting with 1000 (50% chance) or 2000 (50% chance)");
+            
             // Create a Contact object from the input event or use default values
             Contact contact = createContactFromEvent(event);
             logger.log("Created contact: " + gson.toJson(contact));
@@ -73,7 +81,6 @@ public String handleRequest(Map<String, Object> event, Context context) {
                 logger.log("Schema definition: " + schemaDefinition);
 
                 // Send 10 messages
-                int messageCount = 10;
                 logger.log("Sending " + messageCount + " AVRO messages to topic: " + kafkaTopic);
 
                 for (int i = 0; i < messageCount; i++) {
@@ -90,13 +97,34 @@ public String handleRequest(Map<String, Object> event, Context context) {
                     logger.log("Sending contact #" + (i+1) + ": " + gson.toJson(messageContact));
                     logger.log("AVRO record #" + (i+1) + ": " + messageAvroRecord.toString());
 
+                    // Log the zip code prefix for distribution tracking
+                    String zipCode = messageContact.getZip();
+                    if (zipCode != null && zipCode.length() >= 4) {
+                        String prefix = zipCode.substring(0, 4);
+                        logger.log("Contact #" + (i+1) + " zip code prefix: " + prefix);
+                        
+                        // Count zip codes by prefix
+                        if ("1000".equals(prefix)) {
+                            zip1000Count++;
+                        } else if ("2000".equals(prefix)) {
+                            zip2000Count++;
+                        }
+                    }
+                    
                     // Send the message
                     KafkaProducerHelper.sendAvroMessage(producer, kafkaTopic, messageKey, messageAvroRecord);
                     logger.log("Successfully sent AVRO message #" + (i+1) + " to topic: " + kafkaTopic);
                 }
+                
+                // Log summary of zip code distribution
+                logger.log("ZIP CODE DISTRIBUTION SUMMARY:");
+                logger.log("Messages with zip code starting with 1000: " + zip1000Count);
+                logger.log("Messages with zip code starting with 2000: " + zip2000Count);
+                logger.log("Other zip code formats: " + (messageCount - zip1000Count - zip2000Count));
             }
 
-            return "Successfully sent 10 AVRO messages to Kafka topic: " + kafkaTopic;
+            return "Successfully sent " + messageCount + " AVRO messages to Kafka topic: " + kafkaTopic + 
+                   " (Zip codes: " + zip1000Count + " with prefix 1000, " + zip2000Count + " with prefix 2000)";
         } catch (Exception e) {
             logger.log("Error sending AVRO message: " + e.getMessage());
             e.printStackTrace();
@@ -121,7 +149,17 @@ private Contact createContactFromEvent(Map<String, Object> event) {
         contact.setCity(getStringValue(event, "city", "AnyCity"));
         contact.setCounty(getStringValue(event, "county", "AnyCounty"));
         contact.setState(getStringValue(event, "state", "AnyState"));
-        contact.setZip(getStringValue(event, "zip", "1000" + randomDigit()));
+        
+        // Generate zip code starting with 1000 50% of the time and 2000 the other 50%
+        if (event.containsKey("zip") && event.get("zip") != null) {
+            // If zip is provided in the event, use it as is
+            contact.setZip(event.get("zip").toString());
+        } else {
+            // 50% chance for each prefix
+            String prefix = Math.random() < 0.5 ? "1000" : "2000";
+            contact.setZip(prefix + randomDigit());
+        }
+        
         contact.setHomePhone(getStringValue(event, "homePhone", "555-123-" + randomDigits(4)));
         contact.setCellPhone(getStringValue(event, "cellPhone", "555-456-" + randomDigits(4)));
         contact.setEmail(getStringValue(event, "email", "user-" + randomSuffix() + "@example.com"));

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

@@ -50,6 +50,9 @@ Resources:
             DestinationConfig:
               OnFailure:
                 Destination: !GetAtt ConsumerDLQ.Arn
+            FilterCriteria:
+              Filters:
+                - Pattern: '{    "value": {        "zip": [ { "prefix": "2000" } ]      }}'
       Policies:
       - Statement:
         - Sid: KafkaClusterPermissionsPolicy
@@ -198,12 +201,15 @@ Parameters:
   VpcId:
     Type: String
     Description: Enter the VPC ID where the MSK cluster is deployed
+    Default: VPC_ID
   SubnetIds:
     Type: CommaDelimitedList
     Description: Enter the subnet IDs where the MSK cluster is deployed (comma-separated)
+    Default: SUBNET_IDS
   SecurityGroupIds:
     Type: CommaDelimitedList
     Description: Enter the security group IDs that allow access to the MSK cluster (comma-separated)
+    Default: SECURITY_GROUP_IDS
 
 Outputs:
   MSKConsumerLambdaFunction:

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

@@ -50,6 +50,9 @@ Resources:
             DestinationConfig:
               OnFailure:
                 Destination: !GetAtt ConsumerDLQ.Arn
+            FilterCriteria:
+              Filters:
+                - Pattern: '{    "value": {        "zip": [ { "prefix": "2000" } ]      }}'
       Policies:
       - Statement:
         - Sid: KafkaClusterPermissionsPolicy