docs(kafka): provide example payloads (#4094)

Author: dreamorosi

docs/features/kafka.md

@@ -69,22 +69,60 @@ Depending on the schema types you want to use, install the library and the corre
 
 Additionally, if you want to use output parsing with [Standard Schema](https://github.com/standard-schema/standard-schema), you can install [any of the supported libraries](https://standardschema.dev/#what-schema-libraries-implement-the-spec), for example: Zod, Valibot, or ArkType.
 
-<!-- ### Required resources
+### Required resources
 
 To use the Kafka consumer utility, you need an AWS Lambda function configured with a Kafka event source. This can be Amazon MSK, MSK Serverless, or a self-hosted Kafka cluster.
 
 === "gettingStartedWithMsk.yaml"
 
     ```yaml
     --8<-- "examples/snippets/kafka/templates/gettingStartedWithMsk.yaml"
-    ``` -->
+    ```
 
 ### Using ESM with Schema Registry
 
 The Event Source Mapping configuration determines which mode is used. With `JSON`, Lambda converts all messages to JSON before invoking your function. With `SOURCE` mode, Lambda preserves the original format, requiring you function to handle the appropriate deserialization.
 
 Powertools for AWS supports both Schema Registry integration modes in your Event Source Mapping configuration.
 
+For simplicity, we will use a simple schema containing `name` and `age` in most of our examples. You can also copy the payload example with the expected Kafka event to test your code.
+
+=== "JSON"
+
+    ```json
+    --8<-- "examples/snippets/kafka/samples/user.json"
+    ```
+
+=== "Payload JSON"
+
+    ```json
+    --8<-- "examples/snippets/kafka/samples/kafkaEventJson.json"
+    ```
+
+=== "Avro Schema"
+
+    ```json
+    --8<-- "examples/snippets/kafka/samples/user.avsc"
+    ```
+
+=== "Payload Avro"
+
+    ```json
+    --8<-- "examples/snippets/kafka/samples/kafkaEventAvro.json"
+    ```
+
+=== "Protobuf Schema"
+
+    ```typescript
+    --8<-- "examples/snippets/kafka/samples/user.proto"
+    ```
+
+=== "Payload Protobuf"
+
+    ```json
+    --8<-- "examples/snippets/kafka/samples/kafkaEventProtobuf.json"
+    ```
+
 ### Processing Kafka events
 
 The Kafka consumer utility transforms raw Kafka events into an intuitive format for processing. To handle messages effectively, you'll need to configure a schema that matches your data format.
@@ -110,9 +148,9 @@ The Kafka consumer utility transforms raw Kafka events into an intuitive format
     --8<-- "examples/snippets/kafka/gettingStartedJson.ts"
     ```
 
-### Deserializing keys and values
+### Deserializing key and value
 
-The `kafkaConsumer` function can deserialize both keys and values independently based on your schema configuration. This flexibility allows you to work with different data formats in the same message.
+The `kafkaConsumer` function can deserialize both key and value independently based on your schema configuration. This flexibility allows you to work with different data formats in the same message.
 
 === "index.ts"
 

examples/snippets/kafka/advancedWorkingWithIdempotency.ts

@@ -6,7 +6,7 @@ import { DynamoDBPersistenceLayer } from '@aws-lambda-powertools/idempotency/dyn
 import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
 import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
 import { Logger } from '@aws-lambda-powertools/logger';
-import { User } from './samples/user.es6.generated.js'; // protobuf generated class
+import { com } from './samples/user.generated.js'; // protobuf generated class
 
 const logger = new Logger({ serviceName: 'kafka-consumer' });
 const persistenceStore = new DynamoDBPersistenceLayer({
@@ -16,14 +16,14 @@ const persistenceStore = new DynamoDBPersistenceLayer({
 const schemaConfig = {
   value: {
     type: SchemaType.PROTOBUF,
-    schema: User,
+    schema: com.example.User,
   },
 } satisfies SchemaConfig;
 
 const processRecord = makeIdempotent(
   async (user, topic, partition, offset) => {
     logger.info('processing user', {
-      userId: user.id,
+      user,
       meta: {
         topic,
         partition,
@@ -35,7 +35,7 @@ const processRecord = makeIdempotent(
 
     return {
       success: true,
-      userId: user.id,
+      user,
     };
   },
   {

examples/snippets/kafka/advancedWorkingWithRecordMetadata.ts

@@ -1,10 +1,10 @@
 import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
 import { Logger } from '@aws-lambda-powertools/logger';
-import { type IUser, User } from './samples/user.es6.generated.js'; // protobuf generated class
+import { com } from './samples/user.generated.js'; // protobuf generated class
 
 const logger = new Logger({ serviceName: 'kafka-consumer' });
 
-export const handler = kafkaConsumer<unknown, IUser>(
+export const handler = kafkaConsumer<unknown, com.example.IUser>(
   async (event, _context) => {
     for (const record of event.records) {
       const { value, topic, partition, offset, timestamp, headers } = record;
@@ -24,15 +24,15 @@ export const handler = kafkaConsumer<unknown, IUser>(
 
       // Process the deserialized value
       logger.info('User data', {
-        userId: value.id,
         userName: value.name,
+        userAge: value.age,
       });
     }
   },
   {
     value: {
       type: SchemaType.PROTOBUF,
-      schema: User,
+      schema: com.example.User,
     },
   }
 );

examples/snippets/kafka/gettingStartedPrimitiveValues.ts

@@ -3,7 +3,7 @@ import { Logger } from '@aws-lambda-powertools/logger';
 
 const logger = new Logger({ serviceName: 'kafka-consumer' });
 
-export const handler = kafkaConsumer<string, { id: number; name: string }>(
+export const handler = kafkaConsumer<string, { name: string; age: number }>(
   async (event, _context) => {
     for (const record of event.records) {
       // Key is automatically decoded as UTF-8 string
@@ -14,8 +14,8 @@ export const handler = kafkaConsumer<string, { id: number; name: string }>(
       logger.info('received value', {
         key,
         product: {
-          id: value.id,
           name: value.name,
+          age: value.age,
         },
       });
     }

examples/snippets/kafka/gettingStartedProtobuf.ts

@@ -1,14 +1,14 @@
 import { SchemaType, kafkaConsumer } from '@aws-lambda-powertools/kafka';
 import type { SchemaConfig } from '@aws-lambda-powertools/kafka/types';
 import { Logger } from '@aws-lambda-powertools/logger';
-import { User } from './samples/user.es6.generated.js'; // protobuf generated class
+import { com } from './samples/user.generated.js'; // protobuf generated class
 
 const logger = new Logger({ serviceName: 'kafka-consumer' });
 
 const schemaConfig = {
   value: {
     type: SchemaType.PROTOBUF,
-    schema: User,
+    schema: com.example.User,
   },
 } satisfies SchemaConfig;
 

examples/snippets/kafka/gettingStartedValueOnly.ts

@@ -0,0 +1,19 @@
+{
+  "eventSource": "aws:kafka",
+  "eventSourceArn": "arn:aws:kafka:eu-west-3:123456789012:cluster/powertools-kafka-esm/f138df86-9253-4d2a-b682-19e132396d4f-s3",
+  "bootstrapServers": "boot-z3majaui.c3.kafka-serverless.eu-west-3.amazonaws.com:9098",
+  "records": {
+    "python-with-avro-doc-3": [
+      {
+        "topic": "python-with-avro-doc",
+        "partition": 3,
+        "offset": 0,
+        "timestamp": 1750547105187,
+        "timestampType": "CREATE_TIME",
+        "key": "MTIz",
+        "value": "AwBXT2qalUhN6oaj2CwEeaEWFFBvd2VydG9vbHMK",
+        "headers": []
+      }
+    ]
+  }
+}

examples/snippets/kafka/samples/kafkaEventAvro.json

@@ -0,0 +1,19 @@
+{
+  "eventSource": "aws:kafka",
+  "eventSourceArn": "arn:aws:kafka:eu-west-3:123456789012:cluster/powertools-kafka-esm/f138df86-9253-4d2a-b682-19e132396d4f-s3",
+  "bootstrapServers": "boot-z3majaui.c3.kafka-serverless.eu-west-3.amazonaws.com:9098",
+  "records": {
+    "python-with-avro-doc-5": [
+      {
+        "topic": "python-with-avro-doc",
+        "partition": 5,
+        "offset": 0,
+        "timestamp": 1750547462087,
+        "timestampType": "CREATE_TIME",
+        "key": "MTIz",
+        "value": "eyJuYW1lIjogIlBvd2VydG9vbHMiLCAiYWdlIjogNX0=",
+        "headers": []
+      }
+    ]
+  }
+}

examples/snippets/kafka/samples/kafkaEventJson.json

@@ -0,0 +1,19 @@
+{
+  "eventSource": "aws:kafka",
+  "eventSourceArn": "arn:aws:kafka:eu-west-3:992382490249:cluster/powertools-kafka-esm/f138df86-9253-4d2a-b682-19e132396d4f-s3",
+  "bootstrapServers": "boot-z3majaui.c3.kafka-serverless.eu-west-3.amazonaws.com:9098",
+  "records": {
+    "python-with-avro-doc-5": [
+      {
+        "topic": "python-with-avro-doc",
+        "partition": 5,
+        "offset": 1,
+        "timestamp": 1750624373324,
+        "timestampType": "CREATE_TIME",
+        "key": "MTIz",
+        "value": "Cgpwb3dlcnRvb2xzEAU=",
+        "headers": []
+      }
+    ]
+  }
+}

examples/snippets/kafka/samples/kafkaEventProtobuf.json

@@ -0,0 +1,9 @@
+{
+    "type": "record",
+    "name": "User",
+    "namespace": "com.example",
+    "fields": [
+        {"name": "name", "type": "string"},
+        {"name": "age", "type": "int"}
+    ]
+}