[Yandex Cloud documentation](../../index.md) > [Yandex Managed Service for Apache Kafka®](../index.md) > [Tutorials](index.md) > Using data format schemas with Managed Service for Apache Kafka® > Working with the managed schema registry

# Working with the managed schema registry

To use [Managed Schema Registry](../concepts/managed-schema-registry.md#msr) with Managed Service for Apache Kafka®:

1. [Create producer and consumer scripts on your local machine](#create-scripts).
1. [Check that Managed Schema Registry runs correctly](#check-schema-registry).
1. [Delete the resources you created](#clear-out).

This tutorial describes how to register a single data schema. For more information on how to register multiple data schemas, see [this Confluent Schema Registry guide](https://docs.confluent.io/platform/current/control-center/topics/schema.html).


## Required paid resources {#paid-resources}

The infrastructure support cost includes:

* Fee for computing resources of the Managed Service for Apache Kafka® cluster and storage space (see [Managed Service for Apache Kafka® pricing](../pricing.md)).
* Fee for VM computing resources and disks (see [Yandex Compute Cloud pricing](../../compute/pricing.md)).
* Fee for using a [public IP address](../../vpc/concepts/ips.md) (see [Yandex Virtual Private Cloud pricing](../../vpc/pricing.md)).


## Getting started {#before-you-begin}

1. [Create a Managed Service for Apache Kafka® cluster](../operations/cluster-create.md) of any suitable configuration. When creating a cluster, enable **Schema registry** and **Public access**.

    {% note info %}
    
    Public access to cluster hosts is required if you plan to connect to the cluster via the internet. This connection option is simpler and is recommended for the purposes of this guide. You can connect to non-public hosts as well but only from Yandex Cloud virtual machines located in the same cloud network as the cluster.
    
    {% endnote %}

    1. [Create a topic](../operations/cluster-topics.md#create-topic) named `messages` for exchanging messages between the producer and the consumer.
    1. [Create a user](../operations/cluster-accounts.md#create-account) named `user` and [grant them permissions](../operations/cluster-accounts.md#grant-permission) for the `messages` topic:
        * `ACCESS_ROLE_CONSUMER`
        * `ACCESS_ROLE_PRODUCER`

1. In the network hosting the Managed Service for Apache Kafka® cluster, [create a VM](../../compute/operations/vm-create/create-linux-vm.md) running [Ubuntu 20.04](https://yandex.cloud/en/marketplace/products/yc/ubuntu-20-04-lts) with a public IP address.


1. If using security groups, [configure them](../operations/connect/index.md#configuring-security-groups) to allow all required traffic between your Managed Service for Apache Kafka® cluster and VM.


## Create producer and consumer scripts {#create-scripts}

These scripts send and receive messages in the `messages` topic as a `key:value` pair. This example shows the data format schemas in [Avro](https://avro.apache.org/) format.

{% note info %}

Python scripts are provided for demonstration only. You can prepare and send data format schemas and the data itself by creating a similar script in another language.

{% endnote %}

1. [Connect](../../compute/operations/vm-connect/ssh.md) to the VM over SSH.

1. Install the required Python packages:

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

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

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

1. Create a Python script for the consumer.

    Here is how the script works:
    
    1. Connect to the `messages` topic and Confluent Schema Registry.
    1. Continuously read messages arriving in the `messages` topic.
    1. When receiving a message, request the required schemas from Confluent Schema Registry to parse the message.
    1. Parse the message binary data based on the key and value schemas and display the result.

    `consumer.py`

    ```python
    #!/usr/bin/python3

    from confluent_kafka.avro import AvroConsumer
    from confluent_kafka.avro.serializer import SerializerError


    c = AvroConsumer(
        {
            "bootstrap.servers": ','.join([
            "<broker_host_1_FQDN>:9091",
            ...
            "<broker_host_N_FQDN>:9091",
            ]),
            "group.id": "avro-consumer",
            "security.protocol": "SASL_SSL",
            "ssl.ca.location": "/usr/share/ca-certificates/YandexInternalRootCA.crt",
            "sasl.mechanism": "SCRAM-SHA-512",
            "sasl.username": "user",
            "sasl.password": "<user_password>",
            "schema.registry.url": "https://<Managed_Schema_Registry_server_FQDN_or_IP_address>:443",
            "schema.registry.basic.auth.credentials.source": "SASL_INHERIT",
            "schema.registry.ssl.ca.location": "/usr/share/ca-certificates/YandexInternalRootCA.crt",
            "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()
    ```

1. Create a Python script for the producer.

    Here is how the script works:
    
    1. Connect to the schema registry and send the key and value data format schemas.
    1. Generate the key and value based on the schemas you sent.
    1. Send a message containing a `key:value` pair to the `messages` topic. The system will automatically add the schema versions to your message.

    `producer.py`

    ```python
    #!/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([
                "<broker_host_1_FQDN>:9091",
                ...
                "<broker_host_N_FQDN>:9091",
            ]),
            "security.protocol": 'SASL_SSL',
            "ssl.ca.location": '/usr/share/ca-certificates/YandexInternalRootCA.crt',
            "sasl.mechanism": 'SCRAM-SHA-512',
            "sasl.username": 'user',
            "sasl.password": '<user_password>',
            "on_delivery": delivery_report,
            "schema.registry.basic.auth.credentials.source": 'SASL_INHERIT',
            "schema.registry.url": 'https://<Managed_Schema_Registry_server_FQDN_or_IP_address>:443',
            "schema.registry.ssl.ca.location": "/usr/share/ca-certificates/YandexInternalRootCA.crt"
        },
        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 {#check-schema-registry}

1. Start the consumer:

    ```bash
    python3 ./consumer.py
    ```

1. In a separate terminal, start the producer:

    ```bash
    python3 ./producer.py
    ```

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

    ```text
    {'name': 'Value'}
    ```

## Delete the resources you created {#clear-out}

Delete the resources you no longer need to avoid paying for them:

* [Delete the Managed Service for Apache Kafka® cluster](../operations/cluster-delete.md).
* [Delete the VM](../../compute/operations/vm-control/vm-delete.md).
* If you reserved public static IP addresses, release and [delete them](../../vpc/operations/address-delete.md).