Configuring Kafka Connect for Yandex Managed Service for Apache Kafka® clusters

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.

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.

Required paid resourcesRequired paid resources

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 startedGetting started

Configure the VMConfigure 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 dataPrepare the test data

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

{"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 ConnectConfigure 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 itRun 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 createdDelete the resources you created

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