Contact UsGet started
© 2025 Direct Cursus Technology L.L.C.

Configuring Kafka Connect for Managed Service for Apache Kafka® clusters

Written by
Updated at May 5, 2025

Note

Managed Service for Apache Kafka® has built-in support for certain connectors and allows you to manage them. For a list of available connectors, see Connectors. If you need other connectors or want to manage Kafka Connect manually, refer to this tutorial.

Kafka Connect is a tool for streaming data between Apache Kafka® and other data storages.

Data in Kafka Connect is handled using processes called workers. You can deploy the tool either in distributed mode with multiple workers or standalone mode with a single worker.

Data is moved using connectors that are run in separate worker threads.

To learn more about Kafka Connect, see the documentation Apache Kafka®.

Next, we will configure Kafka Connect to interact with a Managed Service for Apache Kafka® cluster. The tool will be deployed on a Yandex Cloud VM as a separate installation. SSL encryption will be used to protect the connection.

We will also set up a simple FileStreamSource connector. Kafka Connect will use it to read data from a test JSON file and provide this data to a cluster topic.

Note

You can use any other Kafka Connect connector to interact with Managed Service for Apache Kafka® clusters.

To configure Kafka Connect to work with a Managed Service for Apache Kafka® cluster:

  1. Configure the VM.
  2. Prepare the test data.
  3. Configure Kafka Connect.
  4. Run Kafka Connect and test it.

If you no longer need the resources you created, delete them.

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 if public access is enabled for cluster hosts (see Virtual Private Cloud pricing).
  • VM fee: using computing resources, storage, and public IP address (see Compute Cloud pricing).

Getting started

  1. Create a Managed Service for Apache Kafka® cluster with any suitable configuration.

  2. Create a topic named messages for exchanging messages between Kafka Connect and the Managed Service for Apache Kafka® cluster.

  3. Create a user named user and grant them permissions for the messages topic:

    • ACCESS_ROLE_CONSUMER
    • ACCESS_ROLE_PRODUCER

  4. In the network hosting the Managed Service for Apache Kafka® cluster, create a virtual machine with Ubuntu 20.04 and a public IP address.

  5. If you are using security groups, configure them to allow all required traffic between the Managed Service for Apache Kafka® cluster and the VM.

  1. If you do not have Terraform yet, install it.

  2. Get the authentication credentials. You can add them to environment variables or specify them later in the provider configuration file.

  3. Configure and initialize a provider. There is no need to create a provider configuration file manually, you can download it.

  4. Place the configuration file in a separate working directory and specify the parameter values. If you did not add the authentication credentials to environment variables, specify them in the configuration file.

  5. Download the kafka-connect.tf configuration file to the same working directory.

    This file describes:

    • Network.

    • Subnet.

    • Default security group and rules required to connect to the cluster and VM from the internet.

    • Virtual machine with Ubuntu 20.04.

    • Properly configured Managed Service for Apache Kafka® cluster.

  6. In the file, specify the password for the user named user you are going to use to access the Managed Service for Apache Kafka® cluster, as well as the username and the public part of the SSH key for the virtual machine. If the virtual machine will be running Ubuntu 20.04 from the recommended image list, the username you put here will be ignored. In which case use ubuntu for username to establish the connection.

  7. Make sure the Terraform configuration files are correct using this command:

    terraform validate

    If there are any errors in the configuration files, Terraform will point them out.

  8. Create the required infrastructure:

    1. Run this command to view the planned changes:

      terraform plan

      If you described the configuration correctly, the terminal will display a list of the resources to update and their parameters. This is a verification step that does not apply changes to your resources.

    2. If everything looks correct, apply the changes:

      1. Run this command:

        terraform apply

      2. Confirm updating the resources.

      3. Wait for the operation to complete.

    All the required resources will be created in the specified folder. You can check resource availability and their settings in the management console.

Configure the VM

  1. Connect to the virtual machine over SSH.

  2. Install JDK and the kcat utility:

    sudo apt update && \
