Kafka Connector Quick Start

Overview

This guide shows you how to configure the MongoDB Kafka Connector to send data between MongoDB and Apache Kafka.

After completing this guide, you should understand how to use the Kafka Connect REST API to configure MongoDB Kafka Connectors to read data from MongoDB and write it to a Kafka topic, and to read data from a Kafka topic and write it to MongoDB.

To complete the steps in this guide, you must download and work in a sandbox, a containerized development environment that includes services required to build a sample data pipeline.

Read the following sections to set up your sandbox and sample data pipeline.

Install the Required Packages

Download and install the following packages:

• Docker

• Git

The sandbox uses Docker for convenience and consistency. To learn more about deployment options for Apache Kafka, see the following resources:

• Official Apache Kafka Quick Start

• Official Install Confluent Platform Guide

Download the Sandbox

We created a sandbox that includes the services you need in this tutorial to build your sample data pipeline.

To download the sandbox, clone the tutorial repository to your development environment. Then navigate to the directory that corresponds to the quickstart tutorial. If you use bash or a similar shell, use the following commands:

git clone https://github.com/mongodb-university/kafka-edu.git
cd kafka-edu/docs-examples/mongodb-kafka-base/

Start the Sandbox

The sandbox starts the following services in Docker containers:

• MongoDB, configured as a replica set

• Apache Kafka

• Kafka Connect with the MongoDB Kafka Connector installed

• Apache Zookeeper which manages the configuration for Apache Kafka

To start the sandbox run the following command from the tutorial directory:

docker compose -p mongo-kafka up -d --force-recreate

When you start the sandbox, Docker downloads any images it needs to run.

After Docker downloads and builds the images, you should see the following output in your development environment:

...
Creating zookeeper ... done
Creating broker    ... done
Creating schema-registry ... done
Creating connect         ... done
Creating rest-proxy      ... done
Creating mongo1          ... done
Creating mongo1-setup    ... done

Add Connectors

To complete the sample data pipeline, you must add connectors to Kafka Connect to transfer data between Kafka Connect and MongoDB. Add a source connector to transfer data from MongoDB to Apache Kafka. Add a sink connector to transfer data from Apache Kafka to MongoDB.

To add connectors in the sandbox, first start an interactive bash shell in your Docker container using the following command:

docker exec -it mongo1 /bin/bash

After your shell session starts, you should see the following prompt:

MongoDB Kafka Connector Sandbox $

curl -X POST \
     -H "Content-Type: application/json" \
     --data '
     {"name": "mongo-source",
      "config": {
         "connector.class":"com.mongodb.kafka.connect.MongoSourceConnector",
         "connection.uri":"mongodb://mongo1:27017/?replicaSet=rs0",
         "database":"quickstart",
         "collection":"sampleData",
         "pipeline":"[{\"$match\": {\"operationType\": \"insert\"}}, {$addFields : {\"fullDocument.travel\":\"MongoDB Kafka Connector\"}}]"
         }
     }
     ' \
     http://connect:8083/connectors -w "\n"

...
curl: (7) Failed to connect to connect port 8083: Connection refused

curl -X GET http://connect:8083/connectors

["mongo-source"]

curl -X POST \
     -H "Content-Type: application/json" \
     --data '
     {"name": "mongo-sink",
      "config": {
         "connector.class":"com.mongodb.kafka.connect.MongoSinkConnector",
         "connection.uri":"mongodb://mongo1:27017/?replicaSet=rs0",
         "database":"quickstart",
         "collection":"topicData",
         "topics":"quickstart.sampleData",
         "change.data.capture.handler": "com.mongodb.kafka.connect.sink.cdc.mongodb.ChangeStreamHandler"
         }
     }
     ' \
     http://connect:8083/connectors -w "\n"

curl -X GET http://connect:8083/connectors

["mongo-source", "mongo-sink"]

mongosh mongodb://mongo1:27017/?replicaSet=rs0

rs0 [primary] test>

use quickstart
db.sampleData.insertOne({"hello":"world"})

db.topicData.find()

[
    {
      _id: ObjectId(...),
      hello: 'world',
      travel: 'MongoDB Kafka Connector'
    }
]

exit

Remove the Sandbox

To conserve resources in your development environment, remove the sandbox.

Before you remove the sandbox, exit from the shell session in your Docker container by running the following command:

exit

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 sandbox 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

Next Steps

To learn how to install the MongoDB Kafka Connector, see the Install the MongoDB Kafka Connector guide.

To learn more about how to process and move data from Apache Kafka to MongoDB, see the Sink Connector guide.

To learn more about how to process and move data from MongoDB to Apache Kafka, see the Source Connector guide.