Using Managed Schema Registry with Yandex Managed Service for Apache Kafka® through the REST API

In Managed Service for Apache Kafka® clusters, you can work with Managed Schema Registry either using Apache Kafka® clients for various programming languages or the REST API.

Managed Service for Apache Kafka® also provides the REST API for Apache Kafka®. Among other things, this API allows you to send and receive messages without using third-party producers and consumers. These features will also be demonstrated in this tutorial.

To get to know the REST API for Managed Schema Registry and Apache Kafka®:

1. Create data format schemas.
2. Send messages to a topic.
3. Get messages from a topic.
4. Delete the resources you created.

Required paid resourcesRequired paid resources

The support cost includes:

• Managed Service for Apache Kafka® cluster fee: Using computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see Apache Kafka® pricing).
• Fee for using public IP addresses (see Virtual Private Cloud pricing).

Getting startedGetting started

Prepare the infrastructurePrepare the infrastructure

Install utilitiesInstall utilities

1. Install cURL:

   sudo apt install curl -y
   

   It will be used to make requests to the API.

   For convenience, this tutorial will use the --user cURL option when making requests to the API. If this option is used, cURL will automatically add to the request the Authorization HTTP header with the value required for authorization.

   Tip

   You can build the Authorization header yourself, e.g., if not using cURL.

2. Install the jq utility:

   sudo apt install jq -y
   

   You can use it to convert schema descriptions to the required format.

   When using the REST API for Managed Schema Registry, you need to provide schema descriptions as an escaped character string, e.g.:

   "schema": "{\"type\": \"record\", \"name\": \"Obj\", \"fields\":[...]}"
   

   For convenience, in this tutorial, schemas are presented as JSON documents with indentations and line breaks; and when making requests to the API, schemas are converted to the required format using jq.

   Tip

   After you request the REST API using cURL, you get the server response as a single JSON string.

   You can additionally process the output of the commands featured in this tutorial with the help of jq to make the server's response more readable.

Create data format schemasCreate data format schemas

Let's assume an Apache Kafka® message in the messages topic must consist of a key and a value in the following format:

{
  "id": <int>,
  "sid": "<string>"
}

{
  "name": "<string>",
  "city": "<string>",
  "age": <int>
}

Create the relevant data format schemas:

1. Create a file named schema-key.json containing the data format schema for the Apache Kafka® message key.

   schema-key.json

   {
     "type": "record",
     "name": "my_key",
     "fields": [
       {
         "name": "id",
         "type": "int"
       },
       {
         "name": "sid",
         "type": "string"
       }
     ]
   }
   

2. Create a data format schema for the Apache Kafka® message key.

   The subject name for the schema must consist of the name of the topic the schema will be used in (messages) and the -key suffix.

   Use the POST /subjects/(subject)/versions REST API method for Managed Schema Registry and send the following request:

   jq \
       -n --slurpfile data schema-key.json \
       '{
          "schemaType": "AVRO",
          "schema": "\($data)"
       }' \
   | curl \
         --request POST \
         --url 'https://<broker_host_FQDN>:443/subjects/messages-key/versions' \
         --user user1:<user_password> \
         --header 'Content-Type: application/vnd.schemaregistry.v1+json' \
         --data "@-"
   

   The response to the request will return the new schema ID, e.g., {"id":1}.

3. Create a schema-value.json file containing the data format schema for the Apache Kafka® message value.

   schema-value.json

   {
     "type": "record",
     "name": "my_value",
     "fields": [
       {
         "name": "name",
         "type": "string"
       },
       {
         "name": "city",
         "type": "string"
       },
       {
         "name": "age",
         "type": "int"
       }
     ]
   }
   

4. Create a data format schema for the Apache Kafka® message value.

   The subject name for the schema must consist of the name of the topic the schema will be used in (messages) and the -value suffix.

   Use the POST /subjects/(subject)/versions REST API method for Managed Schema Registry and send the following request:

   jq \
       -n --slurpfile data schema-value.json \
       '{
          "schemaType": "AVRO",
          "schema": "\($data)"
       }' \
   | curl \
         --request POST \
         --url 'https://<broker_host_FQDN>:443/subjects/messages-value/versions' \
         --user user1:<user_password> \
         --header 'Content-Type: application/vnd.schemaregistry.v1+json' \
         --data "@-"
   

   The response to the request will return the new schema ID, e.g., {"id":2}.

Send messages to a topicSend messages to a topic