sudo apt install default-jdk --yes && \
sudo apt install kafkacat

    Check that you can use it to connect to the Managed Service for Apache Kafka® source cluster over SSL.

  3. Download and unpack the archive containing Apache Kafka®:

    wget https://downloads.apache.org/kafka/3.1.0/kafka_2.12-3.1.0.tgz && tar -xvf kafka_2.12-3.1.0.tgz --strip 1 --directory /opt/kafka/

    This example uses Apache Kafka® version 3.1.0.

  4. Get an SSL certificate.

  5. 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:

    sudo keytool -importcert \
             -alias YandexCA -file /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
             -keystore ssl -storepass <certificate_store_password> \
             --noprompt

  6. Create a folder with worker settings and copy the store there:

    sudo mkdir --parents /etc/kafka-connect-worker && \
sudo cp ssl /etc/kafka-connect-worker/client.truststore.jks

Prepare the test data

Create a file named /var/log/sample.json with test data. This file contains data from car sensors in JSON format:

sample.json
{"device_id":"iv9a94th6rzt********","datetime":"2020-06-05 17:27:00","latitude":55.70329032,"longitude":37.65472196,"altitude":427.5,"speed":0,"battery_voltage":23.5,"cabin_temperature":17,"fuel_level":null}
{"device_id":"rhibbh3y08qm********","datetime":"2020-06-06 09:49:54","latitude":55.71294467,"longitude":37.66542005,"altitude":429.13,"speed":55.5,"battery_voltage":null,"cabin_temperature":18,"fuel_level":32}
{"device_id":"iv9a94th6rzt********","datetime":"2020-06-07 15:00:10","latitude":55.70985913,"longitude":37.62141918,"altitude":417,"speed":15.7,"battery_voltage":10.3,"cabin_temperature":17,"fuel_level":null}

Configure Kafka Connect

  1. Create a file named /etc/kafka-connect-worker/worker.properties with worker settings:

    # AdminAPI connect properties
bootstrap.servers=<broker_host_FQDN>:9091
sasl.mechanism=SCRAM-SHA-512
security.protocol=SASL_SSL
ssl.truststore.location=/etc/kafka-connect-worker/client.truststore.jks
ssl.truststore.password=<certificate_storage_password>
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="user" password="<user_password>";

# Producer connect properties
producer.sasl.mechanism=SCRAM-SHA-512
producer.security.protocol=SASL_SSL
producer.ssl.truststore.location=/etc/kafka-connect-worker/client.truststore.jks
producer.ssl.truststore.password=<certificate_storage_password>
producer.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="user" password="<user_password>";

# Worker properties
plugin.path=/etc/kafka-connect-worker/plugins
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.file.filename=/etc/kafka-connect-worker/worker.offset

    Kafka Connect will connect to the Managed Service for Apache Kafka® cluster as the user named user created earlier.

    You can request the FQDNs of broker hosts with the list of cluster hosts.

  2. Create a file named /etc/kafka-connect-worker/file-connector.properties with connector settings:

    name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=/var/log/sample.json
topic=messages

    Where:

    • file: Name of the file the connector will read data from.
    • topic: Name of the Managed Service for Apache Kafka® cluster topic the connector will feed data to.

Run Kafka Connect and test it

  1. To send test data to the cluster, run the worker on the VM:

    cd ~/opt/kafka/bin/ && \
sudo ./connect-standalone.sh \
     /etc/kafka-connect-worker/worker.properties \
     /etc/kafka-connect-worker/file-connector.properties

  2. Connect to the cluster using kcat and retrieve data from the cluster topic:

    kafkacat -C \
    -b <broker_host_FQDN>:9091 \
    -t messages \
    -X security.protocol=SASL_SSL \
    -X sasl.mechanisms=SCRAM-SHA-512 \
    -X sasl.username=user \
    -X sasl.password="<user_account_password>" \
    -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt -Z -K:

    You can request the FQDNs of broker hosts with the list of cluster hosts.

    In the command output, you will see the contents of the /var/log/sample.json test file provided in the previous step.

Delete the resources you created

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

  1. Delete the VM.
  2. If you reserved a public static IP address for the VM, delete it.
  3. Delete the Managed Service for Apache Kafka® cluster.

  1. In the terminal window, go to the directory containing the infrastructure plan.

    Warning

    Make sure the directory has no Terraform manifests with the resources you want to keep. Terraform deletes all resources that were created using the manifests in the current directory.

  2. Delete resources:

    1. Run this command:

      terraform destroy

    2. Confirm deleting the resources and wait for the operation to complete.

    All the resources described in the Terraform manifests will be deleted.

Previous
Deploying the Apache Kafka® web interface
Next
Migrating a database from a third-party Apache Kafka® cluster