Authentication and database connection using the Kafka API

EndpointEndpoint

The Kafka API endpoint appears in the management console under: Data Streams → [Your Stream] → Overview → Kafka API endpoint.

The endpoint has the following format: <FQDN_YDB>:PORT. For example, ydb-01.serverless.yandexcloud.net:9093.

PrerequisitesPrerequisites

To authenticate, take these steps:

1. Create a service account.
2. Assign the following roles to the service account:
   • ydb.kafkaApi.client and ydb.viewer: for reading from a data stream.
   • ydb.kafkaApi.client and ydb.editor: for writing to a data stream.
3. Create an API key with the yc.ydb.topics.manage scope.

AuthenticationAuthentication

The Kafka API uses the SASL_SSL/PLAIN authentication mechanism.

The following parameters are required:

• <database>: Database path. The database path appears in the management console under: Data Streams → [Your Stream] → Overview → Endpoint (a substring following database=).

  For example, if the Endpoint field contains grpcs://ydb.serverless.yandexcloud.net:2135/?database=/ru-central1/b1gia87mbaomkfvs6rgl/etnudu2n9ri35luqe4h1, the database path is /ru-central1/b1gia87mbaomkfvs6rgl/etnudu2n9ri35luqe4h1.

• <api-key>: API key.

These parameters will be used for authentication when reading and writing messages:

• <sasl.username> = @<database> (Note that the database path must be prefixed with the @ symbol)
• <sasl.password> = <api-key>

Example of writing and reading a messageExample of writing and reading a message

This example uses the following parameters:

• <kafka-api-endpoint>: Endpoint.
• <stream-name>: Data stream name.

1. If you are using a dedicated database, you need to Install an SSL certificate:

    sudo mkdir -p /usr/local/share/ca-certificates/Yandex/ && \
    wget "https://storage.yandexcloud.net/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
   

   The certificate will be saved to the /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt file.

2. Install kcat, an open-source tool for producing and consuming data:

   sudo apt-get install kafkacat
   

3. Run the following command to get messages from the stream:

   Serverless database

   Dedicated database

   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
   

   This command will continuously read new messages from the stream.

4. In a separate terminal, run the following command to send a message to the stream:

   Serverless database

   Dedicated database

   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
   

For details on working with Data Streams via the Kafka API and more examples, refer to the YDB documentation.