[Yandex Cloud documentation](../../index.md) > [Yandex Managed Service for Apache Kafka®](../index.md) > [Tutorials](index.md) > Moving data from Apache Kafka® > Configuring Kafka Connect to work with Managed Service for Apache Kafka®

# Configuring Kafka Connect to work with a Managed Service for Apache Kafka® cluster


{% note info %}

Managed Service for Apache Kafka® has native support for certain connectors and enable you to manage them. For a list of available connectors, see [Connectors](../concepts/connectors.md). If you need other connectors or want to manage Kafka Connect manually, refer to this tutorial.

{% endnote %}

Kafka Connect is designed to move data between Apache Kafka® and other data storages.

Kafka Connect processes data using _workers_. You can deploy the tool either in distributed mode with multiple workers or in standalone mode with a single worker.

_Connectors_ move data while running in separate threads of a worker.

To learn more about Kafka Connect, see [this Apache Kafka® article](https://kafka.apache.org/42/kafka-connect/overview/).

Next, we describe how to configure Kafka Connect to work with a Managed Service for Apache Kafka® cluster. You will deploy Kafka Connect on a [Yandex Cloud VM](../../compute/concepts/vm.md) as a separate installation. To protect the connection, you will use SSL encryption.


You will also set up a simple [FileStreamSource](https://docs.confluent.io/home/connect/filestream_connector.html) connector. Kafka Connect will use it to read data from a test JSON file and provide this data to a cluster topic.

{% note info %}

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

{% endnote %}

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

1. [Configure the VM](#prepare-vm).
1. [Prepare your test data](#prepare-test-data).
1. [Configure Kafka Connect](#configure-kafka-connect).
1. [Run and test Kafka Connect](#test-kafka-connect).

If you no longer need the resources you created, [delete them](#clear-out).


## Required paid resources {#paid-resources}

The support cost for this solution includes:

* Managed Service for Apache Kafka® cluster fee, which covers the use of computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see [Apache Kafka® pricing](../pricing.md)).
* Fee for public IP addresses if public access is enabled for cluster hosts (see [Virtual Private Cloud pricing](../../vpc/pricing.md)).
* VM fee, which covers the use of computing resources, storage, and public IP address (see [Compute Cloud pricing](../../compute/pricing.md)).


## Getting started {#before-you-begin}

{% list tabs group=instructions %}

- Manually {#manual}

    1. [Create a Managed Service for Apache Kafka® cluster](../operations/cluster-create.md) of any suitable configuration.
    1. [Create a topic](../operations/cluster-topics.md#create-topic) named `messages` for exchanging messages between Kafka Connect and the Managed Service for Apache Kafka® cluster.
    1. [Create a user](../operations/cluster-accounts.md#create-account) named `user` and [grant them permissions](../operations/cluster-accounts.md#grant-permission) for the `messages` topic:

        * `ACCESS_ROLE_CONSUMER`
        * `ACCESS_ROLE_PRODUCER`

        
    1. In the network hosting the Managed Service for Apache Kafka® cluster, [create a VM](../../compute/operations/vm-create/create-linux-vm.md) running [Ubuntu 20.04](https://yandex.cloud/en/marketplace/products/yc/ubuntu-20-04-lts) with a public IP address.
    1. If using security groups, [configure them](../operations/connect/index.md#configure-security-groups) to allow all required traffic between your Managed Service for Apache Kafka® cluster and VM.


- Terraform {#tf}

    1. If you do not have Terraform yet, [install it](../../tutorials/infrastructure-management/terraform-quickstart.md#install-terraform).
    1. [Get the authentication credentials](../../tutorials/infrastructure-management/terraform-quickstart.md#get-credentials). You can add them to environment variables or specify them later in the provider configuration file.
    1. [Configure and initialize a provider](../../tutorials/infrastructure-management/terraform-quickstart.md#configure-provider). There is no need to create a provider configuration file manually, you can [download it](https://github.com/yandex-cloud-examples/yc-terraform-provider-settings/blob/main/provider.tf).
    1. Place the configuration file in a separate working directory and [specify the parameter values](../../tutorials/infrastructure-management/terraform-quickstart.md#configure-provider). If you did not add the authentication credentials to environment variables, specify them in the configuration file.

    1. Download the [kafka-connect.tf](https://github.com/yandex-cloud-examples/yc-kafka-connect/blob/main/kafka-connect.tf) configuration file to the same working directory.

        This file describes:

        * Network.
        * Subnet.

        
        * Default security group and inbound internet rules for your cluster and VM.


        * Virtual machine running [Ubuntu 20.04](https://yandex.cloud/en/marketplace/products/yc/ubuntu-20-04-lts).
        * Properly configured Managed Service for Apache Kafka® cluster.

    1. 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 VM runs Ubuntu 20.04 from the recommended [image list](../../compute/operations/images-with-pre-installed-software/get-list.md), the username you put here will be ignored. That being the case, use `ubuntu` as username for the [connection](#prepare-vm).
    1. Validate your Terraform configuration files using this command:

       ```bash
       terraform validate
       ```

       Terraform will display any configuration errors detected in your files.
    1. Create the required infrastructure:

       1. Run this command to view the planned changes:
       
          ```bash
          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.
       
       1. If everything looks correct, apply the changes:
          1. Run this command:
       
             ```bash
             terraform apply
             ```
       
          1. Confirm updating the resources.
          1. 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](https://console.yandex.cloud).

{% endlist %}

## Configure the VM {#prepare-vm}


1. [Connect to the VM over SSH](../../compute/operations/vm-connect/ssh.md).


1. Install the JDK and [kcat](https://docs.confluent.io/platform/current/app-development/kafkacat-usage.html):

    ```bash
    sudo apt update && \
    sudo apt install default-jdk --yes && \
    sudo apt install kafkacat
    ```
    
    Make sure you can use it to [connect to the Managed Service for Apache Kafka® source cluster over SSL](../operations/connect/clients.md#bash-zsh).

1. [Download](https://downloads.apache.org/kafka/) and unpack the archive containing Apache Kafka®:

    ```bash
    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® `3.1.0`.

1. [Get an SSL certificate](../operations/connect/index.md#get-ssl-cert).

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 a folder with worker settings and copy the store into it:

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

## Prepare your test data {#prepare-test-data}

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

{% cut "sample.json" %}

```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}
```

{% endcut %}

## Configure Kafka Connect {#configure-kafka-connect}

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

    ```ini
    # 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](#before-you-begin).

    You can get the broker host FQDNs with the [list of cluster hosts](../operations/cluster-hosts.md).

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

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

    Where:

    * `file`: Name of the file from which the connector will read data.
    * `topic`: Name of the Managed Service for Apache Kafka® cluster topic to which the connector will write data.

## Run and test Kafka Connect {#test-kafka-connect}

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

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

1. Connect to the cluster [using kcat](../operations/connect/clients.md#bash-zsh) and retrieve data from the cluster topic:

    ```bash
    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 get the broker host FQDNs with the [list of cluster hosts](../operations/cluster-hosts.md).

    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 {#clear-out}

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

{% list tabs group=instructions %}

- Manually {#manual}

    
    1. [Delete the VM](../../compute/operations/vm-control/vm-delete.md).
    1. If you reserved a public static IP address for your virtual machine, [delete it](../../vpc/operations/address-delete.md).
    1. [Delete the Managed Service for Apache Kafka® cluster](../operations/cluster-delete.md).


- Terraform {#tf}

    1. In the terminal window, go to the directory containing the infrastructure plan.
    
        {% note 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.
    
        {% endnote %}
    
    1. Delete resources:
    
        1. Run this command:
    
            ```bash
            terraform destroy
            ```
    
        1. Confirm deleting the resources and wait for the operation to complete.
    
        All the resources described in the Terraform manifests will be deleted.

{% endlist %}