Today on the show we're talking about streaming data and streaming applications with Apache Kafka. We're joined by Kris Jenkins, Developer Advocate from Confluent, and Rob Walters, Product Manager at MongoDB, who will discuss how you can leverage this technology to your benefit and use it in your applications. Kafka is traditionally used for building real time streaming data pipelines and real time streaming applications. It began its life in 2010 at LinkedIn and made its way to the public open-source space through a relationship with Apache, the Apache Foundation, in 2011. Since then, the use of Kafka has grown massively and it's estimated that approximately 30% of all Fortune 500 companies are already using Kafka in one way or another. A great example for why you might want to use Kafka would be perhaps capturing all of the user activity that happens on your website. As users visit your website, they're interacting with links on the page and scrolling up and down. This is potentially large volumes of data. You may want to store this to understand how users are interacting with your website in real time. Kafka will aid in this process by ingesting and storing all of this activity data while serving up reads for applications on the other side. Conversation highlights include:<ul><li>[03:38] What is Kafka?</li><li>[05:29] At the heart of every database</li><li>[08:03] The difference between Kafka and a database</li><li>[09:03] What Kafka's architecture looks like</li><li>[12:03] Kafka as a data backbone of system architecture</li><li>[14:06] MongoDB and Kafka working together</li><li>[15:40] What are "Topics" in Kafka?</li><li>[17:53] Chain stream events</li><li>[19:58] Kafka's history</li><li>[22:07] MongoDB Connector, and Kafka via Confluent Cloud</li><li>[25:53] Popular use cases using Kafka and MongoDB</li><li>[27:48] Kafka and stream processing with games and event data</li><li>[29:13] KSQL and processing against the stream of data</li><li>[30:59] Developer.Confluence.io, a place to learn everything about Kafka</li></ul>

MongoDB

Ep. 115 Exploring Kafka with Kris Jenkins and Rob Walters

Francesco Baldissera

Industry Solutions Consultant

Ashwin Gangadhar

Vittal Pai

Senior Solutions Architect

Learn how to utilize MongoDB and Databricks to build ai-enhanced retail search solutions.

Learn to Build AI-Enhanced Retail Search Solutions with MongoDB and Databricks

Igor Alekseev

Partner Solutions Architect, Data & Analytics

Robert Walters

Senior Product Manager

