Working with the managed schema registry

To use Managed Schema Registry with Managed Service for Apache Kafka®:

1. Create the producer and consumer scripts on the local machine.
2. Check that Managed Schema Registry runs correctly.
3. Delete the resources you created.

Before you begin

1. Create a Managed Service for Apache Kafka® cluster with any suitable configuration. When creating a cluster, enable Schema registry and Public access.

   1. Create a topic named messages for exchanging messages between the producer and the consumer.
   2. Create a user named user and grant it the rights for the messages topic:
      • ACCESS_ROLE_CONSUMER,
      • ACCESS_ROLE_PRODUCER.

2. In the network hosting the Managed Service for Apache Kafka® cluster, create a VM with Ubuntu 20.04 and a public IP address.

3. To allow traffic between the Managed Service for Apache Kafka® cluster and the virtual machine, configure security groups.

Create producer and consumer scripts

The above scripts send and receive messages in the messages topic as a key:meaning pair. In the example, data format schemas are described in Avro format.

1. Connect to the virtual machine over SSH.

2. Install the necessary Python packages:

   sudo apt-get update && \
   sudo pip3 install avro confluent_kafka
   

3. To use an encrypted connection, install an SSL certificate.

   sudo mkdir -p /usr/share/ca-certificates && \
   sudo wget "https://storage.yandexcloud.net/cloud-certs/CA.pem" \
             -O /usr/share/ca-certificates/YandexCA.crt && \
   sudo chmod 655 /usr/share/ca-certificates/YandexCA.crt
   

4. Create a Python script for the consumer.

   The script works as follows:

   1. Connect to the messages topic and Confluent Schema Registry.
   2. In a continuous cycle, read messages sent to the messages topic.
   3. When receiving a message, request the necessary schemas in Confluent Schema Registry to parse the message.
   4. Parse binary data from the message according to the schemas for the key and value and display the result on the screen.

   consumer.py

   #!/usr/bin/python3
   
   from confluent_kafka.avro import AvroConsumer
   from confluent_kafka.avro.serializer import SerializerError
   
   
   c = AvroConsumer(
       {
           "bootstrap.servers": ','.join([
           "<FQDN of first broker host>:9091",
           ...
           "<FQDN of the Nth broker host>:9091",
           ]),
           "group.id": "avro-consumer",
           "security.protocol": "SASL_SSL",
           "ssl.ca.location": "/usr/share/ca-certificates/YandexCA.crt",
           "sasl.mechanism": "SCRAM-SHA-512",
           "sasl.username": "user",
           "sasl.password": "<password of the user named user>",
           "schema.registry.url": "https://<FQDN or IP address of the Managed Schema Registry server>:443",
           "schema.registry.basic.auth.credentials.source": "SASL_INHERIT",
           "auto.offset.reset": "earliest"
       }
   )
   
   c.subscribe(["messages"])
   
   while True:
       try:
           msg = c.poll(10)
   
       except SerializerError as e:
           print("Message deserialization failed for {}: {}".format(msg, e))
           break
   
       if msg is None:
           continue
   
       if msg.error():
           print("AvroConsumer error: {}".format(msg.error()))
           continue
   
       print(msg.value())
   
   c.close()
   

5. Create a Python script for the producer.

   The script works as follows:

   1. Connect to the schema registry and pass to it the data format schemas for the key and value.
   2. Generate the key and value based on the passed schemas.
   3. Send a message consisting of the key:meaning pair to the messages topic. The schema versions are added to the message automatically.

   producer.py

   #!/usr/bin/python3
   
   from confluent_kafka import avro
   from confluent_kafka.avro import AvroProducer
   
   
   value_schema_str = """
   {
       "namespace": "my.test",
       "name": "value",
       "type": "record",
       "fields": [
           {
               "name": "name",
               "type": "string"
           }
       ]
   }
   """
   
   key_schema_str = """
   {
       "namespace": "my.test",
       "name": "key",
       "type": "record",
       "fields": [
           {
               "name": "name",
               "type": "string"
           }
       ]
   }
   """
   
   value_schema = avro.loads(value_schema_str)
   key_schema = avro.loads(key_schema_str)
   value = {"name": "Value"}
   key = {"name": "Key"}
   
   
   def delivery_report(err, msg):
       """Called once for each message produced to indicate delivery result.
       Triggered by poll() or flush()."""
       if err is not None:
           print("Message delivery failed: {}".format(err))
       else:
           print("Message delivered to {} [{}]".format(msg.topic(), msg.partition()))
   
   
   avroProducer = AvroProducer(
       {
           "bootstrap.servers": ','.join([
               "<FQDN of the 1st broker host>:9091",
               ...
               "<FQDN of the Nth broker host>:9091",
           ]),
           "security.protocol": 'SASL_SSL',
           "ssl.ca.location": '/usr/share/ca-certificates/YandexCA.crt',
           "sasl.mechanism": 'SCRAM-SHA-512',
           "sasl.username": 'user',
           "sasl.password": '<password of the user named user>',
           "on_delivery": delivery_report,
           "schema.registry.basic.auth.credentials.source": 'SASL_INHERIT',
           "schema.registry.url": 'https://<FQDN or IP address of the Managed Schema Registry server>:443'
       },
       default_key_schema=key_schema,
       default_value_schema=value_schema
   )
   
   avroProducer.produce(topic="messages", key=key, value=value)
   avroProducer.flush()
   

Check that Managed Schema Registry runs correctly

1. Start the consumer:

   python3 ./consumer.py
   

2. In a separate terminal, start the producer:

   python3 ./producer.py
   

3. Make sure that the data sent by the producer is received and correctly interpreted by the consumer:

   {'name': 'Value'}
   

Delete the resources you created

If you no longer need the created resources, delete the VM instance and the Managed Service for Apache Kafka® cluster.

If you have reserved a public static IP address for the created VM, delete it.