Getting Started with the MongoDB Kafka Sink Connector

Complete the Tutorial Setup

Complete the steps in the Kafka Connector Tutorial Setup to start the the Confluent Kafka Connect and MongoDB environment.

Configure the Sink Connector

Create an interactive shell session on the tutorial Docker Container using the following command:

docker exec -it mongo1 /bin/bash

Create a source configuration file called simplesink.json with the following command:

nano simplesink.json

Paste the following configuration information into the file and save your changes:

{
  "name": "mongo-tutorial-sink",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "topics": "Tutorial2.pets",
    "connection.uri": "mongodb://mongo1",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": false,
    "database": "Tutorial2",
    "collection": "pets"
  }
}

Run the following command in the shell to start the sink connector using the configuration file you created:

cx simplesink.json

curl -X POST -H "Content-Type: application/json" -d @simplesink.json http://connect:8083/connectors -w "\n"

Run the following command in the shell to check the status of the connectors:

status

If your sink connector started successfully, you should see the following output:

Kafka topics:
...
The status of the connectors:
sink  |  mongo-tutorial-sink  |  RUNNING  |  RUNNING  |  com.mongodb.kafka.connect.MongoSinkConnector
Currently configured connectors
[
"mongo-tutorial-sink"
]
...

Write Data to a Kafka Topic

In the same shell, create a Python script to write data to a Kafka topic.

nano kafkawrite.py

Paste the following code into the file and save your changes:

from kafka import KafkaProducer
import json
from json import dumps
p = KafkaProducer(bootstrap_servers = ['broker:29092'], value_serializer = lambda x:dumps(x).encode('utf-8'))
data = {'name': 'roscoe'}
p.send('Tutorial2.pets', value = data)
p.flush()

Run the Python script:

python3 kafkawrite.py

View the Data in the MongoDB Collection

In the same shell, connect to MongoDB using mongosh, the MongoDB shell by running the following command:

mongosh "mongodb://mongo1"

After you connect successfully, you should see the following MongoDB shell prompt:

rs0 [direct: primary] test>

At the prompt, type the following commands to retrieve all the documents in the Tutorial2.pets MongoDB namespace:

use Tutorial2
db.pets.find()

You should see the following document returned as the result:

{ _id: ObjectId("62659..."), name: 'roscoe' }

Exit the MongoDB shell by entering the command exit.

(Optional) Stop the Docker Containers

After you complete this tutorial, free resources on your computer by stopping or removing Docker assets. You can choose to remove both the Docker containers and images, or exclusively the containers. If you remove the containers and images, you must download them again to restart your MongoDB Kafka Connector development environment, which is approximately 2.4 GB in size. If you exclusively remove the containers, you can reuse the images and avoid downloading most of the large files in the sample data pipeline.

Select the tab that corresponds to the removal task you want to run.

docker-compose -p mongo-kafka down --rmi all

docker-compose -p mongo-kafka down

To restart the containers, follow the same steps required to start them in the Tutorial Setup.