1. Get the data format schema IDs for the key and the value.

   Use the GET /schemas REST API method for Managed Schema Registry and send the following request:

   curl \
       --request GET \
       --url 'https://<broker_host_FQDN>:443/schemas' \
       --user user1:<user_password> \
       --header 'Accept: application/vnd.schemaregistry.v1+json'
   

   The response to the request contains data format schema IDs (id). These IDs will be used later.

   Sample response

   For brevity, the data format schema named schema in the form of JSON strings is not provided.

   [
     {
       "id": 1,
       "schema": "<data_format_schema>",
       "schemaType": "AVRO",
       "subject": "messages-key",
       "version": 1
     },
     {
       "id": 2,
       "schema": "<data_format_schema>",
       "schemaType": "AVRO",
       "subject": "messages-value",
       "version": 1
     }
   ]
   

2. Create a file named message-list.json containing two messages. For each message, the key and the value are specified according to the data format schemas created earlier.

   message-list.json

   [
     {
       "key": {
         "id": 1111,
         "sid": "AAAAA-BBBBB-CCCCC"
       },
       "value": {
         "name": "Anna",
         "city": "Moscow",
         "age": 44
       }
     },
     {
       "key": {
         "id": 2222,
         "sid": "DDDDD-EEEEE-FFFFF"
       },
       "value": {
         "name": "Alex",
         "city": "London",
         "age": 32
       }
     }
   ]
   

3. Send messages to the messages topic.

   Use the POST /topics/(topic) REST API method for Apache Kafka® and send the following request:

   jq \
       -n --slurpfile data message-list.json \
       '{
         "key_schema_id": <messages-key_schema_ID>,
         "value_schema_id": <messages-value_schema_ID>,
         "records": $data.[]
       }' \
   | curl \
         --request POST \
         --url 'https://<broker_host_FQDN>:443/topics/messages' \
         --user user1:<user_password> \
         --header 'Content-Type: application/vnd.kafka.avro.v2+json' \
         --header 'Accept: application/vnd.kafka.v2+json' \
         --data "@-"
   

   The schema ID values ​​were obtained earlier by requesting the GET /schemas endpoint.

   Example of response to query

   {
     "key_schema_id": 1,
     "offsets": [
       {
         "offset": 0,
         "partition": 0
       },
       {
         "offset": 0,
         "partition": 1
       }
     ],
     "value_schema_id": 2
   }
   

Get messages from a topicGet messages from a topic

1. Create a consumer named my-consumer in the consumer group named my-group.

   Use the POST /consumers/(group) REST API method for Apache Kafka® and send the following request:

   curl \
       --request POST \
       --url 'https://<broker_host_FQDN>:443/consumers/my-group' \
       --user user1:<user_password> \
       --header 'Content-Type: application/vnd.kafka.v2+json' \
       --header 'Accept: application/vnd.kafka.v2+json' \
       --data '{
                 "name": "my-consumer",
                 "format": "avro",
                 "auto.offset.reset": "earliest"
               }'
   

   Example of response to query

   {
     "base_uri": "https://<broker_host_FQDN>:443/consumers/my-group/instances/my-consumer",
     "instance_id": "my-consumer"
   }
   

2. Subscribe to the messages topic for the consumer named my-consumer from the consumer group named my-group.

   Use the POST /consumers/(group)/instances/(instance)/subscription REST API method for Apache Kafka® and send the following request:

   curl \
       --request POST \
       --url 'https://<broker_host_FQDN>:443/consumers/my-group/instances/my-consumer/subscription' \
       --user user1:<user_password> \
       --header 'Content-Type: application/vnd.kafka.v2+json' \
       --header 'Accept: application/vnd.kafka.v2+json' \
       --data '{"topics": ["messages"]}'
   

   The API server does not return a response to this request, only an HTTP status.

3. Get all messages from the messages topic for the consumer named my-consumer from the consumer group named my-group.

   Use the GET /consumers/(group)/instances/(instance)/records REST API method for Apache Kafka® and send the following request:

   curl \
       --request GET \
       --url 'https://<broker_host_FQDN>:443/consumers/my-group/instances/my-consumer/records' \
       --user user1:<user_password> \
       --header 'Accept: application/vnd.kafka.avro.v2+json'
   

   If the response to the request contains messages that were sent earlier, this means the producer and consumer are successfully interpreting the messages in accordance with the specified data format schemas.

   Example of response to query

   [
     {
       "key": {
         "id": 2222,
         "sid": "DDDDD-EEEEE-FFFFF"
       },
       "offset": 0,
       "partition": 1,
       "timestamp": 1726031054186,
       "topic": "messages",
       "value": {
         "age": 32,
         "city": "London",
         "name": "Alex"
       }
     },
     {
       "key": {
         "id": 1111,
         "sid": "AAAAA-BBBBB-CCCCC"
       },
       "offset": 0,
       "partition": 0,
       "timestamp": 1726031054186,
       "topic": "messages",
       "value": {
         "age": 44,
         "city": "Moscow",
         "name": "Anna"
       }
     }
   ]
   

Delete the resources you createdDelete the resources you created

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

• Delete the Managed Service for Apache Kafka® cluster.
• If you reserved public static IP addresses, release and delete them.