[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® > Using Confluent Schema Registry with Managed Service for Apache Kafka®

# Using Confluent Schema Registry with Managed Service for Apache Kafka®

In Managed Service for Apache Kafka®, you can use the integrated [Managed Schema Registry](../concepts/managed-schema-registry.md#msr). For more information, see [Working with the managed schema registry](managed-schema-registry.md). To use [Confluent Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html), follow this tutorial.

{% note info %}

We tested this tutorial with Confluent Schema Registry 6.2 and a VM running Ubuntu 20.04 LTS. We do not guarantee support for newer versions.

{% endnote %}

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

1. [Create a topic for notifications about data format schema changes](#create-schemas-topic).
1. [Install and configure Confluent Schema Registry on your VM](#configure-vm).
1. [Create producer and consumer scripts](#create-scripts).
1. [Make sure Confluent Schema Registry works correctly](#check-schema-registry).

If you no longer need the resources you created, [delete them](#clear-out).


## Required paid resources {#paid-resources}

The support cost for this solution includes:

* Managed Service for Apache Kafka® cluster fee: use of computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see [Apache Kafka® pricing](../pricing.md)).
* Fee for public IP addresses if public access is enabled for cluster hosts (see [Virtual Private Cloud pricing](../../vpc/pricing.md)).
* VM fee: use of computing resources, storage, and public IP address (see [Compute Cloud pricing](../../compute/pricing.md)).


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

1. [Create a Managed Service for Apache Kafka® cluster](../operations/cluster-create.md) of any suitable configuration.

    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 LTS](https://yandex.cloud/en/marketplace/products/yc/ubuntu-20-04-lts) from Cloud Marketplace 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.

1. In the VM security group, [create](../../vpc/operations/security-group-add-rule.md) an inbound rule that allows connections via port `8081` which is used by the producer and consumer to access the schema registry:

    * **Port range**: `8081`.
    * **Protocol**: `TCP`.
    * **Destination name**: `CIDR`.
    * **CIDR blocks**: `0.0.0.0/0` or address ranges of the subnets used by the producer and the consumer.


## Create a topic for notifications about data format schema changes {#create-schemas-topic}

1. [Create a service topic](../operations/cluster-topics.md#create-topic) named `_schemas` with the following settings:

    * **Number of partitions**: `1`.
    * **Cleanup policy**: `Compact`.

    {% note warning %}

    These values for **Number of partitions** and **Cleanup policy** are required for Confluent Schema Registry to run correctly.

    {% endnote %}

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

    Confluent Schema Registry will use this account to work with `_schemas`.

## Install and configure Confluent Schema Registry on your VM {#configure-vm}

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

1. Add the Confluent Schema Registry repository:

    ```bash
    wget -qO - https://packages.confluent.io/deb/6.2/archive.key | sudo apt-key add - && \
    sudo add-apt-repository "deb [arch=amd64] https://packages.confluent.io/deb/6.2 stable main"
    ```

1. Install the packages:

    ```bash
    sudo apt-get update && \
    sudo apt-get install \
         confluent-schema-registry \
         openjdk-11-jre-headless \
         python3-pip --yes
    ```

1. [Get an SSL certificate](../operations/connect/index.md#get-ssl-cert).

1. Create a secure store for the certificate:

    ```bash
    sudo keytool \
         -keystore /etc/schema-registry/client.truststore.jks \
         -alias CARoot \
         -import -file /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
         -storepass <secure_certificate_storage_password> \
         --noprompt
    ```

1. Create a file named `/etc/schema-registry/jaas.conf` with settings for connecting to the cluster:

    ```scala
    KafkaClient {
      org.apache.kafka.common.security.scram.ScramLoginModule required
      username="registry"
      password="<registry_user_password>";
    };
    ```

1. Edit the `/etc/schema-registry/schema-registry.properties` file containing Confluent Schema Registry settings:

    1. Comment out the line as follows:

        ```ini
        kafkastore.bootstrap.servers=PLAINTEXT://localhost:9092
        ```

    1. Uncomment the line with the `listeners` parameter. It configures the network address and port that Confluent Schema Registry listens to. The default port for all network interfaces is `8081`:

        ```ini
        listeners=http://0.0.0.0:8081
        ```

    1. Add the following lines at the end of the file:

        ```ini
        kafkastore.bootstrap.servers=SASL_SSL://<broker_host_1_FQDN:9091>,<broker_host_2_FQDN:9091>,...,<broker_host_N_FQDN:9091>
        kafkastore.ssl.truststore.location=/etc/schema-registry/client.truststore.jks
        kafkastore.ssl.truststore.password=<secure_certificate_storage_password>
        kafkastore.sasl.mechanism=SCRAM-SHA-512
        kafkastore.security.protocol=SASL_SSL
        ```

        You can get a list of broker hosts [with a list of cluster hosts](../operations/cluster-hosts.md).

1. Edit the `/lib/systemd/system/confluent-schema-registry.service` file which describes the _systemd_ module.

    1. Go to the `[Service]` section.
    1. Add the `Environment` parameter with Java settings:

        ```ini
        ...

        [Service]
        Type=simple
        User=cp-schema-registry
        Group=confluent
        Environment="LOG_DIR=/var/log/confluent/schema-registry"
        Environment="_JAVA_OPTIONS='-Djava.security.auth.login.config=/etc/schema-registry/jaas.conf'"
        ...
        ```

1. Update the `systemd` module details:

    ```bash
    sudo systemctl daemon-reload
    ```

1. Start the Confluent Schema Registry service:

    ```bash
    sudo systemctl start confluent-schema-registry.service
    ```

1. Set Confluent Schema Registry to start automatically after a system reboot:

    ```bash
    sudo systemctl enable confluent-schema-registry.service
    ```

## 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. Install the required Python packages:

    ```bash
    sudo pip3 install avro confluent_kafka
    ```

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/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt",
            "sasl.mechanism": "SCRAM-SHA-512",
            "sasl.username": "user",
            "sasl.password": "<user_password>",
            "schema.registry.url": "http://<Confluent_Schema_Registry_server_FQDN_or_IP_address>:8081",
        }
    )

    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/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt",
            "sasl.mechanism": "SCRAM-SHA-512",
            "sasl.username": "user",
            "sasl.password": "<user_password>",
            "on_delivery": delivery_report,
            "schema.registry.url": "http://<Schema_Registry_server_FQDN_or_IP_address>:8081",
        },
        default_key_schema=key_schema,
        default_value_schema=value_schema,
    )

    avroProducer.produce(topic="messages", key=key, value=value)
    avroProducer.flush()
    ```

## Make sure Confluent Schema Registry works 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).