The MongoDB Kafka connector runs on Kafka Connect in both directions. The source connector, com.mongodb.kafka.connect.MongoSourceConnector, opens a MongoDB change stream and publishes each change as a Kafka record. The sink connector, com.mongodb.kafka.connect.MongoSinkConnector, reads Kafka topics and writes the records into MongoDB collections.
Connectors are registered against the Connect REST API. A POST to the workers’ /connectors endpoint with the JSON configuration creates each one. How the framework itself distributes and supervises this work is covered on what is Kafka Connect, the broader CDC landscape in our complete guide to Kafka change data capture, and where Connect sits in the platform in the complete Kafka guide.
The MongoDB source collection, the resulting Kafka topic records, and the dead-letter topic.
The production configuration, both directions
A production pipeline needs two explicit JSON documents, one per direction. The templates in the next two sections carry the reliability settings inline rather than leaving them at defaults, because the defaults are development defaults: the source starts from the latest change only, and the sink fails its task on the first bad record.
Credentials do not belong in the JSON. The connection.uri should be externalised through a Connect ConfigProvider rather than pasted in plain text, per the connector’s own connection guidance.
A template like this hides how narrow the failure surface really is. Derek, our co-founder, once debugged a customer environment by clicking through every connector in the cluster, and only the MongoDB source connector was failing. Thomas, one of our engineers, traced a related class of problem to the connector’s /validate endpoint: it expects configuration data to be passed in, and MongoDB can default zero input to a localhost connection, which looks like a working config right up until it is deployed somewhere localhost means nothing. The lesson we took from it: validate the rendered configuration against the target environment, not just against the connector’s own validator.
MongoDB source connector, CDC via change streams
The source connector tails a change stream, which requires a replica set or sharded cluster. Change stream behaviour on updates is controlled by change.stream.full.document, and publish.full.document.only=true publishes just the current document rather than the change event envelope.
{
"name": "mongo-source-orders",
"config": {
"connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
"connection.uri": "${file:/opt/kafka/secrets/mongo.properties:connection.uri}",
"database": "shop",
"collection": "orders",
"topic.prefix": "cdc",
"startup.mode": "copy_existing",
"change.stream.full.document": "updateLookup",
"pipeline": "[{\"$match\": {\"operationType\": {\"$in\": [\"insert\", \"update\", \"replace\", \"delete\"]}}}]"
}
}Key choices, in plain words first:
Backfill then stream. startup.mode=copy_existing copies the existing collection before streaming new changes. The copy is at-least-once and eventually consistent, so downstream consumers must tolerate duplicates during the backfill window.
Updates carry the full document. change.stream.full.document=updateLookup attaches the post-update document to update events, which is what most downstream consumers want.
Filter in the pipeline. The pipeline setting is a MongoDB aggregation pipeline applied to the change stream, so unwanted operation types or fields are dropped at the source rather than in every consumer.
Topic naming is prefix plus namespace. With topic.prefix=cdc, changes to shop.orders land on the topic: cdc.shop.orders.
For scale context on the source side, Shopify replaced its in-house Longboat pipeline with Kafka Connect and Debezium reading MySQL binary logs directly, capturing every insert, update and delete with a P99 latency under 10 seconds from database write to Kafka. Change-stream CDC on MongoDB buys the same property this template relies on: the connector reads the database’s own ordered change log rather than polling tables, so latency stays bounded and no write is skipped between polls.
MongoDB sink connector
The sink connector consumes topics and applies write models to a collection. The write model strategy decides whether records insert, replace or update, and the document id strategy decides which field identifies the target document.
{
"name": "mongo-sink-orders",
"config": {
"connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
"connection.uri": "${file:/opt/kafka/secrets/mongo.properties:connection.uri}",
"topics": "cdc.shop.orders",
"database": "reporting",
"collection": "orders",
"document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy",
"value.projection.type": "AllowList",
"value.projection.list": "orderId",
"writemodel.strategy": "com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy",
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "dlq.mongo-sink-orders",
"errors.deadletterqueue.context.headers.enable": "true"
}
}The projection keys are documented in the sink post-processor configuration reference.
Sink-side integration is where I see teams lose the most time, and it is rarely the connector’s core path that fails. Kylie, our co-founder, has fielded the canonical version: a team integrating a Connect sink with a cloud key-value store had addressed several visible issues and still had nothing arriving in the target, with the logs offering no obvious cause. Sinks fail quietly when serialization, permissions or networking are wrong, because the task can sit RUNNING while every write is rejected downstream. The lesson: instrument the target system’s write metrics from day one, and treat “RUNNING with zero throughput” as an alert, not a resting state.
The production-grade requirements the template covers
Idempotent writes. A business-key write strategy such as ReplaceOneBusinessKeyStrategy keyed on a stable id means a retried record replaces the same document instead of inserting a duplicate. This is what makes at-least-once delivery safe on the sink side.
Dead letter queue. errors.tolerance=all plus errors.deadletterqueue.topic.name sends malformed records to a DLQ topic instead of failing the whole task. Kafka Connect has had native DLQ support since version 2.0, with error context available as record headers. The DLQ topic needs its own consumer and an alert, or it becomes a silent data drain, a failure mode our guide to dead letter queues in Kafka covers in depth.
Converters. Connect workers commonly default to JSON converters. Where the pipeline uses Avro or Protobuf with a schema registry, set key.converter and value.converter on the connector explicitly so both directions agree on the wire format.
Large documents. Producer settings for a single connector are overridden with the producer.override. prefix, which is how a CDC pipeline carrying oversized documents raises its max request size without touching the worker-wide defaults.
My quick-win checklist for any production Connect pipeline, this one included: offset reset defaults, poll interval limits, DLQ loops, retention sizing, message size limits, linger.ms, client library choice, and transactional settings. Most of the incidents I get called into trace back to one of those eight, set once at deployment and never revisited. Run the list against this template before the first production deploy and again after any connector version upgrade.
FAQ
How can I connect Kafka to MongoDB?
Run the MongoDB Kafka connector on Kafka Connect. The source connector, com.mongodb.kafka.connect.MongoSourceConnector, streams MongoDB changes into topics, and the sink connector, com.mongodb.kafka.connect.MongoSinkConnector, writes topic records into collections. Register each with a POST to the Connect REST API.
Does the MongoDB source connector need a replica set?
Yes. The source connector tails a MongoDB change stream, and change streams require a replica set or a sharded cluster. A standalone mongod cannot feed the connector.
What happens to bad records in the sink?
With errors.tolerance=all and a named DLQ topic, malformed records go to the dead letter queue with error context in the record headers instead of failing the task. The DLQ needs its own consumer and an alert, or it becomes a silent data drain.