[Amazon Managed Streaming for Apache Kafka](https://aws.amazon.com/msk/features/) (MSK) is a fully managed, highly available Apache Kafka service. MSK makes it easy to ingest and process streaming data in real time and leverage that data easily within the AWS ecosystem. By being able to quickly stand up a Kafka solution, you spend less time managing infrastructure and more time solving your business problems, dramatically increasing productivity. MSK also supports integration of data sources such as MongoDB via the [AWS MSK Connect](https://aws.amazon.com/msk/features/msk-connect/) (Connect) service. This Connect service works with the [MongoDB Connector for Apache Kafka](https://www.mongodb.com/kafka-connector), enabling you to easily integrate MongoDB data.  

In this blog post, we will walk through how to set up MSK, configure the MongoDB Connector for Apache Kafka, and create a secured VPC Peered connection with MSK and a MongoDB Atlas cluster. The high-level process is as follows:

* Configure Amazon Managed Streaming for Apache Kafka
* Configure EC2 client 
* Configure a MongoDB Atlas Cluster
* Configure Atlas Private Link including VPC and subnet of the MSK
* Configure plugin in MSK for MongoDB Connector
* Create topic on MSK Cluster
* Install MongoSH command line tool on client
* Configure MongoDB Connector as a source or sink

In this example, we will have two collections in the same MongoDB cluster—the “source” and the “sink.” We will insert sample data into the source collection from the client, and this data will be consumed by MSK via the MongoDB Connector for Apache Kafka running as an MSK connector. As data arrives in the MSK topic, another instance of the MongoDB Connector for Apache Kafka will write the data to the MongoDB Atlas cluster “sink” collection. To align with best practices for secure configuration, we will set up an AWS Network Peered connection between the MongoDB Atlas cluster and the VPC containing MSK and the client EC2 instance.  

![Figure 1. Data flow between AWS MSK and MongoDB Atlas ](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/data_flow_mongodb_atlas_177d1bca15.png)

## Configure AWS Managed Service for Kafka

To create an Amazon MSK cluster using the AWS Management Console, sign in to the AWS Management Console, and open the [Amazon MSK console](https://console.aws.amazon.com/msk/home).

* Choose Create cluster and select Quick create.

For Cluster name, enter MongoDBMSKCluster.

For Apache Kafka version, select one that is 2.6.2 or above.

For broker type, select kafla.t3.small.

From the table under All cluster settings, copy the values of the following settings and save them because you will need them later in this blog:

* VPC
* Subnets
* Security groups associated with VPC

![Figure 2. Cluster settings showing VPC and Subnets](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/cluster_settings_95438aaef0.png)

* Choose “Create cluster.”

## Configure an EC2 client

Next, let's configure an EC2 instance to create a topic. This is where the MongoDB Atlas source collection will write to. This client can also be used to query the MSK topic and monitor the flow of messages from the source to the sink.

To create a client machine, open the Amazon EC2 console at [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/).

* Choose Launch instances.
* Choose Select to create an instance of Amazon Linux 2 AMI (HVM) - Kernel 5.10, SSD Volume Type.
* Choose the t2.micro instance type by selecting the check box.
* Choose Next: Configure Instance Details.
* Navigate to the Network list and choose the VPC whose ID you saved in the previous step.
* Go to Auto-assign Public IP list and choose Enable.
* In the menu near the top, select Add Tags.
* Enter Name for the Key and MongoDBMSKCluster for the Value.
* Choose Review and Launch, and then click Launch.
* Choose Create a new key pair, enter MongoDBMSKKeyPair for Key pair name, and then choose Download Key Pair. Alternatively, you can use an existing key pair if you prefer.
* Start the new instance by pressing Launch Instances.

Next, we will need to configure the networking to allow connectivity between the client instance and the MSK cluster.

* Select View Instances. Then, in the Security Groups column, choose the security group that is associated with the MSKTutorialClient instance.
* Copy the name of the security group, and save it for later.
* Open the Amazon VPC console at [https://console.aws.amazon.com/vpc/](https://console.aws.amazon.com/vpc/).
* In the navigation pane, click on Security Groups. Find the security group whose ID you saved in Step 1 (Create an Amazon MSK Cluster). Choose this row by selecting the check box in the first column.
* In the Inbound Rules tab, choose Edit inbound rules.
* Choose Add rule.
* In the new rule, choose All traffic in the Type column. In the second field in the Source column, select the security group of the client machine. This is the group whose name you saved earlier in this step.
* Click Save rules.

The cluster's security group can now accept traffic that comes from the client machine's security group.

## Create MongoDB Atlas Cluster

To create a MongoDB Atlas Cluster, follow the Getting Started with Atlas tutorial. Note that in this blog, you will need to create an M30 Atlas cluster or above—as VPC peering is not available for M0, M2, and M5 clusters.

![Figure 3: Create new cluster dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_new_cluster_144e85016b.png)

Once the cluster is created, configure an AWS private endpoint in the Atlas Network Access UI supplying the same subnets and VPC.  

* Click on Network Access.
* Click on Private Endpoint, and then the Add Private Endpoint button.

![Figure 4. Add private endpoint](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/creating_private_endpoint_ba8f2ce2b8.png)

* Fill out the VPC and subnet IDs from the previous section.

![Figure 5. Create private endpoint](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_private_endpoint_interface_dialogue_58645ad28f.png)

* SSH into the client machine created earlier and issue the following command in the Atlas portal: **aws ec2 create-vpc-endpoint **
    * Note that you may have to first configure the AWS CLI command using **aws configure** before you can create the VPC through this tool. See [Configuration Basics](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html) for more information.

## Configure MSK plugin

Next, we need to create a custom plugin for MSK. This custom plugin will be the MongoDB Connector for Apache Kafka. For reference, note that the connector will need to be uploaded to an S3 repository **before** you can create the plugin. You can download the MongoDB Connector for Apache Kafka from [Confluent Hub](https://www.confluent.io/hub/mongodb/kafka-connect-mongodb).

* Select “Create custom plugin” from the Custom Plugins menu within MSK.
* Fill out the custom plugin form, including the S3 location of the downloaded connector, and click “Create custom plugin.”

![Figure 6. Create custom plugin](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_custom_plugin_7514327ccf.png)

## Create topic on MSK cluster

When we start reading the data from MongoDB, we also need to create a topic in MSK to accept the data. On the client EC2 instance, let’s install Apache Kafka, which includes some basic tools. 

To begin, run the following command to install Java:

**sudo yum install java-1.8.0**

Next, run the command below to download Apache Kafka.

**wget https://archive.apache.org/dist/kafka/2.6.2/kafka_2.12-2.6.2.tgz**

Building off the previous step, run this command in the directory where you downloaded the TAR file:

**tar -xzf kafka_2.12-2.6.2.tgz**

The distribution of Kafka includes a **bin** folder with tools that can be used to manage topics. Go to the **kafka_2.12-2.6.2** directory.

To create the topic that will be used to write MongoDB events, issue this command:

`bin/kafka-topics.sh --create --zookeeper (INSERT YOUR ZOOKEEPER INFO HERE)--replication-factor 1 --partitions 1 --topic MongoDBMSKDemo.Source`

Also, remember that you can copy the Zookeeper server endpoint from the “View Client Information” page on your MSK Cluster. In this example, we are using plaintext.

![Figure 7. Apache ZooKeeper connection](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/apache_zookeeper_connection_6e23beafd6.png)

## Configure source connector

Once the plugin is created, we can create an instance of the MongoDB connector by selecting “Create connector” from the Connectors menu.

![Figure 8. Custom plugin dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/choose_custom_plugin_f50bd58c30.png)

* Select the MongoDB plug in and click “Next.”
* Fill out the form as follows:

Conector name: **MongoDB Source Connector**

Cluster Type: **MSK Connector**

Select the MSK cluster that was created previously, and select “None” under the authentication drop down menu.

Enter your connector configuration (shown below) in the configuration settings text area.

`connector.class=com.mongodb.kafka.connect.MongoSourceConnector
database=MongoDBMSKDemo
collection=Source
tasks.max=1
connection.uri=(MONGODB CONNECTION STRING HERE)
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter=org.apache.kafka.connect.storage.StringConverter`

**Note**: You can find your Atlas connection string by clicking on the Connect button on your Atlas cluster. Select “Private Endpoint” if you have already configured the Private Endpoint above, then press “Choose a connection method.” Next, select “Connect your application” and copy the **mongodb+srv** connection string.

![Figure 9. Access permissions](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/access_permission_error_403a7a5e48.png)

In the “Access Permissions” section, you will need to create an IAM role with the [required trust policy](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html).

![Figure 10. Log delivery dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/aws_msk_log_delivery_0a4ab236a0.png)

Once this is done, click “Next.” The last section will offer you the ability to use logs—which we highly recommend, as it will simplify the troubleshooting process. 

## Configure sink connector

Now that we have the source connector up and running, let’s configure a sink connector to complete the round trip. Create another instance of the MongoDB connector by selecting “Create connector” from the Connectors menu.

![Figure 8. Custom plugin dialog](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/choose_custom_plugin_f50bd58c30.png)

Select the same plugin that was created previously, and fill out the form as follows: 

Connector name: **MongoDB Sink Connector**
Cluster type: **MSK Connector**
Select the MSK cluster that was created previously and select “None” under the authentication drop down menu.
Enter your connector configuration (shown below) in the Configuration Settings text area.

`connector.class=com.mongodb.kafka.connect.MongoSinkConnector
database=MongoDBMSKDemo
collection=Sink
tasks.max=1
topics=MongoDBMSKDemo.Source
connection.uri=(MongoDB Atlas Connection String Gos Here)
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter=org.apache.kafka.connect.storage.StringConverter`

In the Access Permissions section, select the IAM role created earlier that has the [required trust policy](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-service-execution-role.html). As with the previous connector, be sure to leverage a log service like CloudWatch. 

Once the connector is successfully configured, we can test the round trip by writing to the Source collection and seeing the same data in the Sink collection.  

We can insert data in one of two ways: either through the intuitive Atlas UI, or with the new MongoSH (mongoshell) command line tool. Using [MongoSH](https://docs.mongodb.com/mongodb-shell/), you can interact directly with a MongoDB cluster to test queries, perform ad hoc database operations, and more. 

For your reference, we’ve added a section on how to use the mongoshell on your client EC2 instance below. 

## Install MongoDB shell on client

On the client EC2 instance, create a **/etc/yum.repos.d/mongodb-org-5.0.repo** file by typing:

`sudo nano /etc/yum.repos.d/mongodb-org-5.0.repo`

Paste in the following:

`[mongodb-org-5.0]
name=MongoDB Repository
baseurl=https://repo.mongodb.org/yum/amazon/2/mongodb-org/5.0/x86_64/
gpgcheck=1
enabled=1
gpgkey=[https://www.mongodb.org/static/pgp/server-5.0.asc](https://www.mongodb.org/static/pgp/server-5.0.asc)`

Next, install the MongoSH shell with this command:

`sudo yum install -y mongodb-mongosh`

Use the template below to connect to your MongoDB cluster via mongoshell: 

`mongosh “ (paste in your Atlas connection string here) “`

Once connected, type:
`Use MongoDBMSKDemo
db.Source.insertOne({“Testing”:123})`

To check the data on the sink collection, use this command:
`db.Sink.find({})`

If you run into any issues, be sure to check the log files. In this example, we used CloudWatch to read the events that were generated from MSK and the MongoDB Connector for Apache Kafka.

![Figure 12. Log events  ](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/cloud_watch_log_events_e5a0045c3d.png)

## Summary

Amazon Managed Streaming for Apache Kafka ([MSK](https://aws.amazon.com/msk/)) is a fully managed, secure, and highly available Apache Kafka service that makes it easy to ingest and process streaming data in real time. MSK allows you to import Kafka connectors such as the MongoDB Connector for Apache Kafka. These connectors make working with data sources seamless within MSK. In this article, you learned how to set up MSK, MSK Connect, and the MongoDB Connector for Apache Kafka.  You also learned how to set up a MongoDB Atlas cluster and configure it to use AWS network peering. To continue your learning, check out the following resources:

[MongoDB Connector for Apache Kafka Documentation](https://docs.mongodb.com/kafka-connector/current/)
[Amazon MSK Getting Started](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html)
[Amazon MSK Connect Getting Started](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-getting-started.html)

Learn to set up Amazon MSK, configure the MongoDB Connector for Apache Kafka, and use it as a source and sink for data integration with Atlas in AWS.

Integrating MongoDB With Amazon Managed Streaming for Apache Kafka (MSK)

The MongoDB Connector for Apache Kafka documentation now includes new [tutorials](https://www.mongodb.com/docs/kafka-connector/current/tutorials/)! These tutorials introduce you to key concepts behind the connector and by the end, you’ll have an understanding of how to move data between MongoDB and Apache Kafka. The tutorials are as follows:

* [Explore Change Streams](https://www.mongodb.com/docs/kafka-connector/current/tutorials/explore-change-streams/#std-label-kafka-tutorial-explore-change-streams)

Change streams is a MongoDB server feature that provides change data capture (CDC) capabilities for MongoDB collections. The source connector relies on change streams to move data from MongoDB to a Kafka topic. In this tutorial, you will explore creating a change stream and reading change stream events all through a Python application.

* [Getting Started with the MongoDB Kafka Source Connector](https://www.mongodb.com/docs/kafka-connector/current/tutorials/source-connector/#std-label-kafka-tutorial-source-connector)

In this tutorial, you will configure a source connector to read data from a MongoDB collection into an Apache Kafka topic and examine the content of the event messages.

* [Getting Started with the MongoDB Kafka Sink Connector](https://www.mongodb.com/docs/kafka-connector/current/tutorials/sink-connector/#std-label-kafka-tutorial-sink-connector)

In this tutorial, you will configure a sink connector to copy data from a Kafka topic into a MongoDB cluster and then write a Python application to write data into the topic.

* [Replicate Data with a Change Data Capture Handler](https://www.mongodb.com/docs/kafka-connector/current/tutorials/replicate-with-cdc/)

Configure both a MongoDB source and sink connector to replicate data between two collections using the MongoDB CDC handler.

* [Migrate an Existing Collection to a Time Series Collection](https://www.mongodb.com/docs/kafka-connector/current/tutorials/migrate-time-series/)

Time series collections efficiently store sequences of measurements over a period of time, dramatically increasing the performance of time-based data. In this tutorial, you will configure both a source and sink connector to replicate the data from a collection into a time series collection.

These tutorials run locally within a Docker Compose environment that includes Apache Kafka, Kafka Connect, and MongoDB. Before starting them, follow and complete the [Tutorial Setup](https://www.mongodb.com/docs/kafka-connector/current/tutorials/tutorial-setup/).  You will work through the steps using a tutorial shell and containers available on Docker Hub.  The tutorial shell includes tools such as the new Mongo shell, KafkaCat, and helper scripts that make it easy to configure Kafka Connect from the command line.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_tutorial_shell_docker_f39a104d7b.png)

If you have any questions or feedback on the tutorials, please post them on the [MongoDB Community Forums](https://www.mongodb.com/community/forums/c/data/connectors-integrations/48). 

MongoDB Documentation has released a series of tutorials based on a self-hosted Docker compose environment that includes all components needed.

Learn How to Leverage MongoDB Data Within Kafka with New Tutorials!

Pavel Duchovny

Lead Developer Advocate

Go is a modern language built on typed and native code compiling concepts while feeling and utilizing some benefits of dynamic languages. It is fairly simple to install and use, as it provides readable and robust code for many application use cases.

One of those use cases is building agents that report to a centralized data platform via streaming. A widely accepted approach is to communicate the agent data through subscription of distributed queues like Kafka. The Kafka topics can then propagate the data to many different sources, such as a [MongoDB Atlas](https://mongodb.com/atlas) cluster. 

Having a Go agent allows us to utilize the same code base for various operating systems, and the fact that it has good [integration with JSON data](https://www.mongodb.com/languages/golang) and packages such as a [MongoDB driver](https://docs.mongodb.com/drivers/go/) and [Confluent Go Kafka Client](https://docs.confluent.io/clients-confluent-kafka-go/current/overview.html) makes it a compelling candidate for the presented use case.

This article will demo how file size data on a host is monitored from a cross-platform agent written in Golang via a Kafka cluster using a Confluent hosted sink connector to MongoDB Atlas. MongoDB Atlas stores the data in a [time series collection](https://docs.mongodb.com/manual/core/timeseries-collections/). The [MongoDB Charts](https://www.mongodb.com/products/charts) product is a convenient way to show the gathered data to the user.
![Architecture Overview](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Mongo_DB_Dev_Hub_Article_Go_to_Mongo_DB_using_Kafka_Pavel_D_f06175feb5.jpg)
## Preparing the Golang project, Kafka cluster, and MongoDB Atlas

### Configuring a Go project

Our agent is going to run Go. Therefore, you will need to install the Go language software on your host.

Once this step is done, we will create a Go module to begin our project in our working directory:
``` shell
go mod init example/main
```
Now we will need to add the Confluent Kafka dependency to our Golang project:
``` shell
go get -u gopkg.in/confluentinc/confluent-kafka-go.v1/kafka
```

### Configuring a Kafka cluster 
Creating a Confluent Kafka Cluster is done via the Confluent UI. Start by [creating](https://docs.confluent.io/cloud/current/clusters/create-cluster.html) a basic Kafka cluster in the [Confluent Cloud](https://confluent.cloud/). Once ready, create a topic to be used in the Kafka cluster. I created one named “files.” 

[Generate](https://docs.confluent.io/cloud/current/access-management/authenticate/api-keys/api-keys.html) an api-key and api-secret to interact with this Kafka cluster. For the simplicity of this tutorial, I have selected the “Global Access” api-key. For production, it is recommended to give as minimum permissions as possible for the api-key used. Get a hold of the generated keys for future use.

Obtain the Kafka cluster connection string via Cluster Overview > Cluster Settings > Identification > Bootstrap server for future use. Basic clusters are open to the internet and in production, you will need to amend the access list for your specific hosts to connect to your cluster via advanced cluster ACLs.

> **Important:** The Confluent connector requires that the Kafka cluster and the Atlas cluster are deployed in the same region.
>
### Configuring Atlas project and cluster
[Create a project and cluster](https://docs.atlas.mongodb.com/getting-started/) or use an existing Atlas cluster in your project. 
![Atlas Cluster](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_31_at_13_58_17_fb16546b42.png)
Since we are using a time series collection, the clusters must use a 5.0+ version. Prepare your Atlas cluster for a Confluent sink Atlas [connection](https://docs.mongodb.com/kafka-connector/current/introduction/connect/). Inside your project’s access list, enable user and relevant IP addresses of your connector IPs. The access list IPs should be associated to the Atlas Sink Connector, which we will configure in a following section. Finally, get a hold of the Atlas [connection string](https://docs.atlas.mongodb.com/connect-to-cluster/#use-the-connect-to-your-cluster-dialog-to-connect-to-your-cluster) and the main cluster DNS. For more information about best securing and getting the relevant IPs from your Confluent connector, please read the following article: [MongoDB Atlas Sink Connector for Confluent Cloud](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html).

## Adding agent main logic
Now that we have our Kafka cluster and Atlas clusters created and prepared, we can initialize our agent code by building a small main file that will monitor my `./files` directory and capture the file names and sizes. I’ve added a file called `test.txt` with some data in it to bring it to ~200MB.

Let’s create a file named `main.go` and write a small logic that performs a constant loop with a 1 min sleep to walk through the files in the `files` folder:
``` go
package main

import (
 "fmt"
 "encoding/json"
 "time"
 "os"
 "path/filepath"
)

type Message struct {
 Name string
 Size float64
 Time int64
}

func samplePath (startPath string) error {
 err := filepath.Walk(startPath,
 func(path string, info os.FileInfo, err error) error {
 
 var bytes int64
 bytes = info.Size()


 var kilobytes int64
 kilobytes = (bytes / 1024)

 var megabytes float64
 megabytes = (float64)(kilobytes / 1024) // cast to type float64

 var gigabytes float64
 gigabytes = (megabytes / 1024)

 now := time.Now().Unix()*1000




 m := Message{info.Name(), gigabytes, now}
 value, err := json.Marshal(m)

 
 if err != nil {
 panic(fmt.Sprintf("Failed to parse JSON: %s", err))
 }


 fmt.Printf("value: %v\n", string(value))
 return nil;
 })
 if err != nil {
 return err
 }
 return nil;
}

func main() {
 for {
 err := samplePath("./files");
 if err != nil {
 panic(fmt.Sprintf("Failed to run sample : %s", err))
 }
 time.Sleep(time.Minute)
 }

}
```
The above code simply imports helper modules to traverse the directories and for JSON documents out of the files found. 

Since we need the data to be marked with the time of the sample, it is a great fit for time series data and therefore should eventually be stored in a time series collection on Atlas. If you want to learn more about time series collection and data, please read our article, [MongoDB Time Series Data](http://mongodb.com/features/mongodb-time-series-data).

We can test this agent by running the following command:
``` shell
go run main.go
```
The agent will produce JSON documents similar to the following format:
``` shell
value: {"Name":"files","Size":0,"Time":1643881924000}
value: {"Name":"test.txt","Size":0.185546875,"Time":1643881924000}
```
## Creating a Confluent MongoDB connector for Kafka
Now we are going to create a Kafka Sink connector to write the data coming into the “files” topic to our Atlas Cluster’s time series collection.

Confluent Cloud has a very popular integration running MongoDB’s Kafka connector as a hosted solution integrated with their Kafka clusters. Follow these [steps](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html#using-the-ccloud-console) to initiate a connector deployment.

The following are the inputs provided to the connector:
![Connector Setup 1](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_30_at_14_08_39_e1e3eff816.png)
![Connector Setup 2](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_30_at_14_08_54_6521f4b53e.png)
![Connector Setup 3](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_30_at_14_09_04_4de0d474ac.png)
Once you set it up, following the guide, you will eventually have a similar launch summary page:
![Connector Final Summary](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_24_at_13_02_08_d460ad2de6.png)
After provisioning every populated document into the `files` queue will be pushed to a time series collection `hostMonitor.files` where the date field is `Time` and metadata field is `Name`.
## Pushing data to Kafka
Now let’s edit the `main.go` file to use a Kafka client and push each file measurement into the “files” queue.

Add the client library as an imported module:
``` go
import (
 "fmt"
 "encoding/json"
 "time"
 "os"
 "path/filepath"
 "github.com/confluentinc/confluent-kafka-go/kafka"
)
```
Add the Confluent cloud credentials and cluster DNS information. Replace `<CONFLUENT-SERVER>:<CONFLUENT-PORT>` found on the Kafka Cluster details page and the `<ACCESS-KEY>` , `<SECRET-KEY>` generated in the Kafka Cluster:
``` go
const (
 bootstrapServers = “<CONFLUENT-SERVER>:<CONFLUENT-PORT>"
 ccloudAPIKey = "<ACCESS-KEY>"
 ccloudAPISecret = "<SECRET-KEY>"
)
```
The following code will initiate the producer and produce a message out of the marshaled JSON document: 
``` go
 topic := "files"
 // Produce a new record to the topic...
 producer, err := kafka.NewProducer(&kafka.ConfigMap{
 "bootstrap.servers": bootstrapServers,
 "sasl.mechanisms": "PLAIN",
 "security.protocol": "SASL_SSL",
 "sasl.username": ccloudAPIKey,
 "sasl.password": ccloudAPISecret})

 if err != nil {
 panic(fmt.Sprintf("Failed to create producer: %s", err))
 }

 producer.Produce(&kafka.Message{
 TopicPartition: kafka.TopicPartition{Topic: &topic,
 Partition: kafka.PartitionAny},
 Value: []byte(value)}, nil)

 // Wait for delivery report
 e := <-producer.Events()

 message := e.(*kafka.Message)
 if message.TopicPartition.Error != nil {
 fmt.Printf("failed to deliver message: %v\n",
 message.TopicPartition)
 } else {
 fmt.Printf("delivered to topic %s [%d] at offset %v\n",
 *message.TopicPartition.Topic,
 message.TopicPartition.Partition,
 message.TopicPartition.Offset)
 }

 producer.Close()
 ```

The entire `main.go` file will look as follows:
``` go 
package main


import (
 "fmt"
 "encoding/json"
 "time"
 "os"
 "path/filepath"
 "github.com/confluentinc/confluent-kafka-go/kafka")

type Message struct {
 Name string
 Size float64
 Time int64
}


const (
 bootstrapServers = "<SERVER-DNS>:<SERVER-PORT>"
 ccloudAPIKey = "<ACCESS-KEY>"
 ccloudAPISecret = "<SECRET-KEY>"
)

func samplePath (startPath string) error {
 
 err := filepath.Walk(startPath,
 func(path string, info os.FileInfo, err error) error {
 if err != nil {
 return err
 }
 fmt.Println(path, info.Size())

 var bytes int64
 bytes = info.Size()


 var kilobytes int64
 kilobytes = (bytes / 1024)

 var megabytes float64
 megabytes = (float64)(kilobytes / 1024) // cast to type float64

 var gigabytes float64
 gigabytes = (megabytes / 1024)

 now := time.Now().Unix()*1000


 

 m := Message{info.Name(), gigabytes, now}
 value, err := json.Marshal(m)

 
 if err != nil {
 panic(fmt.Sprintf("Failed to parse JSON: %s", err))
 }


 fmt.Printf("value: %v\n", string(value))

 topic := "files"
 // Produce a new record to the topic...
 producer, err := kafka.NewProducer(&kafka.ConfigMap{
 "bootstrap.servers": bootstrapServers,
 "sasl.mechanisms": "PLAIN",
 "security.protocol": "SASL_SSL",
 "sasl.username": ccloudAPIKey,
 "sasl.password": ccloudAPISecret})
 
 if err != nil {
 panic(fmt.Sprintf("Failed to create producer: %s", err))
 }
 
 producer.Produce(&kafka.Message{
 TopicPartition: kafka.TopicPartition{Topic: &topic,
 Partition: kafka.PartitionAny},
 Value: []byte(value)}, nil)
 
 // Wait for delivery report
 e := <-producer.Events()
 
 message := e.(*kafka.Message)
 if message.TopicPartition.Error != nil {
 fmt.Printf("failed to deliver message: %v\n",
 message.TopicPartition)
 } else {
 fmt.Printf("delivered to topic %s [%d] at offset %v\n",
 *message.TopicPartition.Topic,
 message.TopicPartition.Partition,
 message.TopicPartition.Offset)
 }
 
 producer.Close()

 return nil;
})
if err != nil {
 return err
}
 return nil;
}

func main() {
 for {
 err := samplePath("./files");
 if err != nil {
 panic(fmt.Sprintf("Failed to run sample : %s", err))
 }
 time.Sleep(time.Minute)
 
 }

}
```
Now when we run the agent while the Confluent Atlas sink connector is fully provisioned, we will see messages produced into the `hostMonitor.files` time series collection:
![Atlas Data](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_24_at_14_58_42_3f16961049.png)
## Analyzing the data using MongoDB Charts
To put our data into use, we can create some beautiful charts on top of the time series data. In a [line graph](https://docs.mongodb.com/charts/chart-type-reference/line-area-chart/#std-label-line-area-chart-ref), we configure the X axis to use the Time field, the Y axis to use the Size field, and the series to use the Name field. The following graph shows the colored lines represented as the evolution of the different file sizes over time.
![Line Graph Chart](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2022_01_24_at_13_00_53_0b3e2b9d25.png)
Now we have an agent and a fully functioning Charts dashboard to analyze growing files trends. This architecture allows big room for extensibility as the Go agent can have further functionalities, more subscribers can consume the monitored data and act upon it, and finally, MongoDB Atlas and Charts can be used by various applications and embedded to different platforms.

## Wrap Up
Building Go applications is simple yet has big benefits in terms of performance, cross platform code, and a large number of supported libraries and clients. Adding MongoDB Atlas via a Confluent Cloud Kafka service makes the implementation a robust and extensible stack, streaming data and efficiently storing and presenting it to the end user via Charts.

In this tutorial, we have covered all the basics you need to know in order to start using Go, Kafka, and MongoDB Atlas in your next streaming projects.

Try [MongoDB Atlas](https://www.mongodb.com/basics/mongodb-atlas-tutorial) and Go today!

When combined with Confluent Kafka streaming to MongoDB Atlas, Go lets you form tools and applications with real-time data streaming and analytics.

Go to MongoDB Using Kafka Connectors - Ultimate Agent Guide

Diego Rodriguez

Senior Technical Services Engineer

MongoDB Connector for Apache Kafka (MongoDB Connector) is an [open-source Java application](https://github.com/mongodb/mongo-kafka) that works with Apache Kafka Connect enabling seamless data integration of MongoDB with the Apache Kafka ecosystem. When working with the MongoDB Connector, the default values cover a great variety of scenarios, but there are some scenarios that require more fine-grained tuning. In this article, we will walk through important configuration properties that affect the MongoDB Kafka Source and Sink Connectors performance, and share general recommendations.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_connector_ebf714c66b.png)

## Tuning the source connector

Let’s first take a look at the connector when it is configured to read data from MongoDB and write it into a Kafka topic. When you configure the connector this way, it is known as a “source connector.” 

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_connected_to_apache_kafka_7fc20b4f4c.png)

When the connector is configured as a source, a [change stream](https://docs.mongodb.com/manual/changeStreams/) is opened within the MongoDB cluster based upon any configuration you specified, such as pipeline. These [change stream events](https://docs.mongodb.com/manual/reference/change-events/) get read into the connector and then written out to the Kafka topic, and they resemble the following:

```
{
 _id : { <BSON Object> },
 "operationType" : "<operation>",
 "fullDocument" : { <document> },
 "ns" : {
 "db" : "<database>",
 "coll" : "<collection>"
 },
 "to" : {
 "db" : "<database>",
 "coll" : "<collection>"
 },
 "documentKey" : { "_id" : <value> },
 "updateDescription" : {
 "updatedFields" : { <document> },
 "removedFields" : [ "<field>", ... ],
 "truncatedArrays" : [
 { "field" : <field>, "newSize" : <integer> },
 ...
 ]
 },
 "clusterTime" : <Timestamp>,
 "txnNumber" : <NumberLong>,
 "lsid" : {
 "id" : <UUID>,
 "uid" : <BinData>
 }
}
```
The connector configuration properties help define what data is written out to Kafka. For example, consider the scenario where we insert into MongoDB the following:

```
Use Stocks
db.StockData.insertOne({'symbol':'MDB','price':441.67,'tx_time':Date.now()})
```
When **publish.full.document.only** is set to false (the default setting), the connector writes the entire event as shown below:

```
{"_id": 
{"_data": "826205217F000000022B022C0100296E5A1004AA1707081AA1414BB9F647FD49855EE846645F696400646205217FC26C3DE022E9488E0004"},
"operationType": "insert",
"clusterTime":
 {"$timestamp": 
 {"t": 1644503423, "i": 2}},
"fullDocument":
 {"_id":
 {"$oid": "6205217fc26c3de022e9488e"},
 "symbol": "MDB",
 "price": 441.67,
 "tx_time": 1.644503423267E12},
 "ns":
 {"db": "Stocks", "coll": "StockData"},
 "documentKey":
 {"_id": {"$oid": "6205217fc26c3de022e9488e"}}}}
}
```
When **publish.full.document.only** is set to true and we issue a similar statement, it looks like the following:

```
use Stocks
db.StockData.insertOne({'symbol':'TSLA','price':920.00,'tx_time':Date.now()})
```
We can see that the data written to the Kafka topic is just the changed document itself, which in this example, is an inserted document.

```
{"_id": {"$oid": "620524b89d2c7fb2a606aa16"}, "symbol": "TSLA",
 "price": 920,
 "tx_time": 1.644504248732E12}"}
```
### Resume tokens

Another import concept to understand with source connectors is resume tokens. Resume tokens make it possible for the connector to fail, get restarted, and resume where it left off reading the MongoDB change stream. Resume tokens by default are stored in a Kafka topic defined by the **offset.storage.topic** parameter (configurable at the Kafka Connect Worker level for distributed environments) or in the file system in a file defined by the **offset.storage.file.filename** parameter (configurable at the Kafka Connect Worker level for standalone environments). In the event that the connector has been offline and the underlying MongoDB oplog has rolled over, you may get an error when the connector restarts. Read the [Invalid Resume Token](https://docs.mongodb.com/kafka-connector/current/troubleshooting/recover-from-invalid-resume-token/#std-label-kafka-troubleshoot-invalid-resume-token) section of the online documentation to learn more about this condition.

### Configuration properties

The full set of properties for the Kafka Source Connector can be found [in the documentation](https://docs.mongodb.com/kafka-connector/current/source-connector/configuration-properties/). The properties that should be considered with respect to performance tuning are as follows:

* **batch.size**: the cursor batch size that defines how many change stream documents are retrieved on each **getMore** operation. Defaults to 1,000.
* **poll.await.time.ms**: The maximum amount of time in milliseconds that the server waits for new data changes to report to the change stream cursor before returning an empty batch. Defaults to 5,000.
* **poll.max.batch.size**: maximum number of source records to send to Kafka at once. This setting can be used to limit the amount of data buffered internally in the Connector. Defaults to 1,000.
* **pipeline**: an array of aggregation pipeline stages to run in your change stream. Defaults to an empty pipeline that provides no filtering.
* **copy.existing.max.threads**: the number of threads to use when performing the data copy. Defaults to the number of processors.
* **copy.existing.queue.size**: the max size of the queue to use when copying data. This is buffered internally by the Connector. Defaults to 16,000.

### Recommendations
The following are some general recommendations and considerations when configuring the source connector:

#### Scaling the source
One of the most common questions is how to scale the source connector. For scenarios where you have a large amount of data to be copied via **copy.existing**, keep in mind that using the source connector this way may not be the best way to move this large amount of data. Consider the process for copy.existing:

* Store the latest change stream resume token.
* Spin up a thread (up to **copy.existing.max.threads**) for each namespace that is being copied.
* When all threads finish, the resume tokens are read, written, and caught up to current time.

While technically, the data will eventually be copied, this process is relatively slow. And if your data size is large and your incoming data is faster than the copy process, the connector may never get into a state where new data changes are handled by the connector.

For high throughput datasets trying to be copied with copy.existing, a typical situation is overwriting the resume token stored in (1) due to high write activity. This breaks the copy.existing functionality, and it will need to be restarted, on top of dealing with the messages that were already processed to the Kafka topic. When this happens, the alternatives are:

* Increase the oplog size to make sure the copy.existing phase can finish.
* Throttle write activity in the source cluster until the copy.existing phase finishes. 

Another option for handling high throughput of change data is to configure multiple source connectors. Each source connector should use a **pipeline** and capture changes from a subset of the total data. Keep in mind that each time you create a source connector pointed to the same MongoDB cluster, it creates a separate change stream. Each change stream requires resources from the MongoDB cluster, and continually adding them will decrease server performance. That said, this degradation may not become noticeable until the amount of connectors reaches the 100+ range, so breaking your collections into five to 10 connector pipelines is the best way to increase source performance. In addition, using several different source connectors on the same namespace changes the total ordering of the data on the sink versus the original order of data in the source cluster. 

#### Tune the change stream pipeline
When building your Kafka Source Connector configuration, ensure you appropriately tune the “pipeline” so that only wanted events are flowing from MongoDB to Kafka Connect, which helps reduce network traffic and processing times. For a detailed pipeline example, check out the [Customize a Pipeline to Filter Change Events](https://docs.mongodb.com/kafka-connector/current/source-connector/usage-examples/custom-pipeline/#customize-a-pipeline-to-filter-change-events) section of the online documentation.

#### Adjust to the source cluster throughput
Your Kafka Source Connector can be watching a set of collections with a low volume of events, or the opposite, a set of collections with a very high volume of events.

In addition, you may want to tune your Kafka Source Connector to react faster to changes, reduce round trips to MongoDB or Kafka, and similar changes.

With this in mind, consider adjusting the following properties for the Kafka Source Connector:

* Adjust the value of **batch.size**:
 * Higher values mean longer processing times on the source cluster but fewer round trips to it. It can also increase the chances of finding relevant change events when the volume of events being watched is small.
 * Lower values mean shorter processing times on the source cluster but more round trips to it. It can reduce the chances of finding relevant change events when the volume of events being watched is small.
* Adjust the value of **poll.max.batch.size**:
 * Higher values require more memory to buffer the source records with fewer round trips to Kafka. This comes at the expense of the memory requirements and increased latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic.
 * Lower values require less memory to buffer the source records with more round trips to Kafka. It can also help reduce the latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic.
* Adjust the value of **poll.await.time.ms**:
 * Higher values can allow source clusters with a low volume of events to have any information to be sent to Kafka at the expense of increased latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic.
 * Lower values reduce latency from the moment a change takes place in MongoDB to the point the Kafka message associated with that change reaches the destination topic. But for source clusters with a low volume of events, it can prevent them from having any information to be sent to Kafka.

This information is an overview of what to expect when changing these values, but keep in mind that they are deeply interconnected, with the volume of change events on the source cluster having an important impact too:

1. The Kafka Source Connector issues getMore commands to the source cluster using **batch.size**.
2. The Kafka Source Connector receives the results from step 1 and waits until either **poll.max.batch.size** or **poll.await.time.ms** is reached. While this doesn’t happen, the Kafka Source Connector keeps “feeding” itself with more getMore results.
3. When either **poll.max.batch.size** or **poll.await.time.ms** is reached, the source records are sent to Kafka.

#### “Copy existing” feature 

When running with the **copy.existing** property set to **true**, consider these additional properties:

* **copy.existing.queue.size**: the amount of records the Kafka Source Connector buffers internally. This queue and its size include all the namespaces to be copied by the “Copy Existing” feature. If this queue is full, the Kafka Source Connector blocks until space becomes available.
* **copy.existing.max.threads**: the amount of concurrent threads used for copying the different namespaces. There is a one namespace to one thread mapping, so it is common to increase this up to the maximum number of namespaces being copied. If the number exceeds the number of cores available in the system, then the performance gains can be reduced.
* **copy.existing.allow.disk.use**: allows the copy existing aggregation to use temporary disk storage if required. The default is set to true but should be set to false if the user doesn't have the permissions for disk access.

#### Memory implications 

If you experience JVM “out of memory” issues on the Kafka Connect Worker process, you can try reducing the following two properties that control the amount of data buffered internally:

* **poll.max.batch.size**
* **copy.existing.queue.size**: applicable if the “copy.existing” property is set to true.

It is important to note that lowering these values can result in unwanted impact. Adjusting the JVM Heap Size to your environment needs is recommended as long as you have available resources and the memory needs are not the result of memory leaks.

## Tuning the sink connector
When the MongoDB Connector is configured as a sink, it reads from a Kafka topic and writes to a MongoDB collection.

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/image1_c8f32992a5.png)

As with the source, there exists a mechanism to ensure offsets are stored in the event of a sink failure. Kafka connect manages this, and the information is stored in the __consumer_offsets topic. The MongoDB Connector has configuration properties that affect performance. They are as follows:

* **max.batch.size**: the maximum number of sink records to batch together for processing. A higher number will result in more documents being sent as part of a single bulk command. Default value is 0.
* **rate.limiting.every.n**: number of processed batches that trigger the rate limit. A value of 0 means no rate limiting. Default value is 0. In practice, this setting is rarely used.
* **rate.limiting.timeout**: how long (in milliseconds) to wait before continuing to process data once the rate limit is reached. Default value is 0. This setting is rarely used.
* **tasks.max**: the maximum number of tasks. Default value is 1.

### Recommendations 
#### Add indexes to your collections for consistent performance 
Writes performed by the sink connector take additional time to complete as the size of the underlying MongoDB collection grows. To prevent performance deterioration, use an index to support these write queries.

#### Achieve as much parallelism as possible 
The Kafka Sink Connector (KSC) can take advantage of parallel execution thanks to the **tasks.max** property. The specified number of tasks will only be created if the source topic has the same number of partitions. Note: A partition should be considered as a logic group of ordered records, and the producer of the data determines what each partition contains.
Here is the breakdown of the different combinations of number of partitions in the source topic and tasks.max values:

**If working with more than one partition but one task:**

* The task processes partitions one by one: Once a batch from a partition is processed, it moves on to another one so the order within each partition is still guaranteed.
* Order among all the partitions is not guaranteed.

**If working with more than one partition and an equal number of tasks:**

* Each task is assigned one partition and the order is guaranteed within each partition.
* Order among all the partitions is not guaranteed.

**If working with more than one partition and a smaller number of tasks:**

* The tasks that are assigned more than one partition process partitions one by one: Once a batch from a partition is processed, it moves on to another one so the order within each partition is still guaranteed.
* Order among all the partitions is not guaranteed.

**If working with more than one partition and a higher number of tasks:**

* Each task is assigned one partition and the order is guaranteed within each partition.
* KSC will not generate an excess number of tasks.
* Order among all the partitions is not guaranteed.

Processing of partitions may not be in order, meaning that Partition B may be processed before Partition A. All messages within the partition conserve strict order.

Note: When using MongoDB to write CDC data, the order of data is important since, for example, you do not want to process a delete before an update on the same data. If you specify more than one partition for CDC data, you run the risk of data being out of order on the sink collection.

#### Tune the bulk operations 
The Kafka Sink Connector (KSC) works by issuing [bulk write operations](https://docs.mongodb.com/manual/core/bulk-write-operations/). All the bulk operations that the KSC executes are, by default, ordered and as such, the order of the messages is guaranteed within a partition. See [Ordered vs Unordered Operations](https://docs.mongodb.com/manual/core/bulk-write-operations/#ordered-vs-unordered-operations) for more information. Note: As of 1.7, **bulk.write.ordered**, if set to false, will process the bulk out of order, enabling more documents within the batch to be written in the case of a failure of a portion of the batch.

The amount of operations that are sent in a single bulk command can have a direct impact on performance. You can modify this by adjusting **max.batch.size**:

* A higher number will result in more operations being sent as part of a single bulk command. This helps improve throughput at the expense of some added latency. However, a very big number might result in cache pressure on the destination cluster.
* A small number will ease the potential cache pressure issues which might be useful for destination clusters with fewer resources. However, throughput decreases, and you might experience consumer lag on the source topics as the producer might publish messages in the topic faster than the KSC processes them.
* This value affects processing within each of the tasks of the KSC.

#### Throttle the Kafka sink connector 
In the event that the destination MongoDB cluster is not able to handle consistent throughput, you can configure a throttling mechanism. You can do this with two properties:

* **rate.limiting.every.n**: number of processed batches that should trigger the rate limit. A value of 0 means no rate limiting.
* **rate.limiting.timeout**: how long (in milliseconds) to wait before continuing to process data once the rate limit is reached.

The end result is that whenever the KSC writes **rate.limiting.every.n** number of batches, it waits **rate.limiting.timeout milliseconds** before writing the next batch. This allows a destination MongoDB cluster that cannot handle consistent throughput to recover before receiving new load from the KSC.

To build a MongoDB/Apache Kafka solution, the default configuration values need some tweaks. Learn important configuration properties and best practices. 

Tuning the MongoDB Connector for Apache Kafka

Michael Lynn

Principal Developer Advocate

In this podcast episode transcript, we chat with Tim Fogarty, Varsha Subrahmanyam, and Evgeni Dobranov about BI Connector and mongomirror.

MongoDB Podcast Interview With Connectors and Translators Team

Community

Rajesh S Nair

Senior Software Engineer

In today's data-driven world, businesses are continuously seeking innovative ways to harness the full potential of their data. One critical aspect of this journey involves data migration – the process of transferring data from one database system to another, often driven by evolving business needs, scalability requirements, or the desire to tap into new technologies.

In this era of digital transformation, where agility and scalability are paramount, organizations are increasingly turning to NoSQL databases like MongoDB for their ability to handle unstructured or semi-structured data at scale. On the other hand, relational databases like PostgreSQL have long been trusted for their robustness and support for structured data.

As businesses strive to strike the right balance between the structured and unstructured worlds of data, the question arises: How can you seamlessly migrate from a relational database like PostgreSQL to the flexible [documented-oriented model](https://www.mongodb.com/document-databases) of MongoDB while ensuring data integrity, minimal downtime, and efficient synchronization?

The answer lies in an approach that combines the power of Confluent Kafka, a distributed streaming platform, with the agility of MongoDB. In this article, we'll explore the art and science of migrating from PostgreSQL to MongoDB Atlas, leveraging Confluent Kafka as our data streaming bridge. We'll delve into the step-by-step tutorial that can make this transformation journey a success, unlocking new possibilities for your data-driven initiatives.

## Kafka: a brief introduction

### What is Apache Kafka?
Apache Kafka is an open-source distributed streaming platform developed by the Apache Software Foundation that is designed to handle real-time data streams.

To understand Kafka, imagine a busy postal system in a bustling city. In this city, there are countless businesses and individuals sending packages and letters to one another, and it's essential that these messages are delivered promptly and reliably.

Apache Kafka is like the central hub of this postal system, but it's not an ordinary hub; it's a super-efficient, high-speed hub with a memory that never forgets. When someone sends a message (data) to Kafka, it doesn't get delivered immediately. Instead, it's temporarily stored within Kafka's memory. Messages within Kafka are not just one-time deliveries. They can be read and processed by multiple parties. Imagine if every package or letter sent through the postal system had a copy available for anyone who wanted it. This is the core concept of Kafka: It's a distributed, highly scalable, and fault-tolerant message streaming platform.

From maintaining real-time inventory information for e-commerce to supporting real-time patient monitoring, Kafka has varied business use cases in different industries and can be used for log aggregation and analysis, event sourcing, real-time analytics, data integration, etc.

## Kafka Topics
In the same analogy of the postal system, the system collects and arranges its letters and packages into different sections and organizes them into compartments for each type of item. Kafka does the same. The messages it receives from the producer of data are arranged and organized into Kafka topics. Kafka topics are like different mailboxes where messages with a similar theme are placed, and various programs can send and receive these messages to exchange information. This helps keep data well-organized and ensures that the right people or systems can access the information they need from the relevant topic.

## Kafka connectors
Kafka connectors are like special mailboxes that format and prepare letters (data) in a way that Kafka can understand, making it easier for data to flow between different systems. Say the sender (system) wants to send a letter (data) to the receiver (another system) using our postal system (Kafka). Instead of just dropping the letter in the regular mailbox, the sender places it in a special connector mailbox outside their house. This connector mailbox knows how to format the letter properly. So connectors basically act as a bridge that allows data to flow between Kafka and various other data systems.

## Confluent Kafka
Confluent is a company that builds tools and services. It has built tools and services for Apache Kafka to make it more robust and feature-rich. It is like working with a more advanced post office that not only receives and delivers letters but also offers additional services like certified mail, tracking, and package handling. The migration in this article is done using Confluent Kafka through its browser user interface.

## Setting up a Confluent Kafka account
To begin with, you can set up an account on Confluent Kafka by registering on the [Confluent Cloud website](https://confluent.cloud/login). You can sign up with your email account or using GitHub.

Once you log in, this is how the home page looks:
![Confluent Cloud Homepage](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt18f94d7a969ea461/6576ded4c5fb31557b9f9106/Confluent_Cloud_Homepage.png)

This free account comes with free credits worth $400 which you can use to utilize the resources in the Confluent Cloud. If your database size is small, your migration could also be completed within this free credit limit. If you go to the billing section, you can see the details regarding the credits.

To create a new cluster, topics, and connectors for your migration, click on the Environments tab from the side menu and create a new environment and cluster.

![Confluent environments overview page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltc729be1ddf14c580/6576ded47d37e21c40a4a0f7/Environment.png)

You can select the type of cluster. Select the type “basic” which is the free tier with basic configuration. If you want to have a higher configuration for the cluster, you can select the “standard”, “enterprise,” or “dedicated” cluster types which have higher storage, partition, and uptime SLA respectively with hourly rates.

![Create cluster tier selection page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt08795d40d8637ffd/6576ded480f4ba868a8d11b4/Tier_Selection.png)

Next, you can select the region/zone where your cluster has to be deployed along with the cloud provider you want for your cluster (AWS, GCP, or Azure ). The prerequisite for your data migration to work through Kafka connectors is that the Kafka cluster where you create your connectors should be in the same region as your MongoDB Atlas cluster to where you will migrate your PostgreSQL data.

Then, you can provide your payment information and launch your cluster.

Once your cluster is launched, this is how the cluster menu looks with options to have a cluster overview and create topics and connectors, among other features.

![Cluster overview page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blteb27d3cd5daf96e2/6576ded338d46b5d667cafd8/Cluster_Overview.png)

With this, we are ready with the basic Kafka setup to migrate your data from PostgreSQL to MongoDB Atlas.

## Setting up PostgreSQL test data
For this example walkthrough, if you do not have an existing PostgreSQL database that you would like to migrate to a MongoDB Atlas instance using Confluent Kafka, you can create a sample database in PostgreSQL by following the below steps and then continue with this tutorial.

 1. [Download PostgreSQL Database Server](https://www.postgresql.org/download/) from the official website and start your instance locally.
 2. Download the [pgadmin tool](https://www.pgadmin.org/download/) and connect to your local instance.
 3. Create a database ```mytestdb``` and table ```users``` and put some sample data into the employee table.
```sql
-- Create the database mytestdb
CREATE DATABASE mytestdb;

-- Connect to the mytestdb database
\c org;

-- Create the users table
CREATE TABLE users (
 id SERIAL PRIMARY KEY,
 firstname VARCHAR(50),
 lastname VARCHAR(50),
 age INT
);

-- Insert sample data into the 'users' table
INSERT INTO users (firstname, lastname, age)
VALUES
 ('John', 'Doe', 25),
 ('Jane', 'Smith', 30),
 ('Bob', 'Johnson', 22);
```
Keep in mind that the host where your PostgreSQL is running — in this case, your local machine — should have Confluent Kafka whitelisted in a firewall. Otherwise, the source connector will not be able to reach the PostgreSQL instance.

## Steps for data migration using Confluent Kafka
To migrate the data from PostgreSQL to MongoDB Atlas, we have to configure a source connector to connect to PostgreSQL that will stream the data into the Confluent Cloud topic. Then, we will configure a sink connector for MongoDB Atlas to read the data from the created topic and write to the respective database in the MongoDB Atlas cluster.

### Configuring the PostgreSQL source connector
To configure the PostgreSQL source connector, follow the below steps:

 1. Click on the Connectors tab in your newly created cluster in Confluent. It will list popular plugins available in the Confluent Cloud. You can search for the “postgres source” connector plugin and use that to create your custom connector to connect to your PostgreSQL database.
![Selecting Postgres source as connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt2cc46c66e5f28e8e/6576ded431d3b5bb7f227e9f/Postgres_Source.png)

 2. Next, you will be prompted for the topic prefix. Provide the name of the topic into which you want to stream your PostgreSQL data. If you leave it empty, the topic will be created with the table name for you.

 3. You can then specify the access levels for the new connector you are creating. You can keep it global and also download the API credentials that you can use in your applications, if needed to connect to your cluster. For this migration activity, you will not need it — but you will need to create it to move to the next step.

 4. Next, you will be prompted for connection details of PostgreSQL.You can provide the connection params, schema context, transaction isolation levels, poll intervals, etc. for the connection. 

 5. Select the output record type as JSON. MongoDB natively uses the JSON format. You will also have to provide the name of the table that you are trying to migrate.

 6. In the next screen, you will be redirected to an overview page with all the configurations you provided in JSON format along with the cost for running this source connector per hour.
![Json as output configuration](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3e095a9915109a7d/6576ded33a0a12c0c6960c2f/JSON_Source_Settings.png)

 7. Once you create your source connector, you can see its status in the
 Connectors tab and if it is running or has failed. The source
 connector will start syncing the data to the Confluent Cloud topic
 immediately after starting up. You can check the number of messages
 processed by the connector by clicking on the new connector. If the
 connector has failed to start, you can check connector logs and
 rectify any issues by reconfiguring the connector settings.
![Connectors overview status page](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3620c4b3113b518b/6576ded4116ab85ba86eb1c6/Postgres_Connector.png)

### Validating data in the new topic
Once your Postgres source connector is running, you can switch to the Topics tab to list all the topics in your cluster, and you will be able to view the new topic created by the source connector.

If you click on the newly created topic and navigate to the “Messages” tab, you will be able to view the processed messages. If you are not able to see any recent messages, you can check them by selecting the “Jump to time” option, selecting the default partition 0, and providing a recent past time from the date picker. Here, my topic name is “users.”

![Creating users topic](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt76230cc5ac7c1dd7/6576ded3cc0790c11b39f495/Topics.png)

Below, you can see the messages processed into my “users” topic from the users table in PostgreSQL.
![Processed messages](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blte1c33e44d1027443/6576ded3ec26f800f0a88b15/Users_Topic.png)

### Configuring the MongoDB Atlas sink connector
Now that we have the data that you wanted to migrate (one table, in our example) in our Confluent Cloud topic, we can create a sink connector to stream that data into your MongoDB Atlas cluster. Follow the below steps to configure the data inflow:

 1. Go to the Connectors tab and search for “MongoDB Atlas Sink” to find the MongoDB Atlas connector plugin that you will use to create your custom sink connector.
![Atlas Sink Connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt259f6dbde195b4c2/6576ded319da2638891e88b3/Atlas_Sink_Connector.png)
 2. You will then be asked to select the topic for which you are creating this sink connector. Select the respective topic and click on “Continue.”
![Adding MongoDB Atlas Sink connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt90e77eea6efd5122/6576ded3cd283050e0e464aa/Add_Atlas_Sink_Connector.png)
 3. You can provide the access levels for the sink connector and also download the API credentials if needed, as in the case of the source connector.
 4. In the next section, you will have to provide the connection details for your MongoDB Atlas cluster — including the hostname, username/password, database name, and collection name — into which you want to push the data. The connection string for Atlas will be in the format ```mongodb+srv://<username>:<password>@<hostname>```, so you can get the details from this format. Remember that the Atlas cluster should be in the same region and hosted on the same cloud provider for the Kafka connector to be able to communicate with it. You have to add your Confluent cluster static IP address into the firewall’s allowlist of MongoDB Atlas to allow the connections to your Altas cluster from Confluent Cloud. For non-prod environments, you can also add 0.0.0.0/0 to allow access from anywhere, but it is not recommended for a production environment as it is a security concern allowing any IP access.
![Configuring the MongoDB Atlas Sink connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltdb39db51ec23fe38/6576ded38e05b95a5689ab6c/Configure_Add_Atlas_Sink_Connector.png)
 5. You can select the Kafka input message type as JSON as in the case of the source connector and move to the final review page to view the configuration and cost for your new sink connector.
![Reviewing configurating of MongoDB Atlas Sink connector](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt0adf57836c9d8d1d/6576ded338d46b66977cafd4/Atlas_Sink_Connector_Review_Page.png)
 6. Once the connector has started, you can query the collection mentioned in your sink connector configuration and you would be able to see the data from your PostgreSQL table in the new collection of your MongoDB Atlas cluster.
### Validating PostgreSQL to Atlas data migration
This data is synced in real-time from PostgreSQL to MongoDB Atlas using the source and sink connectors, so if you try adding a new record or updating/deleting existing records in PostgreSQL, you can see it reflect real-time in your MongoDB Atlas cluster collection, as well.

If your data set is huge, the connectors will catch up and process all the data in due time according to the data size. After completion of the data transfer, you can validate your MongoDB Atlas DB and stop the data flow by stopping the source and sink connectors directly from the Confluent Cloud Interface.

Using Kafka, not only can you sync the data using its event-driven architecture, but you can also transform the data in transfer in real-time while migrating it from PostgreSQL to MongoDB. For example, if you would like to rename a field or concat two fields into one for the new collection in Atlas, you can do that while configuring your MongoDB Atlas sink connector.

Let’s say PostgreSQL had the fields “firstname” and “lastname” for your “users” table, and in MongoDB Atlas post-migration, you only want the “name” field which would be a concatenation of the two fields. This can be done using the “transform” attribute in the sink connector configuration. This provides a list of transformations to apply to your data before writing it to the database. Below is an example configuration.
```json
{
 "name": "mongodb-atlas-sink",
 "config": {
 "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
 "tasks.max": "1",
 "topics": "your-topic-name",
 "connection.uri": "mongodb+srv://<username>:<password>@cluster.mongodb.net/test",
 "database": "your-database",
 "collection": "your-collection",
 "key.converter": "org.apache.kafka.connect.storage.StringConverter",
 "value.converter": "org.apache.kafka.connect.json.JsonConverter",
 "value.converter.schemas.enable": "false",
 "transforms": "addFields,unwrap",
 "transforms.addFields.type": "org.apache.kafka.connect.transforms.InsertField$Value",
 "transforms.addFields.static.field": "name",
 "transforms.addFields.static.value": "${r:firstname}-${r:lastname}",
 "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
 "transforms.unwrap.drop.tombstones": "false",
 "transforms.unwrap.delete.handling.mode": "none"
 }
}
```

## Relational Migrator: an intro
As we are discussing data migration from relational to MongoDB, it’s worth mentioning the MongoDB Relational Migrator. This is a tool designed natively by MongoDB to simplify the process of moving data from relational databases into MongoDB. Relational Migrator analyzes your relational schema and gives recommendations for mapping to a new MongoDB schema.

Its features — including schema analysis, data extraction, indexing, and validation — make it a valuable asset for organizations seeking to harness the benefits of MongoDB's NoSQL platform while preserving their existing relational data assets. Whether for application modernization, data warehousing, microservices, or big data analytics, this tool is a valuable asset for those looking to make the shift from relational to NoSQL databases. It helps to migrate from major relational database technologies including Oracle, SQL Server, MySQL, and PostgreSQL.

Get more information and download and use [relational migrator]( https://www.mongodb.com/products/relational-migrator).

## Conclusion
In the ever-evolving landscape of data management, MongoDB has emerged as a leading NoSQL database, known for its flexibility, scalability, and document-oriented structure. However, many organizations still rely on traditional relational databases to store their critical data. The challenge often lies in migrating data between these disparate systems efficiently and accurately.

Confluent Kafka acts as a great leverage in this context with its event driven architecture and native support for major database engines including MongoDB Atlas.The source and sink connectors would have inbound and outbound data through Topics and acts as a platform for a transparent and hassle free data migration from relational to MongoDB Atlas cluster.

How to Migrate PostgreSQL to MongoDB With Confluent Kafka

Anaiya Raisinghani

Associate Developer Advocate

This tutorial explains how to configure CDC jobs on your relational data from MySQL Workbench to MongoDB Atlas using MongoDB Relational Migrator and Confluent Cloud.

The Great Continuous Migration: CDC Jobs With Confluent Cloud and Relational Migrator

This tutorial will cover the Atlas Stream Processing set-up for Kafka Connector users. 

The Atlas Stream Processing Set-up Guide for Kafka Connector Users

Jagadish Nallapaneni

[AWS Identity and Access Management (IAM)](https://aws.amazon.com/iam/) is a widely used service that enables developers to control secure access to AWS services and resources. A key benefit of AWS IAM is its ability to make secure authenticated connections without the client having to pass sensitive passwords.

As of version 1.13, the MongoDB Connector for Apache Kafka (Kafka Connector) now supports authentication via AWS IAM. AWS IAM support has been one of the most commonly requested features for the Kafka Connector, and in this tutorial, we will walk through how to use it.

To achieve support for AWS IAM, we added support for overriding the MongoClient interface allowing a developer to provide a *CustomCredentialProvider*. While this provider can be used for any authentication method, this tutorial will focus on building an AWS IAM credential provider.

To illustrate this, we will walk through a simple configuration of an AWS EC2 instance installed with Apache Kafka. This instance will use the [MongoDB Connector for Apache Kafka](https://www.mongodb.com/docs/kafka-connector/current/) to [sink](https://en.wikipedia.org/wiki/Sink_(computing)) data from a Kafka topic to an Atlas cluster, and authenticate using AWS IAM.

## Prerequisites

If you wish to follow along in this article, make sure you have the following:

- An AWS IAM role, “KafkaBlogAtlasRole”. This role will later be mapped as an Atlas database user.
- MongoDB Atlas cluster. If you do not have an Atlas cluster, learn [how to get started with Atlas](https://www.mongodb.com/docs/atlas/getting-started/)..
- AWS EC2 instance
 - Install [Apache Kafka](https://kafka.apache.org/downloads).
 - Configure the EC2 instance with an AWS IAM role, “KafkaConnectBlogRole.”
 - Add a policy to KafkaConnectBlogRole that allows “AssumeRole” on the KafkaBlogAtlasRole.

Example policy:

```json
{
 "Version": "2012-10-17",
 "Statement": [
 {
 "Sid": "KafkaAtlasBlogPolicy",
 "Effect": "Allow",
 "Action": "sts:AssumeRole",
 "Resource": "arn:aws:iam::979559056307:role/KafkaBlogAtlasRole"
 }
 ]
}
```


- Java Development Kit version 8 or later
- A database user in the desired MongoDB Atlas cluster

 - Configure the database user with AWS IAM and use the AWS Role ARN for KafkaBlogAtlasRole. The Atlas database user should have access to read/write to the desired collection.

![MongoDB Atlas edit user dialog showing AWS IAM authentication method][1]

At this point, we are ready to start building the custom authentication provider.

## Getting started

Starting with version 1.13 of the MongoDB Connector for Apache Kafka, we now have the ability to plug in a custom authentication provider. In this article, we will create a custom authentication provider in Java, compile it, and add it to the lib directory of your Kafka deployment. The provider we will create will support AWS IAM authentication. 

### Building the custom authentication component

To start, in your favorite Java editor, create a project and build the following JAR that contains this source:

SampleAssumeRoleCredential.java

```java
package com.mongodb;

import java.util.Map;
import java.util.function.Supplier;

import com.mongodb.kafka.connect.util.custom.credentials.CustomCredentialProvider;

import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceAsyncClientBuilder;
import com.amazonaws.services.securitytoken.model.AssumeRoleRequest;
import com.amazonaws.services.securitytoken.model.AssumeRoleResult;
import com.amazonaws.services.securitytoken.model.Credentials;
import com.amazonaws.util.StringUtils;

public class SampleAssumeRoleCredential implements CustomCredentialProvider {

 public SampleAssumeRoleCredential() {}
 @Override
 public MongoCredential getCustomCredential(Map<?, ?> map) {
 AWSCredentialsProvider provider = new DefaultAWSCredentialsProviderChain();
 Supplier<AwsCredential> awsFreshCredentialSupplier = () -> {
 AWSSecurityTokenService stsClient = AWSSecurityTokenServiceAsyncClientBuilder.standard()
 .withCredentials(provider)
 .withRegion("us-east-1")
 .build();
 AssumeRoleRequest assumeRoleRequest = new AssumeRoleRequest().withDurationSeconds(3600)
 .withRoleArn((String)map.get("mongodbaws.auth.mechanism.roleArn"))
 .withRoleSessionName("Test_Session");
 AssumeRoleResult assumeRoleResult = stsClient.assumeRole(assumeRoleRequest);
 Credentials creds = assumeRoleResult.getCredentials();
 // Add your code to fetch new credentials
 return new AwsCredential(creds.getAccessKeyId(), creds.getSecretAccessKey(), creds.getSessionToken());
 };
 return MongoCredential.createAwsCredential(null, null)
 .withMechanismProperty(MongoCredential.AWS_CREDENTIAL_PROVIDER_KEY, awsFreshCredentialSupplier);
 }

 @Override
 public void validate(Map<?, ?> map) {
 String roleArn = (String) map.get("mongodbaws.auth.mechanism.roleArn");
 if (StringUtils.isNullOrEmpty(roleArn)) {
 throw new RuntimeException("Invalid value set for customProperty");
 }
 }

 @Override
 public void init(Map<?, ?> map) {

 }
}
```

Example pom.xml file

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <groupId>com.mongodb</groupId>
 <artifactId>SampleAssumeRoleCredential</artifactId>
 <version>1.0-SNAPSHOT</version>
 <build>
 <plugins>
 <plugin>
 <groupId>org.apache.maven.plugins</groupId>
 <artifactId>maven-shade-plugin</artifactId>
 <version>3.5.3</version>
 <configuration>
 
 </configuration>
 <executions>
 <execution>
 <phase>package</phase>
 <goals>
 <goal>shade</goal>
 </goals>
 </execution>
 </executions>
 </plugin>
 </plugins>
 </build>

 <dependencies>
 
 
 <dependency>
 <groupId>org.mongodb</groupId>
 <artifactId>mongodb-driver-sync</artifactId>
 <version>5.1.0</version>
 </dependency>
 
 <dependency>
 <groupId>com.amazonaws</groupId>
 <artifactId>aws-java-sdk</artifactId>
 <version>1.12.723</version>
 </dependency>

 
 <dependency>
 <groupId>org.slf4j</groupId>
 <artifactId>slf4j-jdk14</artifactId>
 <version>1.7.28</version>
 </dependency>

 <dependency>
 <groupId>kafka-connect</groupId>
 <artifactId>kafka-connect</artifactId>
 <scope>system</scope>
 <version>1.13.0</version>
 </dependency>
 </dependencies>

 <properties>
 <maven.compiler.source>17</maven.compiler.source>
 <maven.compiler.target>17</maven.compiler.target>
 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 </properties>

</project>
```

Note: We need to reference MongoDB Connector for Apache Kafka version 1.13 or later. In this example, the connector was downloaded from the Kafka Connector GitHub repository and stored in the local Maven repository. 

```bash
mvn install:install-file -Dfile=/Users/robert.walters/Downloads/mongodb-kafka-connect-mongodb-1.13.0/lib/mongo-kafka-connect-1.13.0-confluent.jar -DgroupId=kafka-connect -DartifactId=kafka-connector -Dversion=1.13.0 -Dpackaging=jar\n
```

Compile and place this SampleAssumeRole.jar file in the Kafka libs directory in your Kafka Connect deployment. You are now ready to configure your MongoDB Connector for Apache Kafka to use this custom authentication provider.

### Configure the MongoDB Connector for Apache Kafka

To support the custom provider in version 1.13, the following configuration options were added to the connector:

| Configuration Parameter | Description |
| ----------------------------------------- | ------------------------------------------------------------ |
| mongo.custom.auth.mechanism.enable | Determines if the connector should use the provider class; default is false |
| mongo.custom.auth.mechanism.providerClass | Java class entry point for the custom provider |
| mongodbaws.auth.mechanism.roleArn | Role that has IAM access to the database |

In this example, the custom authentication component will use AWS IAM AssumeRole to authenticate with MongoDB Atlas. Here is an example of this configuration connecting to a MongoDB Atlas cluster.

```json
{
 "name": "mongo-tutorial-sink",
 "config": {
 "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
 "topics": "Stocks2",
 "connection.uri": "<MongoDB Connection String>",
 "key.converter": "org.apache.kafka.connect.storage.StringConverter",
 "value.converter": "org.apache.kafka.connect.json.JsonConverter",
 "value.converter.schemas.enable": false,
 "database": "Stocks",
 "collection": "StockData",
 "mongo.custom.auth.mechanism.enable":"true",
 "mongo.custom.auth.mechanism.providerClass":"com.mongodb.SampleAssumeRoleCredential",
 "mongodbaws.auth.mechanism.roleArn":"<AWS IAM ARN>" 
 }
}
```

Notice that the roleArn is the ARN (Amazon Resource Name) of the AWS IAM role that is used for Atlas users, KafkaBlogAtlasRole. At runtime, since Kafka Connect is running under the AWS IAM role of the EC2 instance, KafkaConnectBlogRole, and this role has a policy that allows AssumeRole on the KafkaBlogAtlasRole, the service can connect to Atlas without storing any credentials!

You are now ready to send data to Kafka and watch it arrive in a MongoDB collection using the MongoDB Connector for Apache Kafka configured with AWS IAM authentication to MongoDB Atlas.

## Summary

In this article, we discussed the requirements needed to support a custom authentication provider such as AWS IAM authentication. Version 1.13 of the Kafka Connector introduced new parameters that define which class to use within the custom provider JAR file. This JAR file includes the logic for the custom authentication. 

Note that while adding AWS IAM authentication capability natively to the Kafka Connector is possible (it is just software, after all), doing so would not only add a version dependency on the AWS IAM SDK, but it would also significantly increase the size of the connector, making it impractical for all other users who do not wish to use this authentication method. For these reasons, writing the custom credential provider is the solution for integrating the Kafka Connector with authentication methods such as AWS IAM.

Read the [MongoDB Connector for Apache Kafka documentation](https://www.mongodb.com/docs/kafka-connector/current/) to learn more. Questions? Join us in the [MongoDB Developer Community](https://www.mongodb.com/community/forums/).


 [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb44f9cef8df8f2b2/6682f292b1132488c03a422b/image1.png

To connect to an Atlas cluster, users authenticate with AWS IAM authentication. Learn how to set up the MongoDB Connector for Apache Kafka for authentication.

Using AWS IAM Authentication with the MongoDB Connector for Apache Kafka

Timothy Marland

Enterprises around the world are looking to modernize their existing applications. They need a streamlined way to synchronize data from devices at the Edge into their cloud data stores. Whether their goals are business growth or fending off the competition, application modernization is the primary vehicle that will help them get there. 

Often the first step in this process is to move data from an existing relational database repository (like Oracle, SQL Server, DB2, or Postgres, for example) into a JSON-based flexible database in the cloud (like MongoDB, Aerospike, Couchbase, Cassandra, or DocumentDB). Sounds simple, right? I mean, really, if JSON (NoSQL) is so simple and flexible, why would data migration be hard? There must be a bunch of automated tools to facilitate this data migration, right? 

Unfortunately, the answers are “Not really,” “Because data synchronization is rarely simple,” and “The available tools are often DIY-based and don’t provide nearly the level of automation required to facilitate an ongoing, large-scale, production-quality, conflict-resolved data synchronization.”

## Why is this so complex?

### Data modeling

One of the first challenges is data modeling. To effectively leverage the benefits inherent in a JSON-based schema, you need to include data modeling as part of your migration strategy. Simply flattening or de-normalizing a relational schema into nested JSON structures, or worse yet, simply moving from relational to JSON without any data modeling consideration, results in a JSON data repository that is slow, inefficient, and difficult to query. You need an intelligent data modeling platform that automatically creates the most effective JSON structures based on your application needs and the target JSON repository without requiring specialized resources like data scientists and data engineers. 

### Building and monitoring pipelines

Once you’ve mapped the data, you need tools that allow you to build reliable, scalable data pipelines to move the data from the source to the target repository. Sadly, most of the tools available today are primarily DIY scripting tools that require both custom (often complex) coding to transform the data to the new schema properly and custom (often complex) monitoring to ensure that the new data pipelines are working reliably. You need a data pipeline automation and monitoring platform to move the data and ensure its quality. 

### DIY is hard

This process of data synchronization, pipeline automation, and monitoring is where most application modernization projects get bogged down and/or ultimately fail. These failed projects often consume significant resources before they fail, as well as affect the overall business functionality and outcomes, and lead to missed objectives. 

## CDC: MongoDB Atlas, Atlas Device Sync, and Dataworkz

Synchronizing data between edge devices and various databases can be complex. Simplifying this is our goal, and we will demonstrate how to achieve bi-directional synchronization between mobile devices at MySQL in the cloud using MongoDB Atlas Device Sync and Dataworkz.

Let's dive in.

## Prerequisites 

- Accounts with [MongoDB Atlas](https://www.mongodb.com/cloud/atlas/register) (this can be tested on free tiers), [Dataworkz](http://dataworkz.com/), and Google Cloud
- Kafka
- Debezium

## Step 1: prepare your mobile application with Atlas Device Sync

Set up a template app for this test by following the [steps outlined in the docs](https://www.mongodb.com/docs/atlas/app-services/get-started/#tutorials). Once that step is complete, you will have a mobile application running locally, with automated synchronization back to MongoDB Atlas using the Atlas Device Sync SDK. 

## Step 2: set up a source database and target MongoDB Atlas Collection

We used GCP in us-west1-a and Cloud MySQL for this example. Be sure to include [sample data](https://www.mongodb.com/docs/guides/atlas/sample-data/).

### Check if BinLog replication is already enabled

![Check if BinLog Replication is Enabled][1]

1. Access your Google Cloud MySQL instance.
2. In the left navigation pane, under **Databases**, select **Backups**.
3. Access your Google Cloud MySQL instance.
4. In **Settings**, check the status of **Automated backups**.
5. If the status is *Disabled*, follow the steps below.

### Enable BinLog replication

These steps are necessary to set up some of the background processes which enable ongoing two-way synchronization.

1. In **Backups**, **Settings**, click **+ CREATE BACKUP**.
2. In the **Edit backup settings** dialog:
    - Select the **Automate backups** and **Enable point-in-time recovery** check boxes.
    - Click **SAVE** to activate the changes.

![Enable Automated Backups][2]

3. Check the status of **Automated backups** again under **Settings**. It should appear as shown below. This step is important because the metadata created in the backup process is used by the tools to help enable bi-directional synchronization.

![GCP Backup Settings][3]

### Step 3: configure connector for MySQL 

The connector for MySQL needs to be defined in Dataworkz.

![Configure Dataworkz for MySQL][4]

- Log in to the Dataworkz application. 
- Go to **Configuration** -> **Databases** -> **MySQL**. 
- Click the + icon in the above image to add the configuration. 
- It will open the MySQL configuration screen. 

![Connect to MySQL in Dataworkz][5]

- Provide the following details in the configuration screen: 
    - MySQL connector name 
    - Host server and port 
    - User name 
    - Password 
    - Select Deployment Type 
    - Select TLS/SSL option as appropriate 
- Click **Save** to add the MySQL configuration. 

### Step 4: configure connector for MongoDB 

The connector for MongoDB Atlas needs to be defined in Dataworkz. 

![MongoDB Atlas definition in Dataworkz][6]

- Log in to the Dataworkz application. 
- Go to **Configuration** -> **Databases** -> **MongoDB.**
- Click the + icon in the above image to add the configuration. 
- It will open the MongoDB configuration screen. 

![MongoDB configuration screen in Dataworkz][7]

- Provide the following details in the configuration screen (these can be accessed by clicking on the **Connect** button displayed below).
    - MongoDB Connector Name 
    - Host server 
    - User name 
    - Password 

![MongoDB Atlas Connect button display on overview screen][8]

- Click **Save** to add the MongoDB configuration. 

### Step 5: set up Kafka

The connector for Kafka needs to be defined in Dataworkz.

![Kafka connection configuration in Dataworkz][9]

- Log in to the Dataworkz application.
- Go to **Configuration** -> **Streaming Systems** -> **Kafka**. 
- Click the + icon in the above image to add the configuration. 
- It will open the Kafka configuration screen. 

![Kafka connection configuration details in Dataworkz][10]

- Provide the following details in the configuration screen: 
    - Kafka Connector Name 
    - Bootstrap server name and port (as per the format in the image below) 
    - Zookeeper (optional)
- Click **Save** to add the Kafka configuration. 

### Step 6: configure CDC for MySQL 

Configure CDC for MySQL in Dataworkz by following these steps:

![MySQL CDC configuration in Dataworkz][11]

- Log in to the Dataworkz application. 
- Go to **Configuration** -> **Change Data Capture** -> **MySQL**. 
- Click the + icon in the above image to add the configuration. 
- It will open the CDC configuration screen. 

![Dataworkz CDC configuration screen][12]

- Provide the details in the configuration detail screen: 
    - Choose the Workspace. 
    - Choose Kafka (which was configured in the previous step).
    - Add the Debezium server name. 
    - Choose the MYSQL Storage that was previously configured. 
- Select the appropriate database from the list. 
- Choose the Database table for Sync.

![Dataworkz CDC configuration screen][13]

You can see the configured MySQL CDC in the list of configurations. 

![Dataworkz MySQL CDC configuration showing up in list][14]

### Step 7: data preparation — JSON modeling 

This step enables the modeling of the results of RDBMS in JSON structure. To do this, select **Data Prep** -> **JSON Modeller** inside the Dataworkz settings.

![Dataworkz settings JSON modeling setup][15]

![Dataworkz settings JSON modeling detail screen][16]

- Click on **Select root dataset for json modeling** (above image).

![Workspace selection screen for CDC][17]

- Select the Workspace in which MySQL CDC was configured. 
- Select cdc_kafka as the Source Collection. 
- Choose root/base table/dataset. 
- Click on the **Enable Bi-directional sync** checkbox. 
- Select a sync field from the list of the columns that will be used to determine the last updated time for a row. 
- Select columns for the target dataset. 
- Click **Update** and you will see the below screen. 

![JSON modeling screen in Dataworkz][18]

- Choose the desired mapping with the base table by right-clicking. 
    - If the table has one-to-many mapping, choose **Create array document**. 
    - If the table has one-to-one mapping, choose **Create field in child document** 
- Selecting **create array document** would require selecting a child table relationship (for creation of an array document).
- Choose **Child Table** in the screen that pops up. 

![Dataworkz array document child table screen][19]

- Choose **join key** in the **Relationship mapping** section. 
- In the **Unique Key** section, choose the unique key for the child table (used for update or delete events). 
- Select a sync field from the list of the columns that will be used to determine the last updated time for a row. 
- Click **update**. 
- A one-to-many relationship shows up. 

![Dataworkz JSON model detail screen][20]

- Click on **Save** and choose the target destination (MongoDB Collection, in this case).

![Dataworkz JSON Upsert and key settings][21]

- Choose **Transfer Mode** as Upsert and select the unique key for the base table in the pop-up box. 

![Dataworkz custom mapping settings screen][22]

- Choose the desired job name and select the job frequency as **Recurring**. Set this job as **Continuous** from the **Choose Frequency** section. 

![Dataworkz job frequency setitngs screen][23]

### Step 8: monitoring/maintaining the sync job 

The sync job that was scheduled in the previous step can be viewed under **Jobs** -> **Continuous Job**” in the Dataworkz portal. It provides the following: 

- job status (running/deleted/paused) 
- ability to pause a running job 
- ability to resume a paused job 
- ability to delete a continuous job

![Dataworkz sync job summary screen][24]

## Conclusion

MongoDB Atlas Device Sync streamlines and automates the challenges of mobile data synchronization with Atlas. By integrating with Dataworkz, that same data can now be automatically and bi-directionally synchronized with MySQL, Oracle and other databases.

Set up your account at [cloud.mongodb.com](https://www.mongodb.com/products/platform/cloud) and [dataworkz.com](https://www.dataworkz.com/) to create accounts and begin your automated bi-directional data synchronization journey.


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt85b18bd98135559d/65c54454245ed9597d91062b/1.jpg
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt0f313395e6bb7e0a/65c5447cf0270544bcea8b0f/2.jpg
  [3]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt42fc95f74b034c2c/65c54497ab9c0fe8aab945fc/3.jpg
  [4]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt0bf05ceea39406ef/65c544b068e9235c39e585bc/4.png
  [5]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt852256fbd6c92a1d/65c544cf245ed9f5af910634/5.png
  [6]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt6f7a22db7bfeb113/65c544f78b3a0d12277c6c70/6.png
  [7]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt732e20dad37e46e1/65c54515fb34d04d731b1a80/7.png
  [8]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltc1f83c67ed8bf7e5/65c5453625aa94148c3513c5/8.png
  [9]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltcfe7cf5ed073b181/65c5454a49edef40e16a355a/9.png
  [10]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt99688601bdbef64a/65c5455d211bae4eaea55ad0/10.png
  [11]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt7317f754880d41ba/65c545780acbc5455311135a/11.png
  [12]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt690ff6f3783cb06f/65c5458b68e92372a8e585d2/12.png
  [13]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt0810a0d139692e45/65c545a5ab9c0fdc87b9460a/13.png
  [14]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt39b298970485dc92/65c545ba4cd37037ee70ec51/14.png
  [15]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt6596816400af6c46/65c545d4eed32eadf6ac449d/15.png
  [16]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt4eaf0f6c5b0ca4a5/65c545e3ff4e591910ad0ed6/16.png
  [17]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt9ffa8c37f5e26e5f/65c545fa4cd3709a7870ec56/17.png
  [18]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt1678582be8e8cc07/65c5461225aa943f393513cd/18.png
  [19]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltd4b2f6a8e389884a/65c5462625aa94d9b93513d1/19.png
  [20]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltbd4a6e31ba6103f6/65c5463fd4db5559ac6efc99/20.png
  [21]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt6e3be8f5f7f0e0f1/65c546547998dae7b86b5e4b/21.png
  [22]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltd2fccad909850d63/65c54669d2c66186e28430d5/22.png
  [23]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blted07a5674eae79e0/65c5467d8b3a0d226c7c6c7f/23.png
  [24]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt8b8b5ce479b8fc7b/65c5469125aa94964c3513d5/24.png

Learn how to set up automated, automated, bi-directional synchronization of data from mobile devices to MongoDB Atlas  and Google Cloud MySQL.

Synchronize Your Mobile Application With MongoDB Atlas and Google Cloud MySQL

## Overview
When you are the best of breed, you have many imitators. MongoDB is no different in the database world. If you are reading this blog, you are most likely an Azure customer that ended up using CosmosDB. 

You needed a database that could handle unstructured data in Azure and eventually realized CosmosDB wasn’t the best fit. Perhaps you found that it is too expensive for your workload or not performing well or simply have no confidence in the platform. You also might have tried using the MongoDB API and found that the queries you wanted to use simply don’t work in CosmosDB because it fails [67% of the compatibility tests](https://www.mongodb.com/cloud/atlas/compare). 

Whatever the path you took to CosmosDB, know that you can easily migrate your data to MongoDB Atlas while still leveraging the full power of Azure. With MongoDB Atlas in Azure, there are no more failed queries, slow performance, and surprise bills from [not optimizing your RDUs](https://docs.microsoft.com/en-us/azure/cosmos-db/cosmos-db-reserved-capacity). MongoDB Atlas in Azure also gives you access to the latest releases of MongoDB and the flexibility to leverage any of the three cloud providers if your business needs change.

Note: When you originally created your CosmosDB, you were presented with these API options:
![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/which_api_best_suits_your_workload_aa9a89b977.png)

If you created your CosmosDB using Azure Cosmos DB API for MongoDB, you can use [mongo tools](https://docs.mongodb.com/database-tools/) such as mongodump, mongorestore, mongoimport, and mongoexport to move your data. The [Azure CosmosDB Connector](https://www.confluent.io/hub/microsoftcorporation/kafka-connect-cosmos) for Kafka Connect does not work with CosmosDB databases that were created for the Azure Cosmos DB API for MongoDB.

In this blog post, we will cover how to leverage Apache Kafka to move data from Azure CosmosDB Core (Native API) to MongoDB Atlas. While there are many ways to move data, using Kafka will allow you to not only perform a one-time migration but to stream data from CosmosDB to MongoDB. This gives you the opportunity to test your application and compare the experience so that you can make the final application change to MongoDB Atlas when you are ready. The complete example code is available in this [GitHub repository](https://github.com/RWaltersMA/CosmosDB2MongoDB).
![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/apache_kafka_connect_335ed70aa3.png)
## Getting started
You’ll need access to an Apache Kafka cluster. There are many options available to you, including Confluent Cloud, or you can deploy your own Apache Kafka via Docker as shown in this blog. Microsoft Azure also includes an event messaging service called Azure Event Hubs. This service provides a Kafka endpoint that can be used as an alternative to running your own Kafka cluster. Azure Event Hubs exposes the same Kafka Connect API, enabling the use of the MongoDB connector and Azure CosmosDB DB Connector with the Event Hubs service.

If you do not have an existing Kafka deployment, perform these steps. You will need docker installed on your local machine:
```
git clone ​​https://github.com/RWaltersMA/CosmosDB2MongoDB.git
```
Next, build the docker containers.
```
docker-compose up -d --build
```

The docker compose script (docker-compose.yml) will stand up all the components you need, including Apache Kafka and Kafka Connect. Install the CosmosDB and MongoDB connectors.
## Configuring Kafka Connect
Modify the **cosmosdb-source.json** file and replace the placeholder values with your own.
```
{
 "name": "cosmosdb-source",
 "config": {
 "connector.class": "com.azure.cosmos.kafka.connect.source.CosmosDBSourceConnector",
 "tasks.max": "1",
 "key.converter": "org.apache.kafka.connect.json.JsonConverter",
 "value.converter": "org.apache.kafka.connect.json.JsonConverter",
 "connect.cosmos.task.poll.interval": "100",
 "connect.cosmos.connection.endpoint": 
"https://**<cosmosinstance-name>**.documents.azure.com:443/",
 "connect.cosmos.master.key": **"<cosmosdbprimarykey>",**
 "connect.cosmos.databasename": **"<database name>",**
 "connect.cosmos.containers.topicmap": **"<containers>#<topicname>”,**
 "connect.cosmos.offset.useLatest": false,
 "value.converter.schemas.enable": "false",
 "key.converter.schemas.enable": "false"
 }
}

```
Modify the **mongo-sink.json** file and replace the placeholder values with your own.
```
{"name": "mongo-sink",
 "config": {
 "connector.class":"com.mongodb.kafka.connect.MongoSinkConnector",
 "tasks.max":"1",
 "topics":"<topicname>",
 "connection.uri":"<MongoDB Atlas Connection String>",
 "database":"<Desired Database Name>",
 "collection":"<Desired Collection Name>",
 "key.converter": "org.apache.kafka.connect.json.JsonConverter",
 "value.converter":"org.apache.kafka.connect.json.JsonConverter",
 "value.converter.schemas.enable": "false",
 "key.converter.schemas.enable": "false"
 
 }}

```
Note: Before we configure Kafka Connect, make sure that your network settings on both CosmosDB and MongoDB Atlas will allow communication between these two services. In CosmosDB, select the Firewall and Virtual Networks. While the easiest configuration is to select “All networks,” you can provide a more secure connection by specifying the IP range from the Firewall setting in the Selected networks option. MongoDB Atlas Network access also needs to be configured to allow remote connections. By default, MongoDB Atlas does not allow any external connections. See [Configure IP Access List](https://docs.atlas.mongodb.com/security/ip-access-list/) for more information.

To configure our two connectors, make a REST API call to the Kafka Connect service:

```
curl -X POST -H "Content-Type: application/json" -d @cosmosdb-source.json http://localhost:8083/connectors


curl -X POST -H "Content-Type: application/json" -d @mongodb-sink.json http://localhost:8083/connectors

```
That’s it!

Provided the network and database access was configured properly, data from your CosmosDB should begin to flow into MongoDB Atlas. If you don’t see anything, here are some troubleshooting tips:

* Try connecting to your MongoDB Atlas cluster using the mongosh tool from the server running the docker container.
* View the docker logs for the Kafka Connect service.
* Verify that you can connect to the CosmosDB instance using the Azure CLI from the server running the docker container.


**Summary**
In this post, we explored how to move data from CosmosDB to MongoDB using Apache Kafka. If you’d like to explore this method and other ways to migrate data, check out the 2021 MongoDB partner of the year award winner, [Peerslands'](https://www.peerislands.io/), five-part blog post on [CosmosDB migration](https://www.peerislands.io/azure-cosmos-db-for-mongo-series-part-1-setting-up-of-cosmosdb/).

Learn how to migrate your data in Azure CosmosDB to MongoDB Atlas using Apache Kafka.

Migrate from Azure CosmosDB to MongoDB Atlas Using Apache Kafka

Data and event-driven applications are in high demand in a large variety of industries. With this demand, there is a growing challenge with how to sync the data across different data sources. 

A widely adopted solution for communicating real-time data transfer across multiple components in organization systems is implemented via clustered queues. One of the popular and proven solutions is [Apache Kafka](https://kafka.apache.org/).

The Kafka cluster is designed for streams of data that sequentially write events into commit logs, allowing real-time data movement between your services. Data is grouped into topics inside a Kafka cluster.

MongoDB provides a [Kafka connector](https://www.mongodb.com/kafka-connector) certified by [Confluent](https://www.confluent.io/), one of the largest Kafka providers. With the Kafka connector and Confluent software, you can publish data from a MongoDB cluster into Kafka topics using a source connector. Additionally, with a sink connector, you can consume data from a Kafka topic to persist directly and consistently into a MongoDB collection inside your MongoDB cluster.

In this article, we will provide a simple step-by-step guide on how to connect a remote Kafka cluster—in this case, a Confluent Cloud service—with a [MongoDB Atlas](https://mongodb.com/atlas) cluster. For simplicity purposes, the installation is minimal and designed for a small development environment. However, through the article, we will provide guidance and links for production-related considerations.

> **Pre-requisite**: To avoid JDK [known](https://docs.mongodb.com/drivers/java/sync/current/faq/#how-do-i-fix---javax.net.ssl.sslhandshakeexception--extension--5--should-not-be-presented-in-certificate_request--) certificate issues please update your JDK to one of the following patch versions or newer:
> - JDK 11.0.7+
> - JDK 13.0.3+
> - JDK 14.0.2+

## Table of Contents

1. [Create a Basic Confluent Cloud Cluster](#create-kafka-cluster)
1. [Create an Atlas Project and Cluster](#create-atlas-cluster) 
1. [Install Local Confluent Community Binaries to Run a Kafka Connect Instance](#install-kafka-connect)
1. [Configure the MongoDB Connector with Kafka Connect Locally](#configure-connectors)
1. [Start and Test Sink and Source MongoDB Kafka Connectors](#test-connectors)
1. [Summary](#summary)

## <a name="create-kafka-cluster"></a>Create a Basic Confluent Cloud Cluster

We will start by [creating](https://docs.confluent.io/cloud/current/clusters/create-cluster.html) a basic Kafka cluster in the [Confluent Cloud](https://confluent.cloud/). 

Once ready, create a topic to be used in the Kafka cluster. I created one named “orders.”

This “orders” topic will be used by Kafka Sink connector. Any data in this topic will be persisted automatically in the Atlas database.

![Kafka Cluster Topics](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/kafka_Topics_7eb45621e7.png)

You will also need another topic called "outsource.kafka.receipts". This topic will be used by the MongoDB Source connector, streaming reciepts from Atlas database.

[Generate](https://docs.confluent.io/cloud/current/access-management/authenticate/api-keys/api-keys.html) an `api-key` and `api-secret` to interact with this Kafka cluster. For the simplicity of this tutorial, I have selected the “Global Access” api-key. For production, it is recommended to give as minimum permissions as possible for the api-key used. Get a hold of the generated keys for future use.

Obtain the Kafka cluster connection string via `Cluster Overview > Cluster Settings > Identification > Bootstrap server` for future use. Basic clusters are open to the internet and in production, you will need to amend the access list for your specific hosts to connect to your cluster via advanced cluster ACLs.

## <a name="create-atlas-cluster"></a>Create a MongoDB Atlas Project and Cluster

[Create a project and cluster](https://docs.atlas.mongodb.com/getting-started/) or use an existing Atlas cluster in your project. 

![M0 Atlas Cluster](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_13_at_14_04_39_5fc276d2d6.png)

Prepare your Atlas cluster for a kafka-connect [connection](https://docs.mongodb.com/kafka-connector/current/introduction/connect/). Inside your project’s access list, enable user and relevant IP addresses of your local host, the one used for Kafka Connect binaries. Finally, get a hold of the Atlas [connection string](https://docs.atlas.mongodb.com/connect-to-cluster/#use-the-connect-to-your-cluster-dialog-to-connect-to-your-cluster) for future use.

## <a name="install-kafka-connect"></a>Install a Kafka Connect Worker

[Kafka Connect](https://docs.confluent.io/platform/current/connect) is one of the mechanisms to reliably stream data between different data systems and a Kafka cluster. For production use, we recommend using a [distributed deployment](https://docs.confluent.io/home/connect/self-managed/userguide.html#standalone-vs-distributed-mode) for high availability, fault tolerance, and scalability. There is also a cloud version to [install the connector](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html) on the Confluent Cloud.

For this simple tutorial, we will use a standalone local Kafka Connect [installation](https://docs.confluent.io/platform/current/installation/installing_cp/zip-tar.html).

To have the binaries to install kafka-connect and all of its dependencies, let’s download the files:
```shell 
curl -O http://packages.confluent.io/archive/7.0/confluent-community-7.0.1.tar.gz
tar -xvf confluent-community-7.0.1.tar.gz
```

## <a name="configure-connectors"></a> Configure Kafka Connect

Configure the plugins directory where we will host the MongoDB Kafka Connector plugin:
```shell
mkdir -p /usr/local/share/kafka/plugins
```


Edit the `<confluent-install-dir>/etc/schema-registry/connect-avro-standalone.properties` using the content provided below. Ensure that you replace the `<kafka-cluster>:<kafka-port>` with information taken from Confluent Cloud bootstrap server earlier. 

Additionally, replace the generated `<api-key>` and `<api-secret>` taken from Confluent Cloud in every section.
```
bootstrap.servers=<kafka-cluster>:<kafka-port>

Connect data. Every Connect user will
# need to configure these based on the format they want their data in when loaded from or stored into Kafka
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter you want to apply
# it to
key.converter.schemas.enable=false
value.converter.schemas.enable=false

# The internal converter used for offsets and config data is configurable and must be specified, but most users will
# always want to use the built-in default. Offset and config data is never visible outside of Kafka Connect in this format.
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

# Store offsets on local filesystem
offset.storage.file.filename=/tmp/connect.offsets
# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000

ssl.endpoint.identification.algorithm=https


sasl.mechanism=PLAIN
request.timeout.ms=20000
retry.backoff.ms=500
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="<api-key>" password="<api-secret>";
security.protocol=SASL_SSL

consumer.ssl.endpoint.identification.algorithm=https
consumer.sasl.mechanism=PLAIN
consumer.request.timeout.ms=20000
consumer.retry.backoff.ms=500
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="<api-key>" password="<api-secret>";
consumer.security.protocol=SASL_SSL

producer.ssl.endpoint.identification.algorithm=https
producer.sasl.mechanism=PLAIN
producer.request.timeout.ms=20000
producer.retry.backoff.ms=500
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="<api-key>" password="<api-secret>";
producer.security.protocol=SASL_SSL

plugin.path=/usr/local/share/kafka/plugins
```

**Important**: Place the `plugin.path` to point to our plugin directory with permissions to the user running the kafka-connect process.

### Install the MongoDB connector JAR: 
Download the “[all](https://docs.mongodb.com/kafka-connector/current/introduction/install/#install-the-connector-on-confluent-platform)” jar and place it inside the plugin directory.

```shell
cp ~/Downloads/mongo-kafka-connect-1.6.1-all.jar /usr/local/share/kafka/plugins/
```
### Configure a MongoDB Sink Connector

The [MongoDB Sink connector](https://docs.mongodb.com/kafka-connector/current/sink-connector/) will allow us to read data off a specific Kafka topic and write to a MongoDB collection inside our cluster. Create a MongoDB sink connector properties file in the main working dir: `mongo-sink.properties` with your Atlas cluster details replacing `<username>:<password>@<atlas-cluster>/<database>` from your Atlas connect tab. The working directory can be any directory that the `connect-standalone` binary has access to and its path can be provided to the `kafka-connect` command shown in ["Start Kafka Connect and Connectors"](#test-connectors) section.

```
name=mongo-sink
topics=orders
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1
connection.uri=mongodb+srv://<username>:<password>@<atlas-cluster>/<database>?retryWrites=true&w=majority
database=kafka
collection=orders
max.num.retries=1
retries.defer.timeout=5000
```

With the above configuration, we will listen to the topic called “orders” and publish the input documents into database `kafka` and collection name `orders`. 

### Configure Mongo Source Connector

The [MongoDB Source connector](https://docs.mongodb.com/kafka-connector/current/source-connector/) will allow us to read data off a specific MongoDB collection topic and write to a Kafka topic. When data will arrive into a collection called `receipts`, we can use a source connector to transfer it to a Kafka predefined topic named “outsource.kafka.receipts” (the configured prefix followed by the `<database>.<collection>` name as a topic—it's possible to use advanced mapping to change that). 

Let’s create file `mongo-source.properties` in the main working directory:
```
name=mongo-source
connector.class=com.mongodb.kafka.connect.MongoSourceConnector
tasks.max=1

# Connection and source configuration
connection.uri=mongodb+srv://<username>:<password>@<atlas-cluster>/<database>?retryWrites=true&w=majority
database=kafka
collection=receipts

topic.prefix=outsource
topic.suffix=
poll.max.batch.size=1000
poll.await.time.ms=5000

# Change stream options
pipeline=[]
batch.size=0
change.stream.full.document=updateLookup
publish.full.document.only=true
collation=
```

The main properties here are the database, collection, and aggregation pipeline used to listen for incoming changes as well as the connection string. The `topic.prefix` adds a prefix to the `<database>.<collection>` namespace as the Kafka topic on the Confluent side. In this case, the topic name that will receive new MongoDB records is “outsource.kafka.receipts” and was predefined earlier in this tutorial.

I have also added `publish.full.document.only=true` as I only need the actual document changed or inserted without the change stream event wrapping information.

### Start Kafka Connect and Connectors

For simplicity reasons, I am running the standalone Kafka Connect in the foreground.

```
 ./confluent-7.0.1/bin/connect-standalone ./confluent-7.0.1/etc/schema-registry/connect-avro-standalone.properties mongo-sink.properties mongo-source.properties
```

> **Important**: Run with the latest Java version to avoid JDK SSL bugs.

Now every document that will be populated to topic “orders” will be inserted into the `orders` collection using a sink connector. A source connector we configured will transmit every receipt document from `receipt` collection back to another topic called "outsource.kafka.receipts" to showcase a MongoDB consumption to a Kafka topic.

## <a name="test-connectors"></a> Publish Documents to the Kafka Queue

Through the Confluent UI, I have submitted a test document to my “orders” topic.
![Produce data into "orders" topic](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_29_43_befe30feed.png)

### Atlas Cluster is Being Automatically Populated with the Data

Looking into my Atlas cluster, I can see a new collection named `orders` in the `kafka` database.

![Orders collection](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_31_56_9a6f41a74d.png)

Now, let's assume that our application received the order document from the `orders` collection and produced a receipt. We can replicate this by inserting a document in the `kafka.reciepts` collection:

![Recipets collection](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_33_22_0763e1d5a7.png)

This operation will cause the source connector to produce a message into “outsource.kafka.reciepts” topic.
### Kafka "outsource.kafka.reciepts" Topic

![Recieved data into "outsource.kafka.reciepts" topic](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screen_Shot_2021_12_21_at_15_33_56_7ab50fe570.png)

Log lines on kafka-connect will show that the process received and published the document: 

```
[2021-12-14 15:31:18,376] INFO [mongo-source|task-0] [Producer clientId=connector-producer-mongo-source-0] Cluster ID: lkc-65rmj (org.apache.kafka.clients.Metadata:287)
[2021-12-14 15:31:18,675] INFO [mongo-source|task-0] Opened connection [connectionId{localValue:21, serverValue:99712}] to dev-shard-00-02.uvwhr.mongodb.net:27017 (org.mongodb.driver.connection:71)
[2021-12-14 15:31:18,773] INFO [mongo-source|task-0] Started MongoDB source task (com.mongodb.kafka.connect.source.MongoSourceTask:203)
[2021-12-14 15:31:18,773] INFO [mongo-source|task-0] WorkerSourceTask{id=mongo-source-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask:233)
[2021-12-14 15:31:27,671] INFO [mongo-source|task-0|offsets] WorkerSourceTask{id=mongo-source-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505
[2021-12-14 15:31:37,673] INFO [mongo-source|task-0|offsets] WorkerSourceTask{id=mongo-source-0} flushing 1 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505)
```

## <a name="summary"></a>Summary

In this how-to article, I have covered the fundamentals of building a simple yet powerful integration of MongoDB Atlas to Kafka clusters using MongoDB Kafka Connector and Kafka Connect.

This should be a good starting point to get you going with your next event-driven application stack and a successful integration between MongoDB and Kafka.

Try out [MongoDB Atlas](https://mongodb.com/atlas) and [Kafka connector ](https://www.mongodb.com/kafka-connector) today!

A simple step-by-step tutorial on how to use MongoDB Atlas with a Kafka Connector and connect it to any Remote Kafka Cluster.

Kafka

Featured

Learn to Build AI-Enhanced Retail Search Solutions with MongoDB and Databricks

The Great Continuous Migration: CDC Jobs With Confluent Cloud and Relational Migrator

The Atlas Stream Processing Set-up Guide for Kafka Connector Users

All Kafka Content