Juan Soto

Senior Consulting Engineer

Robert Walters

Senior Product Manager

With today’s need of flexible event-driven architectures, companies across the globe choose best of breed technologies like MongoDB and Apache Kafka to help solve these challenges.  While these two complementary technologies provide the power and flexibility to solve these large scale challenges, performance has always been at the forefront of concerns. In this blog, we will cover how to measure performance of the MongoDB Connector for Apache Kafka in both a source and sink configuration. 

## Measuring Sink Performance

Recall that the MongoDB sink connector writes data from a Kafka topic into MongoDB. Writes by default use the [ReplaceOneModel](https://mongodb.github.io/mongo-java-driver/4.2/apidocs/mongodb-driver-core/com/mongodb/client/model/ReplaceOneModel.html) where the data is either updated if it's present on the destination cluster or created as a new document if it is not present. You are not limited to this upsert behavior. In fact, you can change the sink to perform deletes or inserts only. These write behaviors are defined by the [Write Model Strategy](https://docs.mongodb.com/kafka-connector/current/sink-connector/fundamentals/write-strategies/) setting in the sink configuration.

To determine the performance of the sink connector, we need a timestamp of when the document was written to MongoDB. Currently, the only write model strategy that writes a timestamp field on behalf of the user is [UpdateOneTimestampsStrategy](http://UpdateOneTimestampsStrategy) and [UpdateOneBusinessKeyTimestampStrategy](http://UpdateOneTimestampsStrategy). These two write models insert a new field named **_insertedTS**, which can be used to query the lag between Kafka and MongoDB.

In this example, we’ll use [MongoDB Atlas](https://www.mongodb.com/cloud/atlas). MongoDB Atlas is a public cloud MongoDB data platform providing out-of-the-box capabilities such as [MongoDB Charts](https://docs.mongodb.com/charts/saas/), a tool to create visual representations of your MongoDB data. If you wish to follow along, you can create a [free forever tier](https://www.mongodb.com/cloud/atlas/register).

### Generate Sample Data

We will generate sample data using the [datagen](https://www.confluent.io/blog/easy-ways-generate-test-data-kafka/) Kafka Connector provided by Confluent.  [Datagen](https://www.confluent.io/hub/confluentinc/kafka-connect-datagen) is a convenient way of creating test data in the Kafka ecosystem. There are a few quickstart schema specifications bundled with this connector. We will use a quickstart called **users**.

```
curl -X POST -H "Content-Type: application/json" --data '
  {"name": "datagen-users",
   "config": { "connector.class":    "io.confluent.kafka.connect.datagen.DatagenConnector",
   "kafka.topic": "topic333",
   "quickstart": "users",
   "key.converter": "org.apache.kafka.connect.storage.StringConverter",
   "value.converter": "org.apache.kafka.connect.json.JsonConverter",
   "value.converter.schemas.enable": "false",
   "max.interval": 50,
   "iterations": 5000,
   "tasks.max": "2"
}}' http://localhost:8083/connectors -w "\n"

```

### Configure Sink Connector

Now that the data is generated and written to the Kafka topic, “topic333,” let’s create our MongoDB sink connector to write this topic data into MongoDB Atlas. As stated earlier, we will add a field **_insertedTS** for use in calculating the lag between the message timestamp and this value. To perform the insert, let’s use the **UpdateOneTimestampsStrategy** write mode strategy.

```
curl -X POST -H "Content-Type: application/json" --data '
{"name": "kafkametadata3",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "topics": "topic333",
    "connection.uri": "MONGODB CONNECTION STRING GOES HERE",
    "writemodel.strategy": "com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneTimestampsStrategy",
    "database": "kafka",
    "collection": "datagen",
    "errors.log.include.messages": true,
    "errors.deadletterqueue.context.headers.enable": true,
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.KafkaMetaDataStrategy",
    "tasks.max": 2,
    "value.converter.schemas.enable":false,
    "transforms": "InsertField",
    "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertField.offset.field": "offsetColumn",
    "transforms": "InsertField",
    "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.InsertField.timestamp.field": "timestampColumn"
}}' http://localhost:8083/connectors -w "\n"

```

Note: The field **_insertedTS** is populated with the time value of the Kafka connect server.

### Viewing Results with MongoDB Charts

Take a look at the MongoDB Atlas collection “datagen” and familiarize yourself with the added fields.

![Figure 1: Datagen collection as seen in MongoDB Atlas Collections page](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_atlas_web_page_00e3cd3149.png)

In this blog, we will use [MongoDB Charts](https://www.mongodb.com/products/charts) to display a performance graph. To make it easy to build the chart, we will create a view.

```
use kafka
db.createView("SinkView","datagen",
[
	{
		"$sort" : {
			"_insertedTS" : 1,
			"timestampColumn" : 1
		}
	},
	{
		"$project" : {
			"_insertedTS" : 1,
			"timestampColumn" : 1,
			"_id" : 0
		}
	},
	{
		"$addFields" : {
			"diff" : {
				"$subtract" : [
					"$_insertedTS",
					{
						"$convert" : {
							"input" : "$timestampColumn",
							"to" : "date"
						}
					}
				]
			}
		}
	}
])

```

To create a chart, click on the Charts tab in MongoDB Atlas:

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

Click on Datasources and “Add Data Source.”  The dialog will show the view that was created.

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

Select the SinkView and click Finish.

Download the [MongoDB Sink performance Chart](https://gist.github.com/RWaltersMA/555b5f17791ecb58e6e683c54bafd381) from Gist. ​​

```
curl https://gist.githubusercontent.com/RWaltersMA/555b5f17791ecb58e6e683c54bafd381/raw/748301bcb7ae725af4051d40b2e17a8882ef2631/sink-chart-performance.charts -o sink-performance.charts

```

Choose **Import Dashbaord** from the Add Dashboard dropdown and select the downloaded file.

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

Load the sink-perfromance.chart file.

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

Select the kafka.SinkView as the data source at the destination then click Save.

Now the KafkaPerformance chart is ready to view. When you click on the chart, you will see something like the following: 

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

This chart shows statistics on the differences between the timestamp in the Kafka topic and Kafka connector. In the above example, the maximum time delta is approximately one second (997ms) from inserting 40,000 documents.

## Measuring Source Performance

To measure the source, we will take a different approach using KSQL to create a stream of the clusterTime timestamp from the MongoDB change stream and the time the row was written in the Kafka topic. From here, we can push this data into a MongoDB sink and display the results in a MongoDB Chart. 

### Configure Source Connector

The first step will be to create the MongoDB Source connector that will be used to push data onto the Kafka topic.

```
curl -X POST -H "Content-Type: application/json" --data '
{"name": "mongo-source-perf",
  "config": {
  "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "connection.uri": "mongodb+srv://MONGODB CONNECTION STRING HERE",
  "database": "kafka",
  "collection": "source-perf-test",
  "mongo.errors.log.enable": "true",
  "topic.prefix":"mdb",
  "output.json.formatter" : "com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson",
    "output.format.value":"schema",
    "output.schema.infer.value":true,
    "output.format.key":"json",
    "publish.full.document.only": "false",
  "change.stream.full.document": "updateLookup"
}}' http://localhost:8083/connectors -w "\n"
```

### Generate Sample Data

There are many ways to generate sample data on MongoDB. In this blog post, we will use the [doc-gen](https://hub.docker.com/repository/docker/robwma/doc-gen/general) tool (Github repo) to quickly create sample documents based upon the user’s schema, which is defined as follows:

```
{
   "_id" : ObjectId("59b99db4cfa9a34dcd7885b6"),
   "name" : "Ned Stark",
   "email" : "sean_bean@gameofthron.es",
   "password" : "$2b$12$UREFwsRUoyF0CRqGNK0LzO0HM/jLhgUCNNIJ9RJAqMUQ74crlJ1Vu"
}

```

To generate data in your MongoDB cluster, issue the following:

```
docker run robwma/doc-gen:1.0 python doc-gen.py -s '{"name":"string","email":"string","password":"string"}' -c "MONGODB CONNECTION STRING GOES HERE" -t 1000 -db "kafka" -col "source-perf-test"
```

### Create KSQL Queries

Launch KSQL and create a stream of the clusterTime within the message. 

Note: If you do not have KSQL, you can run it as part of the Confluent Platform all in Docker using the following instructions.

If using Control Center, click ksQLDB, click Editor, and then paste in the following KSQL:

```
CREATE STREAM stats (
   clusterTime BIGINT
 ) WITH (
   KAFKA_TOPIC='kafka.source-perf-test',
       VALUE_FORMAT='AVRO'
 );

```

The only information that we need from the message is the clusterTime. This value is provided within the [change stream](https://docs.mongodb.com/kafka-connector/current/kafka-source/#change-stream-event-document-format) event. For reference, this is a sample event from change streams.

```
{
  _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>, ... ]
  },
  "clusterTime": <Timestamp>,
  "txnNumber": <NumberLong>,
  "lsid": {
    "id": <UUID>,
    "uid": <BinData>
  }
}

```

**Step 3**

Next, we will create a ksql stream that calculates the difference between the cluster time (time when it was created on MongoDB) and the time where it was inserted on the broker. 

```
CREATE STREAM STATS2 AS
  select ROWTIME - CLUSTERTIME as diff, 1 AS ROW from STATS EMIT CHANGES;

```

As stated previously, this diff value may not be completely accurate if the clocks on Kafka and MongoDB are different. 

**Step 4**

To see how the values change over time, we can use a window function and write the results to a table which can then be written into MongoDB via a sink connector.

```
SET 'ksql.suppress.enabled' = 'true';

CREATE TABLE STATSWINDOW2 AS
  SELECT  AVG( DIFF ) AS AVG, MAX(DIFF) AS MAX,  count(*) AS COUNT,  ROW  FROM STATS2
 WINDOW TUMBLING (SIZE 10 SECONDS)
 GROUP BY ROW
 EMIT FINAL;

```

Windowing lets you control how to group records that have the same key for stateful operations, such as aggregations or joins into so-called windows. There are three ways to define time windows in ksqlDB: hopping windows, tumbling windows, and session windows. In this example, we will use tumbling as it is a fixed-duration, non-overlapping, and gap-less window.

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

### Configure Sink Connector

The final step is to create a sink connector to insert all this aggregate data on MongoDB.

```
curl -X POST -H "Content-Type: application/json" --data '
{
 "name": "MongoSource-SinkPerf",
 "config": {
   "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
   "tasks.max": "1",
   "errors.log.enable": true,
   "errors.log.include.messages": true,
   "topics": "STATSWINDOW2",
   "errors.deadletterqueue.context.headers.enable": true,
   "connection.uri": "MONGODB CONNECTION STRING GOES HERE",
   "database": "kafka",
   "collection": "sourceStats",
   "mongo.errors.log.enable": true,
   "transforms": "InsertField",
   "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
   "transforms.InsertField.timestamp.field": "timestampColumn"
}}' http://localhost:8083/connectors -w "\n"

```

### Viewing Results with MongoDB Charts

Download the [MongoDB Source performance Chart](https://gist.githubusercontent.com/RWaltersMA/011f1473cf937badc61b752a6ab769d4/raw/bc180b9c2db533536e6c65f34c30b2d2145872f9/mongodb-source-performance.chart) from Gist. ​​

```
curl https://gist.githubusercontent.com/RWaltersMA/011f1473cf937badc61b752a6ab769d4/raw/bc180b9c2db533536e6c65f34c30b2d2145872f9/mongodb-source-performance.chart -o source-performance.charts

```

Choose **Import Dashboard** from the Add Dashboard dropdown and select the downloaded file.

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

You will need to create a Datasource to the new sink collection, “kafka.sourceStats.”

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

Click on the Kafka Performance Source chart to view the statistics.

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

In the above example, you can see the 10-second sliding window performance statistics for 1.5M documents. The average difference was 252s, with the maximum difference being 480s.  Note that some of this delta could be differences in clocks between MongoDB and Kafka. While not taking these numbers as absolute, simply using this technique is good enough to determine trends and if the performance is getting worse or better.

If you have any opinions on features or functionality enhancements that you would like to see with respect to monitoring performance or monitoring the MongoDB Connector for Apache Kafka in general, please add a comment to [KAFKA-64](https://jira.mongodb.org/browse/KAFKA-64). 

Have any questions? Check out our [Connectors and Integrations](https://www.mongodb.com/community/forums/c/data/connectors-integrations/48) MongoDB community forum.

Learn about measuring the performance of the MongoDB Connector for Apache Kafka in both a source and sink configuration.

Juan Soto

Content by Juan Soto

Measuring MongoDB Kafka Connector Performance