[Yandex Cloud documentation](../../../index.md) > [Yandex Managed Service for Apache Kafka®](../../index.md) > [Step-by-step guides](../index.md) > [Connection](index.md) > Connecting from applications

# Connecting to an Apache Kafka® cluster from applications

This section provides settings for connecting to Managed Service for Apache Kafka® cluster hosts using [command line tools](#command-line-tools) and from a [Docker container](#docker). To learn how to connect from your application code, see [Code examples](code-examples.md).

You can only connect to public Apache Kafka® cluster hosts using an [SSL certificate](index.md#get-ssl-cert). The examples below assume that the `YandexInternalRootCA.crt` certificate is located in this directory:

* `/usr/local/share/ca-certificates/Yandex/` for Ubuntu.
* `$HOME\.kafka\` for Windows.


Connecting without an SSL certificate is only supported for non-public hosts. If this is the case, internal virtual network traffic will not be encrypted when connecting to a database.

Before connecting, [configure security groups](index.md#configuring-security-groups) for the cluster, if required.


The examples for Linux were tested in the following environment:

* Yandex Cloud VM running Ubuntu 20.04 LTS.
* OpenJDK: `11.0.24`.
* Bash: `5.0.16`.

The examples for Windows were tested in the following environment:

* Yandex Cloud virtual machine running Windows Server 2019 Datacenter.
* Microsoft OpenJDK: `11.0.11`.
* PowerShell: `5.1.17763.1490 Desktop`.

## Command line tools {#command-line-tools}

To see code examples with the host FQDN filled in, open the cluster page in the [management console](https://console.yandex.cloud) and click **Connect**.

### kafkactl {#kafkactl}

`kafkactl` is a command line interface for Apache Kafka®. It enables you to create, update, and delete resources in a Apache Kafka® cluster and get information, e.g., on brokers, topics, messages, etc. The utility supports both message consumer and producer workflows.

Before connecting, install `kafkactl`:

```bash
wget https://github.com/deviceinsight/kafkactl/releases/download/v<version_number>/<archive_name>.tar.gz
tar xzf <archive_name>.tar.gz kafkactl
sudo mv kafkactl /usr/local/bin
```

You can look up the archive version and name on the [`kafkactl` releases page](https://github.com/deviceinsight/kafkactl/releases).

For other `kafkactl` installation options, see [this `kafkactl` guide](https://github.com/deviceinsight/kafkactl#installation).

{% list tabs group=connection %}

- Connecting without SSL {#without-ssl}

  1. Create a folder named `~/.config/kafkactl` and place the `config.yml` file with your Apache Kafka® cluster connection parameters in that folder:

      ```bash
      cd ~/ && \
      mkdir --parents .config/kafkactl && \
      cd ~/.config/kafkactl
      ```

      Here is an example of the `config.yml` file:
      
      ```yaml
      contexts:
        default:
          brokers:
          - <broker_FQDN>:9092
          sasl:
            enabled: true
            username: <consumer_login>
            password: <consumer_password>
            mechanism: scram-sha512
      ```

      To learn how to get a broker host FQDN, see [this guide](index.md#get-fqdn).

  1. Run the following command to receive messages from the topic:

      ```bash
      kafkactl consume <topic_name>
      ```

     This command will continuously read new messages from the topic.

  1. In a separate terminal, run the following command to send a message to the topic:

      ```bash
      echo "test message" | kafkactl produce <topic_name>
      ```

- Connecting with SSL {#with-ssl}

  1. Create a folder named `~/.config/kafkactl` and place the `config.yml` file with your Apache Kafka® cluster connection parameters in that folder:

      ```bash
      cd ~/ && \
      mkdir --parents .config/kafkactl && \
      cd ~/.config/kafkactl
      ```

      Here is an example of the `config.yml` file:
      
      ```yaml
      contexts:
        default:
          brokers:
          - <broker_FQDN>:9091
          sasl:
            enabled: true
            username: <consumer_login>
            password: <consumer_password>
            mechanism: scram-sha512
          tls:
            enabled: true
            ca: /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
      ```

      To learn how to get a broker host FQDN, see [this guide](index.md#get-fqdn).

  1. Run the following command to receive messages from the topic:

      ```bash
      kafkactl consume <topic_name>
      ```

     This command will continuously read new messages from the topic.

  1. In a separate terminal, run the following command to send a message to the topic:

      ```bash
      echo "test message" | kafkactl produce <topic_name>
      ```

{% endlist %}

Make sure the first terminal displays `test message` sent in the second terminal.

### kafkacat {#bash-zsh}

{% note info %}

`kafkacat` is a deprecated tool that only works with Apache Kafka® cluster versions below 4.0. For clusters running Apache Kafka® 4.0 or later, use [kafkactl](clients.md#kafkactl).

{% endnote %}

[kafkacat](https://github.com/edenhill/kcat), or `kcat`, is an open source tool for producing and consuming data without installing Java Runtime Environment.

Before connecting, install the required dependencies:

```bash
sudo apt update && sudo apt install -y kafkacat
```

{% note info %}

On Ubuntu 24.04 or higher, use `kcat`.

{% endnote %}

{% list tabs group=connection %}

- Connecting without SSL {#without-ssl}

  1. Run the following command to receive messages from the topic:

      ```bash
      kafkacat -C \
               -b <broker_FQDN>:9092 \
               -t <topic_name> \
               -X security.protocol=SASL_PLAINTEXT \
               -X sasl.mechanism=SCRAM-SHA-512 \
               -X sasl.username="<consumer_login>" \
               -X sasl.password="<consumer_password>" -Z
      ```

     This command will continuously read new messages from the topic.

  1. In a separate terminal, run the following command to send a message to the topic:

      ```bash
      echo "test message" | kafkacat -P \
             -b <broker_FQDN>:9092 \
             -t <topic_name> \
             -k key \
             -X security.protocol=SASL_PLAINTEXT \
             -X sasl.mechanism=SCRAM-SHA-512 \
             -X sasl.username="<producer_login>" \
             -X sasl.password="<producer_password>" -Z
      ```

- Connecting with SSL {#with-ssl}

  1. Run the following command to receive messages from the topic:

     ```bash
     kafkacat -C \
              -b <broker_FQDN>:9091 \
              -t <topic_name> \
              -X security.protocol=SASL_SSL \
              -X sasl.mechanism=SCRAM-SHA-512 \
              -X sasl.username="<consumer_username>" \
              -X sasl.password="<consumer_password>" \
              -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt -Z -K:
     ```

     This command will continuously read new messages from the topic.

  1. In a separate terminal, run the following command to send a message to the topic:

     ```bash
     echo "test message" | kafkacat -P \
         -b <broker_FQDN>:9091 \
         -t <topic_name> \
         -k key \
         -X security.protocol=SASL_SSL \
         -X sasl.mechanism=SCRAM-SHA-512 \
         -X sasl.username="<producer_login>" \
         -X sasl.password="<producer_password>" \
         -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt -Z
     ```

{% endlist %}

To learn how to get a broker host FQDN, see [this guide](index.md#get-fqdn).

Make sure the first terminal displays `test message` sent in the second terminal.

### Apache Kafka® tools for Linux (Bash)/macOS (Zsh) {#kafka-sh}

[Archives with Apache Kafka® binary files](https://kafka.apache.org/community/downloads/) include tools that allow managing the Apache Kafka® cluster and its entities. For more information about these tools, see [this Apache Kafka® vendor guide](https://docs.confluent.io/kafka/operations-tools/kafka-tools.html).

The example below shows how to specify user credentials for connection and use the tools with these credentials:

* A message will be sent to the topic using [kafka-console-producer](https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-console-producer-sh).
* A message will be received from the topic using [kafka-console-consumer](https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-console-consumer-sh).

Before connecting:

1. Install OpenJDK:

    ```bash
    sudo apt update && sudo apt install --yes default-jdk
    ```

1. Download the [archive with binary files](https://kafka.apache.org/community/downloads/) for the Apache Kafka® version running in your cluster. Your Scala version is irrelevant.

1. Unpack the archive.

{% list tabs group=connection %}

- Connecting without SSL {#without-ssl}

    1. Create two configuration files to connect to the cluster: one for the producer and one for the consumer.
       
       These files have the same content and differ only in user credentials:
       
       ```ini
       sasl.mechanism=SCRAM-SHA-512
       sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
         username="<producer_or_consumer_login>" \
         password="<producer_or_consumer_password>";
       security.protocol=SASL_PLAINTEXT
       ```

    1. Run the following command to receive messages from the topic:

        ```bash
        <path_to_directory_with_Apache_Kafka_files>/bin/kafka-console-consumer.sh \
          --consumer.config <path_to_file_with_consumer_configuration> \
          --bootstrap-server <broker_FQDN>:9092 \
          --topic <topic_name> \
          --property print.key=true \
          --property key.separator=":"
        ```

        This command will continuously read new messages from the topic.

    1. In a separate terminal, run the following command to send a message to the topic:

        ```bash
        echo "key:test message" | <path_to_directory_with_Apache_Kafka_files>/bin/kafka-console-producer.sh \
          --producer.config <path_to_file_with_producer_configuration> \
          --bootstrap-server <broker_FQDN>:9092 \
          --topic <topic_name> \
          --property parse.key=true \
          --property key.separator=":"
        ```

- Connecting with SSL {#with-ssl}

    1. Go to the folder where the Java certificate store will reside:

        ```bash
        cd /etc/security
        ```

    1. Add the SSL certificate to the Java trusted certificate store (Java Key Store) so that the Apache Kafka® driver can use this certificate for secure connections to the cluster hosts. Set a password of at least 6 characters using the `-storepass` parameter for additional storage protection:
       
       ```bash
       sudo keytool -importcert \
                    -alias YandexCA -file /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
                    -keystore ssl -storepass <certificate_store_password> \
                    --noprompt
       ```

    1. Create two configuration files to connect to the cluster: one for the producer and one for the consumer.

        These files have the same content and differ only in user credentials:

        ```ini
        sasl.mechanism=SCRAM-SHA-512
        sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
          username="<producer_or_consumer_login>" \
          password="<producer_or_consumer_password>";
        security.protocol=SASL_SSL
        ssl.truststore.location=/etc/security/ssl
        ssl.truststore.password=<certificate_store_password>
        ```

    1. Run the following command to receive messages from the topic:

        ```bash
        <path_to_directory_with_Apache_Kafka_files>/bin/kafka-console-consumer.sh \
          --consumer.config <path_to_file_with_consumer_configuration> \
          --bootstrap-server <broker_FQDN>:9091 \
          --topic <topic_name> \
          --property print.key=true \
          --property key.separator=":"
        ```

        This command will continuously read new messages from the topic.

    1. In a separate terminal, run the following command to send a message to the topic:

        ```bash
        echo "key:test message" | <path_to_directory_with_Apache_Kafka_files>/bin/kafka-console-producer.sh \
          --producer.config <path_to_file_with_producer_configuration> \
          --bootstrap-server <broker_FQDN>:9091 \
          --topic <topic_name> \
          --property parse.key=true \
          --property key.separator=":"
        ```

{% endlist %}

To learn how to get a broker host FQDN, see [this guide](index.md#get-fqdn).

Make sure the first terminal displays `test message` sent in the second terminal.

### Apache Kafka® tools for Windows (PowerShell) {#powershell}

[Archives with Apache Kafka® binary files](https://kafka.apache.org/community/downloads/) include tools that allow managing the Apache Kafka® cluster and its entities. For more information about these tools, see [this Apache Kafka® vendor guide](https://docs.confluent.io/kafka/operations-tools/kafka-tools.html).

The example below shows how to specify user credentials for connection and use the tools with these credentials:

* A message will be sent to the topic using [kafka-console-producer](https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-console-producer-sh).
* A message will be received from the topic using [kafka-console-consumer](https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-console-consumer-sh).

While mentioning `.sh` scripts, the documentation for the tools is relevant for Windows as well. The tools are the same across platforms, and only the scripts that run them differ, as shown below:

* `bin/kafka-console-producer.sh` for Linux (Bash)/macOS (Zsh).
* `bin\windows\kafka-console-producer.bat` for Windows (PowerShell).

Before connecting:

1. Install the latest available version of [Microsoft OpenJDK](https://docs.microsoft.com/en-us/java/openjdk/download).

1. Download the [archive with binary files](https://kafka.apache.org/community/downloads/) for the Apache Kafka® version running in your cluster. Your Scala version is irrelevant.

1. Unpack the archive.

   {% note tip %}

   Unpack the Apache Kafka® files to the disk root folder, e.g., `C:\kafka_2.12-2.6.0\`.

   If the path to the Apache Kafka® executables and batch files is too long, you will get the `The input line is too long` error when trying to run them.

   {% endnote %}

{% list tabs group=connection %}

- Connecting without SSL {#without-ssl}

  1. Create two configuration files to connect to the cluster: one for the producer and one for the consumer.
     
     These files have the same content and differ only in user credentials:
     
     ```ini
     sasl.mechanism=SCRAM-SHA-512
     sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
       username="<producer_or_consumer_login>" \
       password="<producer_or_consumer_password>";
     security.protocol=SASL_PLAINTEXT
     ```

  1. Run the following command to receive messages from the topic:

      ```powershell
      <path_to_directory_with_Apache_Kafka_files>\bin\windows\kafka-console-consumer.bat `
          --consumer.config <path_to_file_with_consumer_configuration> `
          --bootstrap-server <broker_FQDN>:9092 `
          --topic <topic_name> `
          --property print.key=true `
          --property key.separator=":"
      ```

     This command will continuously read new messages from the topic.

  1. In a separate terminal, run the following command to send a message to the topic:

      ```powershell
      echo "key:test message" | <path_to_directory_with_Apache_Kafka_files>\bin\windows\kafka-console-producer.bat `
          --producer.config <path_to_file_with_producer_configuration> `
          --bootstrap-server <broker_FQDN>:9092 `
          --topic <topic_name> `
          --property parse.key=true `
          --property key.separator=":"
      ```

- Connecting with SSL {#with-ssl}

  1. Add the SSL certificate to the Java trusted certificate store (Java Key Store) so that the Apache Kafka® driver can use this certificate for secure connections to the cluster hosts. Set a password in the `--storepass` parameter for additional storage protection:

     ```powershell
     keytool.exe -importcert -alias YandexCA `
       --file $HOME\.kafka\YandexInternalRootCA.crt `
       --keystore $HOME\.kafka\ssl `
       --storepass <certificate_store_password> `
       --noprompt
     ```

  1. Create two configuration files to connect to the cluster: one for the producer and one for the consumer.

     These files have the same content and differ only in user credentials:

     ```ini
     sasl.mechanism=SCRAM-SHA-512
     sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
       username="<producer_or_consumer_login>" \
       password="<producer_or_consumer_password>";
     security.protocol=SASL_SSL
     ssl.truststore.location=<$HOME_variable_value>\\.kafka\\ssl
     ssl.truststore.password=<certificate_store_password>
     ```

     Specify the full path to the certificate store as the `ssl.truststore.location` parameter value. Here is an example:

     ```ini
     ssl.truststore.location=C:\\Users\\Administrator\\.kafka\\ssl
     ```

     The certificate store is located at `$HOME\.kafka\ssl`, but you cannot use environment variables in the value. To expand the variable, run this command:

     ```powershell
     echo $HOME
     ```

     {% note warning %}

     Use `\\` instead of `\` when specifying the `ssl.truststore.location` parameter value, otherwise you will not be able to access the certificate store when running commands.

     {% endnote %}

  1. Run the following command to receive messages from the topic:

      ```powershell
      <path_to_directory_with_Apache_Kafka_files>\bin\windows\kafka-console-consumer.bat `
          --consumer.config <path_to_file_with_consumer_configuration> `
          --bootstrap-server <broker_FQDN>:9091 `
          --topic <topic_name> `
          --property print.key=true `
          --property key.separator=":"
      ```

     This command will continuously read new messages from the topic.

  1. In a separate terminal, run the following command to send a message to the topic:

      ```powershell
      echo "key:test message" | <path_to_directory_with_Apache_Kafka_files>\bin\windows\kafka-console-producer.bat `
          --producer.config <path_to_file_with_producer_configuration> `
          --bootstrap-server <broker_FQDN>:9091 `
          --topic <topic_name> `
          --property parse.key=true `
          --property key.separator=":"
      ```

{% endlist %}

To learn how to get a broker host FQDN, see [this guide](index.md#get-fqdn).

Make sure the first terminal displays `test message` sent in the second terminal.

## Before you connect from a Docker container {#docker}

### Connecting with kafkactl {docker-kafkactl}

To connect to a Managed Service for Apache Kafka® cluster from a Docker container using `kafkactl`:

{% list tabs group=connection %}


- Connecting without SSL {#without-ssl}

  1. In the folder housing your Dockerfile, create a configuration file named `config.yml`.

      Here is an example of the `config.yml` file:
      
      ```yaml
      contexts:
        default:
          brokers:
          - <broker_FQDN>:9092
          sasl:
            enabled: true
            username: <consumer_login>
            password: <consumer_password>
            mechanism: scram-sha512
      ```

  1. Add the following lines to the Dockerfile:

      ```bash
      COPY config.yml $HOME/.config/kafkactl/config.yml
      RUN apt-get update && \
          apt-get install -y wget && \
          wget https://github.com/deviceinsight/kafkactl/releases/download/v<version_number>/<archive_name>.tar.gz && \
          tar xzf <archive_name>.tar.gz kafkactl && \
          mv kafkactl /usr/local/bin
      ```

      You can look up the archive version and name on the [`kafkactl` releases page](https://github.com/deviceinsight/kafkactl/releases).


- Connecting with SSL {#with-ssl}

  1. In the folder housing your Dockerfile, create a configuration file named `config.yml`.

      Here is an example of the `config.yml` file:
      
      ```yaml
      contexts:
        default:
          brokers:
          - <broker_FQDN>:9091
          sasl:
            enabled: true
            username: <consumer_login>
            password: <consumer_password>
            mechanism: scram-sha512
          tls:
            enabled: true
            ca: /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
      ```

  1. Add the following lines to the Dockerfile:

      ```bash
      COPY config.yml $HOME/.config/kafkactl/config.yml
      RUN apt-get update && \
          apt-get install -y wget && \
          wget https://github.com/deviceinsight/kafkactl/releases/download/v<version_number>/<archive_name>.tar.gz && \
          tar xzf <archive_name>.tar.gz kafkactl && \
          mv kafkactl /usr/local/bin && \
          mkdir --parents /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 && \
          chmod 0655 /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
      ```

      You can look up the archive version and name on the [`kafkactl` releases page](https://github.com/deviceinsight/kafkactl/releases).

{% endlist %}

### Connecting with kafkacat {docker-kafkacat}

{% note info %}

`kafkacat` is a deprecated tool that only works with Apache Kafka® cluster versions below 4.0. For clusters running Apache Kafka® 4.0 or later, use [kafkactl](clients.md#kafkactl).

{% endnote %}

To connect to a Managed Service for Apache Kafka® cluster from a Docker container using `kafkacat`, add the following lines to the Dockerfile:

{% list tabs group=connection %}


- Connecting without SSL {#without-ssl}

  ```bash
  RUN apt-get update && \
      apt-get install kafkacat --yes
  ```


- Connecting with SSL {#with-ssl}

  ```bash
  RUN apt-get update && \
      apt-get install wget kafkacat --yes && \
      mkdir --parents /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 && \
      chmod 0655 /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt
  ```

{% endlist %}