As Kafka producer and consumer apps are decoupled, they operate on Kafka topics rather than communicating with each other directly. As described in the Confluent document, Schema Registry provides a centralized repository for managing and validating schemas for topic message data, and for serialization and deserialization of the data over the network. Producers and consumers to Kafka topics can use schemas to ensure data consistency and compatibility as schemas evolve. In AWS, the Glue Schema Registry supports features to manage and enforce schemas on data streaming applications using convenient integrations with Apache Kafka, Amazon Managed Streaming for Apache Kafka, Amazon Kinesis Data Streams, Amazon Kinesis Data Analytics for Apache Flink, and AWS Lambda. In this post, we will discuss how to integrate Python Kafka producer and consumer apps In AWS Lambda with the Glue Schema Registry.
Architecture
Fake online order data is generated by multiple Lambda functions that are invoked by an EventBridge schedule rule. The schedule is set to run every minute and the associating rule has a configurable number (e.g. 5) of targets. Each target points to the same Kafka producer Lambda function. In this way we are able to generate test data using multiple Lambda functions according to the desired volume of messages. Note that, before sending a message, the producer validates the schema and registers a new one if it is not registered yet. Then it serializes the message and sends it to the cluster.
Once messages are sent to a Kafka topic, they can be consumed by Lambda where Amazon MSK is configured as an event source. The serialized record (message key or value) includes the schema ID so that the consumer can request the schema from the schema registry (if not cached) in order to deserialize it.
The infrastructure is built by Terraform and the AWS SAM CLI is used to develop the producer Lambda function locally before deploying to AWS.
Infrastructure
A VPC with 3 public and private subnets is created using the AWS VPC Terraform module (vpc.tf). Also, a SoftEther VPN server is deployed in order to access the resources in the private subnets from the developer machine (vpn.tf). It is particularly useful to monitor and manage the MSK cluster and Kafka topic as well as developing the Kafka producer Lambda function locally. The details about how to configure the VPN server can be found in an earlier post. The source can be found in the GitHub repository of this post.
MSK
An MSK cluster with 3 brokers is created. The broker nodes are deployed with the kafka.m5.large instance type in private subnets and IAM authentication is used for the client authentication method. Finally, additional server configurations are added such as enabling auto creation of topics and topic deletion.
1# glue-schema-registry/infra/variable.tf
2locals {
3 ...
4 msk = {
5 version = "3.3.1"
6 instance_size = "kafka.m5.large"
7 ebs_volume_size = 20
8 log_retention_ms = 604800000 # 7 days
9 }
10 ...
11}
12# glue-schema-registry/infra/msk.tf
13resource "aws_msk_cluster" "msk_data_cluster" {
14 cluster_name = "${local.name}-msk-cluster"
15 kafka_version = local.msk.version
16 number_of_broker_nodes = length(module.vpc.private_subnets)
17 configuration_info {
18 arn = aws_msk_configuration.msk_config.arn
19 revision = aws_msk_configuration.msk_config.latest_revision
20 }
21
22 broker_node_group_info {
23 instance_type = local.msk.instance_size
24 client_subnets = module.vpc.private_subnets
25 security_groups = [aws_security_group.msk.id]
26 storage_info {
27 ebs_storage_info {
28 volume_size = local.msk.ebs_volume_size
29 }
30 }
31 }
32
33 client_authentication {
34 sasl {
35 iam = true
36 }
37 }
38
39 logging_info {
40 broker_logs {
41 cloudwatch_logs {
42 enabled = true
43 log_group = aws_cloudwatch_log_group.msk_cluster_lg.name
44 }
45 s3 {
46 enabled = true
47 bucket = aws_s3_bucket.default_bucket.id
48 prefix = "logs/msk/cluster-"
49 }
50 }
51 }
52
53 tags = local.tags
54
55 depends_on = [aws_msk_configuration.msk_config]
56}
57
58resource "aws_msk_configuration" "msk_config" {
59 name = "${local.name}-msk-configuration"
60
61 kafka_versions = [local.msk.version]
62
63 server_properties = <<PROPERTIES
64 auto.create.topics.enable = true
65 delete.topic.enable = true
66 log.retention.ms = ${local.msk.log_retention_ms}
67 PROPERTIES
68}
Security Groups
Two security groups are created - one for the MSK cluster and the other for the Lambda apps.
The inbound/outbound rules of the former are created for accessing the cluster by
- Event Source Mapping (ESM) for Lambda
- This is for the Lambda consumer that subscribes the MSK cluster. As described in the AWS re:Post doc, when a Lambda function is configured with an Amazon MSK trigger or a self-managed Kafka trigger, an ESM resource is automatically created. An ESM is separate from the Lambda function, and it continuously polls records from the topic in the Kafka cluster. The ESM bundles those records into a payload. Then, it calls the Lambda Invoke API to deliver the payload to your Lambda function for processing. Note it doesn’t inherit the VPC network settings of the Lambda function but uses the subnet and security group settings that are configured on the target MSK cluster. Therefore, the MSK cluster’s security group must include a rule that grants ingress traffic from itself and egress traffic to itself. For us, the rules on port 9098 as the cluster only supports IAM authentication. Also, an additional egress rule is created to access the Glue Schema Registry.
- Other Resources
- Two ingress rules are created for the VPN server and Lambda. The latter is only for the Lambda producer because the consumer doesn’t rely on the Lambda network setting.
The second security group is created here, while the Lambda function is created in a different Terraform stack. This is for ease of adding it to the inbound rule of the MSK’s security group. Later we will discuss how to make use of it with the Lambda function. The outbound rule allows all outbound traffic although only port 9098 for the MSK cluster and 443 for the Glue Schema Registry would be sufficient.
1# glue-schema-registry/infra/msk.tf
2resource "aws_security_group" "msk" {
3 name = "${local.name}-msk-sg"
4 vpc_id = module.vpc.vpc_id
5
6 lifecycle {
7 create_before_destroy = true
8 }
9
10 tags = local.tags
11}
12
13# for lambda event source mapping
14resource "aws_security_group_rule" "msk_ingress_self_broker" {
15 type = "ingress"
16 description = "msk ingress self"
17 security_group_id = aws_security_group.msk.id
18 protocol = "tcp"
19 from_port = 9098
20 to_port = 9098
21 source_security_group_id = aws_security_group.msk.id
22}
23
24resource "aws_security_group_rule" "msk_egress_self_broker" {
25 type = "egress"
26 description = "msk egress self"
27 security_group_id = aws_security_group.msk.id
28 protocol = "tcp"
29 from_port = 9098
30 to_port = 9098
31 source_security_group_id = aws_security_group.msk.id
32}
33
34resource "aws_security_group_rule" "msk_all_outbound" {
35 type = "egress"
36 description = "allow outbound all"
37 security_group_id = aws_security_group.msk.id
38 protocol = "-1"
39 from_port = "0"
40 to_port = "0"
41 cidr_blocks = ["0.0.0.0/0"]
42}
43
44# for other resources
45resource "aws_security_group_rule" "msk_vpn_inbound" {
46 count = local.vpn.to_create ? 1 : 0
47 type = "ingress"
48 description = "VPN access"
49 security_group_id = aws_security_group.msk.id
50 protocol = "tcp"
51 from_port = 9098
52 to_port = 9098
53 source_security_group_id = aws_security_group.vpn[0].id
54}
55
56resource "aws_security_group_rule" "msk_lambda_inbound" {
57 type = "ingress"
58 description = "lambda access"
59 security_group_id = aws_security_group.msk.id
60 protocol = "tcp"
61 from_port = 9098
62 to_port = 9098
63 source_security_group_id = aws_security_group.kafka_app_lambda.id
64}
65
66...
67
68# lambda security group
69resource "aws_security_group" "kafka_app_lambda" {
70 name = "${local.name}-lambda-msk-access"
71 vpc_id = module.vpc.vpc_id
72
73 lifecycle {
74 create_before_destroy = true
75 }
76
77 tags = local.tags
78}
79
80resource "aws_security_group_rule" "kafka_app_lambda_msk_egress" {
81 type = "egress"
82 description = "allow outbound all"
83 security_group_id = aws_security_group.kafka_app_lambda.id
84 protocol = "-1"
85 from_port = 0
86 to_port = 0
87 cidr_blocks = ["0.0.0.0/0"]
88}
Kafka Apps
The resources related to the Kafka producer and consumer Lambda functions are managed in a separate Terraform stack. This is because it is easier to build the relevant resources iteratively. Note the SAM CLI builds the whole Terraform stack even for a small change of code, and it wouldn’t be convenient if the entire resources are managed in the same stack.
Producer App
Order Data
Fake order data is generated using the Faker package and the dataclasses_avroschema package is used to automatically generate the Avro schema according to its attributes. A mixin class called InjectCompatMixin is injected into the Order class, which specifies a schema compatibility mode into the generated schema. The auto() class method is used to instantiate the class automatically. Finally, the OrderMore class is created for the schema evolution demo, which will be discussed later.
1# glue-schema-registry/app/producer/src/order.py
2import datetime
3import string
4import json
5import typing
6import dataclasses
7import enum
8
9from faker import Faker
10from dataclasses_avroschema import AvroModel
11
12
13class Compatibility(enum.Enum):
14 NONE = "NONE"
15 DISABLED = "DISABLED"
16 BACKWARD = "BACKWARD"
17 BACKWARD_ALL = "BACKWARD_ALL"
18 FORWARD = "FORWARD"
19 FORWARD_ALL = "FORWARD_ALL"
20 FULL = "FULL"
21 FULL_ALL = "FULL_ALL"
22
23
24class InjectCompatMixin:
25 @classmethod
26 def updated_avro_schema_to_python(cls, compat: Compatibility = Compatibility.BACKWARD):
27 schema = cls.avro_schema_to_python()
28 schema["compatibility"] = compat.value
29 return schema
30
31 @classmethod
32 def updated_avro_schema(cls, compat: Compatibility = Compatibility.BACKWARD):
33 schema = cls.updated_avro_schema_to_python(compat)
34 return json.dumps(schema)
35
36
37@dataclasses.dataclass
38class OrderItem(AvroModel):
39 product_id: int
40 quantity: int
41
42
43@dataclasses.dataclass
44class Order(AvroModel, InjectCompatMixin):
45 "Online fake order item"
46 order_id: str
47 ordered_at: datetime.datetime
48 user_id: str
49 order_items: typing.List[OrderItem]
50
51 class Meta:
52 namespace = "Order V1"
53
54 def asdict(self):
55 return dataclasses.asdict(self)
56
57 @classmethod
58 def auto(cls, fake: Faker = Faker()):
59 rand_int = fake.random_int(1, 1000)
60 user_id = "".join(
61 [string.ascii_lowercase[int(s)] if s.isdigit() else s for s in hex(rand_int)]
62 )[::-1]
63 order_items = [
64 OrderItem(fake.random_int(1, 9999), fake.random_int(1, 10))
65 for _ in range(fake.random_int(1, 4))
66 ]
67 return cls(fake.uuid4(), datetime.datetime.utcnow(), user_id, order_items)
68
69 def create(self, num: int):
70 return [self.auto() for _ in range(num)]
71
72
73@dataclasses.dataclass
74class OrderMore(Order):
75 is_prime: bool
76
77 @classmethod
78 def auto(cls, fake: Faker = Faker()):
79 o = Order.auto()
80 return cls(o.order_id, o.ordered_at, o.user_id, o.order_items, fake.pybool())
The generated schema of the Order class can be found below.
1{
2 "doc": "Online fake order item",
3 "namespace": "Order V1",
4 "name": "Order",
5 "compatibility": "BACKWARD",
6 "type": "record",
7 "fields": [
8 {
9 "name": "order_id",
10 "type": "string"
11 },
12 {
13 "name": "ordered_at",
14 "type": {
15 "type": "long",
16 "logicalType": "timestamp-millis"
17 }
18 },
19 {
20 "name": "user_id",
21 "type": "string"
22 },
23 {
24 "name": "order_items",
25 "type": {
26 "type": "array",
27 "items": {
28 "type": "record",
29 "name": "OrderItem",
30 "fields": [
31 {
32 "name": "product_id",
33 "type": "long"
34 },
35 {
36 "name": "quantity",
37 "type": "long"
38 }
39 ]
40 },
41 "name": "order_item"
42 }
43 }
44 ]
45}
Below shows an example order record.
1{
2 "order_id": "53263c42-81b3-4a53-8067-fcdb44fa5479",
3 "ordered_at": 1680745813045,
4 "user_id": "dicxa",
5 "order_items": [
6 {
7 "product_id": 5947,
8 "quantity": 8
9 }
10 ]
11}
Producer
The aws-glue-schema-registry package is used serialize the value of order messages. It provides the KafkaSerializer class that validates, registers and serializes the relevant records. It supports Json and Avro schemas, and we can add it to the value_serializer argument of the KafkaProducer class. By default, the schemas are named as <topic>-key and <topic>-value and it can be changed by updating the schema_naming_strategy argument. Note that, when sending a message, the value should be a tuple of data and schema. Note also that the stable version of the kafka-python package does not support the IAM authentication method. Therefore, we need to install the package from a forked repository as discussed in this GitHub issue.
1# glue-schema-registry/app/producer/src/producer.py
2import os
3import datetime
4import json
5import typing
6
7import boto3
8import botocore.exceptions
9from kafka import KafkaProducer
10from aws_schema_registry import SchemaRegistryClient
11from aws_schema_registry.avro import AvroSchema
12from aws_schema_registry.adapter.kafka import KafkaSerializer
13from aws_schema_registry.exception import SchemaRegistryException
14
15from .order import Order
16
17
18class Producer:
19 def __init__(self, bootstrap_servers: list, topic: str, registry: str, is_local: bool = False):
20 self.bootstrap_servers = bootstrap_servers
21 self.topic = topic
22 self.registry = registry
23 self.glue_client = boto3.client(
24 "glue", region_name=os.getenv("AWS_DEFAULT_REGION", "ap-southeast-2")
25 )
26 self.is_local = is_local
27 self.producer = self.create()
28
29 @property
30 def serializer(self):
31 client = SchemaRegistryClient(self.glue_client, registry_name=self.registry)
32 return KafkaSerializer(client)
33
34 def create(self):
35 params = {
36 "bootstrap_servers": self.bootstrap_servers,
37 "key_serializer": lambda v: json.dumps(v, default=self.serialize).encode("utf-8"),
38 "value_serializer": self.serializer,
39 }
40 if not self.is_local:
41 params = {
42 **params,
43 **{"security_protocol": "SASL_SSL", "sasl_mechanism": "AWS_MSK_IAM"},
44 }
45 return KafkaProducer(**params)
46
47 def send(self, orders: typing.List[Order], schema: AvroSchema):
48 if not self.check_registry():
49 print(f"registry not found, create {self.registry}")
50 self.create_registry()
51
52 for order in orders:
53 data = order.asdict()
54 try:
55 self.producer.send(
56 self.topic, key={"order_id": data["order_id"]}, value=(data, schema)
57 )
58 except SchemaRegistryException as e:
59 raise RuntimeError("fails to send a message") from e
60 self.producer.flush()
61
62 def serialize(self, obj):
63 if isinstance(obj, datetime.datetime):
64 return obj.isoformat()
65 if isinstance(obj, datetime.date):
66 return str(obj)
67 return obj
68
69 def check_registry(self):
70 try:
71 self.glue_client.get_registry(RegistryId={"RegistryName": self.registry})
72 return True
73 except botocore.exceptions.ClientError as e:
74 if e.response["Error"]["Code"] == "EntityNotFoundException":
75 return False
76 else:
77 raise e
78
79 def create_registry(self):
80 try:
81 self.glue_client.create_registry(RegistryName=self.registry)
82 return True
83 except botocore.exceptions.ClientError as e:
84 if e.response["Error"]["Code"] == "AlreadyExistsException":
85 return True
86 else:
87 raise e
Lambda Handler
The Lambda function sends 100 records at a time followed by sleeping for 1 second. It repeats until it reaches MAX_RUN_SEC (e.g. 60) environment variable value. The last conditional block is for demonstrating a schema evolution example, which will be discussed later.
1# glue-schema-registry/app/producer/lambda_handler.py
2import os
3import datetime
4import time
5
6from aws_schema_registry.avro import AvroSchema
7
8from src.order import Order, OrderMore, Compatibility
9from src.producer import Producer
10
11
12def lambda_function(event, context):
13 producer = Producer(
14 bootstrap_servers=os.environ["BOOTSTRAP_SERVERS"].split(","),
15 topic=os.environ["TOPIC_NAME"],
16 registry=os.environ["REGISTRY_NAME"],
17 )
18 s = datetime.datetime.now()
19 ttl_rec = 0
20 while True:
21 orders = Order.auto().create(100)
22 schema = AvroSchema(Order.updated_avro_schema(Compatibility.BACKWARD))
23 producer.send(orders, schema)
24 ttl_rec += len(orders)
25 print(f"sent {len(orders)} messages")
26 elapsed_sec = (datetime.datetime.now() - s).seconds
27 if elapsed_sec > int(os.getenv("MAX_RUN_SEC", "60")):
28 print(f"{ttl_rec} records are sent in {elapsed_sec} seconds ...")
29 break
30 time.sleep(1)
31
32
33if __name__ == "__main__":
34 producer = Producer(
35 bootstrap_servers=os.environ["BOOTSTRAP_SERVERS"].split(","),
36 topic=os.environ["TOPIC_NAME"],
37 registry=os.environ["REGISTRY_NAME"],
38 is_local=True,
39 )
40 use_more = os.getenv("USE_MORE") is not None
41 if not use_more:
42 orders = Order.auto().create(1)
43 schema = AvroSchema(Order.updated_avro_schema(Compatibility.BACKWARD))
44 else:
45 orders = OrderMore.auto().create(1)
46 schema = AvroSchema(OrderMore.updated_avro_schema(Compatibility.BACKWARD))
47 print(orders)
48 producer.send(orders, schema)
Lambda Resource
The VPC, subnets, Lambda security group and MSK cluster are created in the infra Terraform stack, and they need to be obtained from the Kafka app stack. It can be achieved using the Terraform data sources as shown below. Note that the private subnets can be filtered by a specific tag (Tier: Private), which is added when creating them.
1# glue-schema-registry/infra/vpc.tf
2module "vpc" {
3 source = "terraform-aws-modules/vpc/aws"
4 version = "~> 3.14"
5
6 name = "${local.name}-vpc"
7 cidr = local.vpc.cidr
8
9 azs = local.vpc.azs
10 public_subnets = [for k, v in local.vpc.azs : cidrsubnet(local.vpc.cidr, 3, k)]
11 private_subnets = [for k, v in local.vpc.azs : cidrsubnet(local.vpc.cidr, 3, k + 3)]
12
13 ...
14
15 private_subnet_tags = {
16 "Tier" = "Private"
17 }
18
19 tags = local.tags
20}
21
22# glue-schema-registry/app/variables.tf
23data "aws_caller_identity" "current" {}
24
25data "aws_region" "current" {}
26
27data "aws_vpc" "selected" {
28 filter {
29 name = "tag:Name"
30 values = ["${local.infra_prefix}"]
31 }
32}
33
34data "aws_subnets" "private" {
35 filter {
36 name = "vpc-id"
37 values = [data.aws_vpc.selected.id]
38 }
39
40 tags = {
41 Tier = "Private"
42 }
43}
44
45data "aws_msk_cluster" "msk_data_cluster" {
46 cluster_name = "${local.infra_prefix}-msk-cluster"
47}
48
49data "aws_security_group" "kafka_producer_lambda" {
50 name = "${local.infra_prefix}-lambda-msk-access"
51}
52
53
54locals {
55 ...
56 infra_prefix = "glue-schema-registry"
57 ...
58}
The AWS Lambda Terraform module is used to create the producer Lambda function. Note that, in order to develop a Lambda function using AWS SAM, we need to create SAM metadata resource, which provides the AWS SAM CLI with the information it needs to locate Lambda functions and layers, along with their source code, build dependencies, and build logic from within your Terraform project. It is created by default by the Terraform module, which is convenient. Also, we need to give permission to the EventBridge rule to invoke the Lambda function, and it is given by the aws_lambda_permission resource.
1# glue-schema-registry/app/variables.tf
2locals {
3 name = local.infra_prefix
4 region = data.aws_region.current.name
5 environment = "dev"
6
7 infra_prefix = "glue-schema-registry"
8
9 producer = {
10 src_path = "producer"
11 function_name = "kafka_producer"
12 handler = "lambda_handler.lambda_function"
13 concurrency = 5
14 timeout = 90
15 memory_size = 128
16 runtime = "python3.8"
17 schedule_rate = "rate(1 minute)"
18 to_enable_trigger = true
19 environment = {
20 topic_name = "orders"
21 registry_name = "customer"
22 max_run_sec = 60
23 }
24 }
25
26 ...
27}
28
29# glue-schema-registry/app/main.tf
30module "kafka_producer_lambda" {
31 source = "terraform-aws-modules/lambda/aws"
32
33 function_name = local.producer.function_name
34 handler = local.producer.handler
35 runtime = local.producer.runtime
36 timeout = local.producer.timeout
37 memory_size = local.producer.memory_size
38 source_path = local.producer.src_path
39 vpc_subnet_ids = data.aws_subnets.private.ids
40 vpc_security_group_ids = [data.aws_security_group.kafka_app_lambda.id]
41 attach_network_policy = true
42 attach_policies = true
43 policies = [aws_iam_policy.msk_lambda_producer_permission.arn]
44 number_of_policies = 1
45 environment_variables = {
46 BOOTSTRAP_SERVERS = data.aws_msk_cluster.msk_data_cluster.bootstrap_brokers_sasl_iam
47 TOPIC_NAME = local.producer.environment.topic_name
48 REGISTRY_NAME = local.producer.environment.registry_name
49 MAX_RUN_SEC = local.producer.environment.max_run_sec
50 }
51
52 tags = local.tags
53}
54
55resource "aws_lambda_function_event_invoke_config" "kafka_producer_lambda" {
56 function_name = module.kafka_producer_lambda.lambda_function_name
57 maximum_retry_attempts = 0
58}
59
60resource "aws_lambda_permission" "allow_eventbridge" {
61 count = local.producer.to_enable_trigger ? 1 : 0
62 statement_id = "InvokeLambdaFunction"
63 action = "lambda:InvokeFunction"
64 function_name = local.producer.function_name
65 principal = "events.amazonaws.com"
66 source_arn = module.eventbridge.eventbridge_rule_arns["crons"]
67
68 depends_on = [
69 module.eventbridge
70 ]
71}
IAM Permission
The producer Lambda function needs permission to send messages to the orders topic of the MSK cluster. Also, it needs permission on the Glue schema registry and schema. The following IAM policy is added to the Lambda function.
1# glue-schema-registry/app/main.tf
2resource "aws_iam_policy" "msk_lambda_producer_permission" {
3 name = "${local.producer.function_name}-msk-lambda-producer-permission"
4
5 policy = jsonencode({
6 Version = "2012-10-17"
7 Statement = [
8 {
9 Sid = "PermissionOnCluster"
10 Action = [
11 "kafka-cluster:Connect",
12 "kafka-cluster:AlterCluster",
13 "kafka-cluster:DescribeCluster"
14 ]
15 Effect = "Allow"
16 Resource = "arn:aws:kafka:${local.region}:${data.aws_caller_identity.current.account_id}:cluster/${local.infra_prefix}-msk-cluster/*"
17 },
18 {
19 Sid = "PermissionOnTopics"
20 Action = [
21 "kafka-cluster:*Topic*",
22 "kafka-cluster:WriteData",
23 "kafka-cluster:ReadData"
24 ]
25 Effect = "Allow"
26 Resource = "arn:aws:kafka:${local.region}:${data.aws_caller_identity.current.account_id}:topic/${local.infra_prefix}-msk-cluster/*"
27 },
28 {
29 Sid = "PermissionOnGroups"
30 Action = [
31 "kafka-cluster:AlterGroup",
32 "kafka-cluster:DescribeGroup"
33 ]
34 Effect = "Allow"
35 Resource = "arn:aws:kafka:${local.region}:${data.aws_caller_identity.current.account_id}:group/${local.infra_prefix}-msk-cluster/*"
36 },
37 {
38 Sid = "PermissionOnGlueSchema"
39 Action = [
40 "glue:*Schema*",
41 "glue:GetRegistry",
42 "glue:CreateRegistry",
43 "glue:ListRegistries",
44 ]
45 Effect = "Allow"
46 Resource = "*"
47 }
48 ]
49 })
50}
EventBridge Rule
The AWS EventBridge Terraform module is used to create the EventBridge schedule rule and targets. Note that 5 targets that point to the Kafka producer Lambda function are created so that it is invoked concurrently every minute.
1module "eventbridge" {
2 source = "terraform-aws-modules/eventbridge/aws"
3
4 create_bus = false
5
6 rules = {
7 crons = {
8 description = "Kafka producer lambda schedule"
9 schedule_expression = local.producer.schedule_rate
10 }
11 }
12
13 targets = {
14 crons = [for i in range(local.producer.concurrency) : {
15 name = "lambda-target-${i}"
16 arn = module.kafka_producer_lambda.lambda_function_arn
17 }]
18 }
19
20 depends_on = [
21 module.kafka_producer_lambda
22 ]
23
24 tags = local.tags
25}
Consumer App
Lambda Handler
The Lambda event includes records, which is a dictionary where the key is a topic partition (topic_name-partiton_number) and the value is a list of consumer records. The consumer records include both the message metadata (topic, partition, offset, timestamp…), key and value. An example payload is shown below.
1{
2 "eventSource": "aws:kafka",
3 "eventSourceArn": "<msk-cluster-arn>",
4 "bootstrapServers": "<bootstrap-server-addresses>",
5 "records": {
6 "orders-2": [
7 {
8 "topic": "orders",
9 "partition": 2,
10 "offset": 10293,
11 "timestamp": 1680631941838,
12 "timestampType": "CREATE_TIME",
13 "key": "eyJvcmRlcl9pZCI6ICJkNmQ4ZDJjNi1hODYwLTQyNTYtYWY1Yi04ZjU3NDkxZmM4YWYifQ==",
14 "value": "AwDeD/rgjCxCeawN/ZaIO6VuSGQ2ZDhkMmM2LWE4NjAtNDI1Ni1hZjViLThmNTc0OTFmYzhhZu6pwtfpYQppYWJ4YQa8UBSEHgbkVAYA",
15 "headers": [],
16 }
17 ]
18 }
19}
The ConsumerRecord class parses/formats a consumer record. As the key and value are returned as base64 encoded string, it is decoded into bytes, followed by decoding or deserializing appropriately. The LambdaDeserializer class is created to deserialize the value. Also, the message timestamp is converted into the datetime object. The parse_record() method returns the consumer record with parsed/formatted values.
1# glue-schema-registry/app/consumer/lambda_handler.py
2import os
3import json
4import base64
5import datetime
6
7import boto3
8from aws_schema_registry import SchemaRegistryClient
9from aws_schema_registry.adapter.kafka import Deserializer, KafkaDeserializer
10
11
12class LambdaDeserializer(Deserializer):
13 def __init__(self, registry: str):
14 self.registry = registry
15
16 @property
17 def deserializer(self):
18 glue_client = boto3.client(
19 "glue", region_name=os.getenv("AWS_DEFAULT_REGION", "ap-southeast-2")
20 )
21 client = SchemaRegistryClient(glue_client, registry_name=self.registry)
22 return KafkaDeserializer(client)
23
24 def deserialize(self, topic: str, bytes_: bytes):
25 return self.deserializer.deserialize(topic, bytes_)
26
27
28class ConsumerRecord:
29 def __init__(self, record: dict):
30 self.topic = record["topic"]
31 self.partition = record["partition"]
32 self.offset = record["offset"]
33 self.timestamp = record["timestamp"]
34 self.timestamp_type = record["timestampType"]
35 self.key = record["key"]
36 self.value = record["value"]
37 self.headers = record["headers"]
38
39 def parse_key(self):
40 return base64.b64decode(self.key).decode()
41
42 def parse_value(self, deserializer: LambdaDeserializer):
43 parsed = deserializer.deserialize(self.topic, base64.b64decode(self.value))
44 return parsed.data
45
46 def format_timestamp(self, to_str: bool = True):
47 ts = datetime.datetime.fromtimestamp(self.timestamp / 1000)
48 if to_str:
49 return ts.isoformat()
50 return ts
51
52 def parse_record(
53 self, deserializer: LambdaDeserializer, to_str: bool = True, to_json: bool = True
54 ):
55 rec = {
56 **self.__dict__,
57 **{
58 "key": self.parse_key(),
59 "value": self.parse_value(deserializer),
60 "timestamp": self.format_timestamp(to_str),
61 },
62 }
63 if to_json:
64 return json.dumps(rec, default=self.serialize)
65 return rec
66
67 def serialize(self, obj):
68 if isinstance(obj, datetime.datetime):
69 return obj.isoformat()
70 if isinstance(obj, datetime.date):
71 return str(obj)
72 return obj
73
74
75def lambda_function(event, context):
76 deserializer = LambdaDeserializer(os.getenv("REGISTRY_NAME", "customer"))
77 for _, records in event["records"].items():
78 for record in records:
79 cr = ConsumerRecord(record)
80 print(cr.parse_record(deserializer))
Lambda Resource
The AWS Lambda Terraform module is used to create the consumer Lambda function as well. Lambda event source mapping is created so that it polls messages from the orders topic and invoke the consumer function. Also, we need to give permission to the MSK cluster to invoke the Lambda function, and it is given by the aws_lambda_permission resource.
1# glue-schema-registry/app/variables.tf
2locals {
3 name = local.infra_prefix
4 region = data.aws_region.current.name
5 environment = "dev"
6
7 infra_prefix = "glue-schema-registry"
8
9 ...
10
11 consumer = {
12 src_path = "consumer"
13 function_name = "kafka_consumer"
14 handler = "lambda_handler.lambda_function"
15 timeout = 90
16 memory_size = 128
17 runtime = "python3.8"
18 topic_name = "orders"
19 starting_position = "TRIM_HORIZON"
20 environment = {
21 registry_name = "customer"
22 }
23 }
24}
25
26# glue-schema-registry/app/main.tf
27module "kafka_consumer_lambda" {
28 source = "terraform-aws-modules/lambda/aws"
29
30 function_name = local.consumer.function_name
31 handler = local.consumer.handler
32 runtime = local.consumer.runtime
33 timeout = local.consumer.timeout
34 memory_size = local.consumer.memory_size
35 source_path = local.consumer.src_path
36 vpc_subnet_ids = data.aws_subnets.private.ids
37 vpc_security_group_ids = [data.aws_security_group.kafka_app_lambda.id]
38 attach_network_policy = true
39 attach_policies = true
40 policies = [aws_iam_policy.msk_lambda_consumer_permission.arn]
41 number_of_policies = 1
42 environment_variables = {
43 REGISTRY_NAME = local.producer.environment.registry_name
44 }
45
46 tags = local.tags
47}
48
49resource "aws_lambda_event_source_mapping" "kafka_consumer_lambda" {
50 event_source_arn = data.aws_msk_cluster.msk_data_cluster.arn
51 function_name = module.kafka_consumer_lambda.lambda_function_name
52 topics = [local.consumer.topic_name]
53 starting_position = local.consumer.starting_position
54 amazon_managed_kafka_event_source_config {
55 consumer_group_id = "${local.consumer.topic_name}-group-01"
56 }
57}
58
59resource "aws_lambda_permission" "allow_msk" {
60 statement_id = "InvokeLambdaFunction"
61 action = "lambda:InvokeFunction"
62 function_name = local.consumer.function_name
63 principal = "kafka.amazonaws.com"
64 source_arn = data.aws_msk_cluster.msk_data_cluster.arn
65}
IAM Permission
As the Lambda event source mapping uses the permission of the Lambda function, we need to add permission related to Kafka cluster, Kafka and networking - see the AWS documentation for details. Finally, permission on the Glue schema registry and schema is added as the consumer should be able to request relevant schemas.
1# glue-schema-registry/app/main.tf
2resource "aws_iam_policy" "msk_lambda_consumer_permission" {
3 name = "${local.consumer.function_name}-msk-lambda-consumer-permission"
4
5 policy = jsonencode({
6 Version = "2012-10-17"
7 Statement = [
8 {
9 Sid = "PermissionOnKafkaCluster"
10 Action = [
11 "kafka-cluster:Connect",
12 "kafka-cluster:DescribeGroup",
13 "kafka-cluster:AlterGroup",
14 "kafka-cluster:DescribeTopic",
15 "kafka-cluster:ReadData",
16 "kafka-cluster:DescribeClusterDynamicConfiguration"
17 ]
18 Effect = "Allow"
19 Resource = [
20 "arn:aws:kafka:${local.region}:${data.aws_caller_identity.current.account_id}:cluster/${local.infra_prefix}-msk-cluster/*",
21 "arn:aws:kafka:${local.region}:${data.aws_caller_identity.current.account_id}:topic/${local.infra_prefix}-msk-cluster/*",
22 "arn:aws:kafka:${local.region}:${data.aws_caller_identity.current.account_id}:group/${local.infra_prefix}-msk-cluster/*"
23 ]
24 },
25 {
26 Sid = "PermissionOnKafka"
27 Action = [
28 "kafka:DescribeCluster",
29 "kafka:GetBootstrapBrokers"
30 ]
31 Effect = "Allow"
32 Resource = "*"
33 },
34 {
35 Sid = "PermissionOnNetwork"
36 Action = [
37 # The first three actions also exist in netwrok policy attachment in lambda module
38 # "ec2:CreateNetworkInterface",
39 # "ec2:DescribeNetworkInterfaces",
40 # "ec2:DeleteNetworkInterface",
41 "ec2:DescribeVpcs",
42 "ec2:DescribeSubnets",
43 "ec2:DescribeSecurityGroups"
44 ]
45 Effect = "Allow"
46 Resource = "*"
47 },
48 {
49 Sid = "PermissionOnGlueSchema"
50 Action = [
51 "glue:*Schema*",
52 "glue:ListRegistries"
53 ]
54 Effect = "Allow"
55 Resource = "*"
56 }
57 ]
58 })
59}
Schema Evolution Demo
Before testing the Kafka applications, I’ll quickly demonstrate how the schema registry can be used for managing and validating schemas for topic message data. Each schema can have a compatibility mode (or disabled) and the scope of changes is restricted by it. For example, the default BACKWARD mode only allows to delete fields or add optional fields. (See the Confluent document for a quick summary.) Therefore, if we add a mandatory field to an existing schema, it will be not validated, and it fails to send a message to the topic. In order to illustrate it, I created a single node Kafka cluster using docker-compose as shown below.
1# glue-schema-registry/compose-demo.yml
2version: "3.5"
3
4services:
5 zookeeper:
6 image: docker.io/bitnami/zookeeper:3.8
7 container_name: zookeeper
8 ports:
9 - "2181"
10 networks:
11 - kafkanet
12 environment:
13 - ALLOW_ANONYMOUS_LOGIN=yes
14 volumes:
15 - zookeeper_data:/bitnami/zookeeper
16 kafka-0:
17 image: docker.io/bitnami/kafka:3.3
18 container_name: kafka-0
19 expose:
20 - 9092
21 ports:
22 - "9093:9093"
23 networks:
24 - kafkanet
25 environment:
26 - ALLOW_PLAINTEXT_LISTENER=yes
27 - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
28 - KAFKA_CFG_BROKER_ID=0
29 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
30 - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
31 - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka-0:9092,EXTERNAL://localhost:9093
32 - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT
33 volumes:
34 - kafka_0_data:/bitnami/kafka
35 depends_on:
36 - zookeeper
37
38networks:
39 kafkanet:
40 name: kafka-network
41
42volumes:
43 zookeeper_data:
44 driver: local
45 kafka_0_data:
46 driver: local
Recall that the producer lambda handler has a conditional block, and it is executed when it is called as a script. If we don’t specify the environment variable of USE_MORE, it sends a messages based on the Order class. Otherwise, a message is created from the OrderMore class, which has an additional boolean attribute called is_prime. As the compatibility mode is set to be BACKWARD, we can expect the second round of execution will not be successful. As shown below, I executed the lambda handler twice and the second round failed with the following error, which indicates schema validation failure.
aws_schema_registry.exception.SchemaRegistryException: Schema Found but status is FAILURE
1export BOOTSTRAP_SERVERS=localhost:9093
2export TOPIC_NAME=demo
3export REGISTRY_NAME=customer
4
5cd glue-schema-registry/app/producer
6
7## Round 1 - send message from the Order class
8python lambda_handler.py
9
10## Round 2 - send message from the OrderMore class
11export USE_MORE=1
12
13python lambda_handler.py
We can see the details from the schema version and the second version is marked as failed.
Note that schema versioning and validation would be more relevant to the clients that tightly link the schema and message records. However, it would still be important for a Python client in order to work together with those clients or Kafka connect.
Deployment
Topic Creation
We plan to create the orders topic with multiple partitions. Although we can use the Kafka CLI tool, it can be performed easily using Kpow. It is a Kafka monitoring and management tool, which provides a web UI. Also, it supports the Glue Schema Registry and MSK Connect out-of-box, which is quite convenient. In the docker-compose file, we added environment variables for the MSK cluster, MSK Connect and Glue Schema Registry details. Note it fails to start if the schema registry does not exist. I created the registry while I demonstrated schema evolution, or it can be created simply as shown below.
1$ aws glue create-registry --registry-name customer
1# glue-schema-registry/docker-compose.yml
2version: "3.5"
3
4services:
5 kpow:
6 image: factorhouse/kpow-ce:91.2.1
7 container_name: kpow
8 ports:
9 - "3000:3000"
10 networks:
11 - kafkanet
12 environment:
13 AWS_ACCESS_KEY_ID: $AWS_ACCESS_KEY_ID
14 AWS_SECRET_ACCESS_KEY: $AWS_SECRET_ACCESS_KEY
15 AWS_SESSION_TOKEN: $AWS_SESSION_TOKEN
16 # kafka cluster
17 BOOTSTRAP: $BOOTSTRAP_SERVERS
18 SECURITY_PROTOCOL: SASL_SSL
19 SASL_MECHANISM: AWS_MSK_IAM
20 SASL_CLIENT_CALLBACK_HANDLER_CLASS: software.amazon.msk.auth.iam.IAMClientCallbackHandler
21 SASL_JAAS_CONFIG: software.amazon.msk.auth.iam.IAMLoginModule required;
22 # msk connect
23 CONNECT_AWS_REGION: $AWS_DEFAULT_REGION
24 # glue schema registry
25 SCHEMA_REGISTRY_ARN: $SCHEMA_REGISTRY_ARN
26 SCHEMA_REGISTRY_REGION: $AWS_DEFAULT_REGION
27
28networks:
29 kafkanet:
30 name: kafka-network
Once started, we can visit the UI on port 3000. The topic is created in the Topic menu by specifying the topic name and the number of partitions.
Once created, we can check details of the topic by selecting the topic from the drop-down menu.
Local Testing with SAM
To simplify development, the EventBridge permission is disabled by setting to_enable_trigger to false. Also, it is shortened to loop before it gets stopped by reducing max_run_sec to 10.
1# glue-schema-registry/app/variables.tf
2locals {
3 ...
4
5 producer = {
6 ...
7 to_enable_trigger = false
8 environment = {
9 topic_name = "orders"
10 registry_name = "customer"
11 max_run_sec = 10
12 }
13 }
14 ...
15}
The Lambda function can be built with the SAM build command while specifying the hook name as terraform and enabling beta features. Once completed, it stores the build artifacts and template in the .aws-sam folder.
1$ sam build --hook-name terraform --beta-features
2
3# Apply complete! Resources: 3 added, 0 changed, 0 destroyed.
4
5# Build Succeeded
6
7# Built Artifacts : .aws-sam/build
8# Built Template : .aws-sam/build/template.yaml
9
10# Commands you can use next
11# =========================
12# [*] Invoke Function: sam local invoke --hook-name terraform
13# [*] Emulate local Lambda functions: sam local start-lambda --hook-name terraform
14
15# SAM CLI update available (1.78.0); (1.70.0 installed)
16# To download: https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-sam-cli-install.html
We can invoke the Lambda function locally using the SAM local invoke command. The Lambda function is invoked in a Docker container and the invocation logs are printed in the terminal as shown below. Note that we should be connected to the VPN server in order to send messages into the MSK cluster, which is deployed in private subnets.
1$ sam local invoke --hook-name terraform module.kafka_producer_lambda.aws_lambda_function.this[0] --beta-features
2
3# Experimental features are enabled for this session.
4# Visit the docs page to learn more about the AWS Beta terms https://aws.amazon.com/service-terms/.
5
6# Skipped prepare hook. Current application is already prepared.
7# Invoking lambda_handler.lambda_function (python3.8)
8# Skip pulling image and use local one: public.ecr.aws/sam/emulation-python3.8:rapid-1.70.0-x86_64.
9
10# Mounting /home/jaehyeon/personal/kafka-pocs/glue-schema-registry/app/.aws-sam/build/ModuleKafkaProducerLambdaAwsLambdaFunctionThis069E06354 as /var/task:ro,delegated inside runtime container
11# START RequestId: fdbba255-e5b0-4e21-90d3-fe0b2ebbf629 Version: $LATEST
12# sent 100 messages
13# sent 100 messages
14# sent 100 messages
15# sent 100 messages
16# sent 100 messages
17# sent 100 messages
18# sent 100 messages
19# sent 100 messages
20# sent 100 messages
21# sent 100 messages
22# 1000 records are sent in 11 seconds ...
23# END RequestId: fdbba255-e5b0-4e21-90d3-fe0b2ebbf629
24# REPORT RequestId: fdbba255-e5b0-4e21-90d3-fe0b2ebbf629 Init Duration: 0.22 ms Duration: 12146.61 ms Billed Duration: 12147 ms Memory Size: 128 MB Max Memory Used: 128 MB
25# null
Once completed, we can check the value schema (orders-value) is created in the Kpow UI as shown below.
We can check the messages. In order to check them correctly, we need to select AVRO as the value deserializer and glue1 as the schema registry.
Kafka App Deployment
Now we can deploy the Kafka applications using Terraform as usual after resetting the configuration variables. Once deployed, we can see that the scheduler rule has 5 targets of the same Lambda function.
We can see the Lambda consumer parses the consumer records correctly in CloudWatch logs.
Summary
Schema registry provides a centralized repository for managing and validating schemas for topic message data. In AWS, the Glue Schema Registry supports features to manage and enforce schemas on data streaming applications using convenient integrations with a range of AWS services. In this post, we discussed how to integrate Python Kafka producer and consumer apps in AWS Lambda with the Glue Schema Registry.
Comments