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®:
- Create data format schemas.
- Send messages to a topic.
- Get messages from a topic.
- Delete the resources you created.
Getting started
Prepare the infrastructure
-
Create a Managed Service for Apache Kafka® cluster with any suitable configuration.
When creating a cluster, enable the following options:
-
Schema registry.
The cluster will deploy a Managed Schema Registry schema registry, and the REST API for Managed Schema Registry will become available.
-
Kafka Rest API.
The REST API for Apache Kafka® will become available in the cluster.
-
Public access.
Broker hosts will become available from the internet.
-
-
Create a topic named
messages
for exchanging messages between the producer and the consumer. -
Create a user named
user1
and grant them permissions for themessages
topic:ACCESS_ROLE_CONSUMER
ACCESS_ROLE_PRODUCER
This user will be able to send and receive messages within the topic, as well as perform any operations on subjects in Managed Schema Registry that are associated with the topic.
-
Complete all pre-configuration steps to connect to the cluster.
Install utilities
-
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. -
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 schemas
Note
This tutorial uses the Avro
You can use other types of schemas supported in Managed Schema Registry.
Let's assume an Apache Kafka® message in the messages
topic must consist of a key and a value in the following format:
Key |
Value |
|
|
Create the relevant data format schemas:
-
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" } ] }
-
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}
. -
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" } ] }
-
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 topic
-
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.Example of response to query
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 } ]
-
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 } } ]
-
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 topic
-
Create a consumer named
my-consumer
in the consumer group namedmy-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" }
-
Subscribe to the
messages
topic for the consumer namedmy-consumer
from the consumer group namedmy-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.
-
Get all messages from the
messages
topic for the consumer namedmy-consumer
from the consumer group namedmy-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 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.