Аутентификация и соединение с БД по Kafka API

ЭндпоинтЭндпоинт

Эндпоинт Kafka API отображается в консоли управления, на странице потока данных, на вкладке Обзор, в поле Kafka API эндпоинт.

Эндпоинт имеет следующий вид: <FQDN_YDB>:PORT. Например, ydb-01.serverless.yandexcloud.net:9093.

Предварительные требованияПредварительные требования

Для аутентификации требуется:

1. Создать сервисный аккаунт.
2. Назначить сервисному аккаунту роли:
   • Для чтения из потока данных: ydb.kafkaApi.client и ydb.viewer.
   • Для записи в поток данных: ydb.kafkaApi.client и ydb.editor.
3. Создать API-ключ c областью действия yc.ydb.topics.manage.

АутентификацияАутентификация

В Kafka API аутентификация выполняется через механизм SASL_SSL/PLAIN.

Для этого необходимы:

• <database> — путь базы данных. Путь базы данных отображается в консоли управления, на странице потока данных, на вкладке Обзор, в поле Эндпоинт после database=.

  Например, если эндпоинт — grpcs://ydb.serverless.yandexcloud.net:2135/?database=/ru-central1/b1gia87mbaom********/etnudu2n9ri3********, то путь базы данных — /ru-central1/b1gia87mbaom********/etnudu2n9ri3********.

• <api-key> — API-ключ.

Эти параметры будут использоваться для аутентификации при чтении и записи сообщений:

• <sasl.username> = @<database> (обратите внимание, что перед путем к базе данных необходимо поставить символ @)
• <sasl.password> = <api-key>

Пример записи и чтения сообщенияПример записи и чтения сообщения

В примере используются:

• <kafka-api-endpoint> — эндпоинт.
• <stream-name> — имя потока данных.

1. Установите SSL-сертификат, если используете Dedicated базу:

    sudo mkdir -p /usr/local/share/ca-certificates/Yandex/ && \
    wget "https://storage.yandexcloud.kz/cloud-certs/CA.pem" \
     --output-document /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt && \
    sudo chmod 0655 /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
   

   Сертификат будет сохранен в файле /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt.

2. Установите kcat — приложение с открытым исходным кодом, которое может работать как универсальный производитель или потребитель данных:

   sudo apt-get install kafkacat
   

3. Запустите команду получения сообщений из потока:

   Serverless база

   Dedicated база

   kcat -C \
     -b <kafka-api-endpoint> \
     -t <stream-name> \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanism=PLAIN \
     -X sasl.username="<sasl.username>" \
     -X sasl.password="<sasl.password>"
   

   kcat -C \
     -b <kafka-api-endpoint> \
     -t <stream-name> \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanism=PLAIN \
     -X sasl.username="<sasl.username>" \
     -X sasl.password="<sasl.password>" \
     -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
   

   Команда будет непрерывно считывать новые сообщения из потока.

4. В отдельном терминале запустите команду отправки сообщения в поток:

   Serverless база

   Dedicated база

   echo "test message" | kcat -P \
       -b <kafka-api-endpoint> \
       -t <stream-name> \
       -k key \
       -X security.protocol=SASL_SSL \
       -X sasl.mechanism=PLAIN \
       -X sasl.username="<sasl.username>" \
       -X sasl.password="<sasl.password>"
   

   echo "test message" | kcat -P \
       -b <kafka-api-endpoint> \
       -t <stream-name> \
       -k key \
       -X security.protocol=SASL_SSL \
       -X sasl.mechanism=PLAIN \
       -X sasl.username="<sasl.username>" \
       -X sasl.password="<sasl.password>" \
       -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
   

Основную документацию по работе с Data Streams через Kafka API и больше примеров см. в документации YDB.