MongoDB and Kafka: a MongoDB & NoSQL guide for developers

Introduction

Real-time data synchronization across services is no longer a nice-to-have—it's table stakes for modern systems. When you're running MongoDB alongside Kafka, you unlock the ability to stream changes from your database to any downstream consumer, or push event data back into MongoDB with exactly-once semantics. MongoDB and Kafka: a MongoDB & NoSQL guide for developers walks you through the patterns that let teams at scale keep their data consistent, maintain audit trails, and decouple their backend services.

Over the next sections, we'll explore how to configure Kafka connectors to capture MongoDB change events, sink those events reliably, handle idempotent writes to avoid duplicates, and manage schema evolution as your events grow more complex. By the end, you'll have a playbook for streaming architectures that actually work in production.

Kafka Connector — The MongoDB connector streams change events to Kafka

The MongoDB Kafka connector is built on Change Streams, MongoDB's native capability to watch for document modifications in real-time. Instead of polling collections or scanning oplog entries yourself, the connector does the heavy lifting: it listens for inserts, updates, replaces, and deletes, then emits them as Kafka messages.

To get started, you'll need to deploy the connector in your Kafka Connect cluster. The connector runs as a task and maintains its own offset tracking, so if it restarts, it picks up from where it left off rather than replaying the entire history.

Configuration basics

Here's a minimal connector configuration that watches a MongoDB collection and publishes changes to a Kafka topic:

{
  "name": "mongodb-source-users",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "mongodb.connection.uri": "mongodb+srv://user:pass@cluster.mongodb.net",
    "database": "production",
    "collection": "users",
    "topic.prefix": "mongodb",
    "pipeline": "[{\"$match\": {\"operationType\": {\"$in\": [\"insert\", \"update\", \"replace\", \"delete\"]}}}]"
  }
}

Key parameters:

• mongodb.connection.uri — Connection string to your MongoDB instance or cluster.
• database and collection — The source collection to watch.
• topic.prefix — Kafka topic will be prefixed with this value; e.g., mongodb.production.users.
• pipeline — An aggregation pipeline to filter which operations you care about. You can exclude deletes, ignore certain field changes, or route to different topics.

When a user document is inserted or updated, the connector emits a message with the full change event, including the operation type, the document's new state, and metadata about the change.

Real-world scenario: Audit trail

Imagine you're building an e-commerce platform. Every order status change in MongoDB should be captured in an immutable audit log. You configure the connector to watch the orders collection and publish to kafka-audit-events. A payment service consumes these events, processes refunds asynchronously, and logs financial decisions. By streaming from MongoDB to Kafka, you've decoupled the order service from payment reconciliation without adding complexity to your application code.

Sink connector — Write Kafka events into MongoDB collections

The inverse pattern is equally powerful: take events from Kafka and write them into MongoDB. The MongoDB Kafka Sink Connector handles this, applying transformations and ensuring data lands in the right collection.

A typical use case is aggregating enriched data from a stream. For example, your payment service publishes transaction events to Kafka; you want to materialize a real-time summary in MongoDB for dashboarding. Or an external system publishes user profile changes via Kafka, and you keep a denormalized copy in MongoDB for fast lookups.

Sink configuration example

Here's how to set up a sink that consumes events and writes them into a MongoDB collection:

{
  "name": "kafka-sink-transactions",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "topics": "payment.transactions",
    "mongodb.connection.uri": "mongodb+srv://user:pass@cluster.mongodb.net",
    "mongodb.database": "analytics",
    "mongodb.collection": "transaction_summaries",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": false,
    "writemodel.strategy": "com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy",
    "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.IdStrategy",
    "mongodb.bulk.write.batch.size": 1000
  }
}

Key parameters:

• topics — Kafka topic(s) to consume from.
• mongodb.database and mongodb.collection — Target collection in MongoDB.
• writemodel.strategy — Determines how records are written. ReplaceOneBusinessKeyStrategy treats the Kafka key as a unique identifier and replaces the entire document, which is useful for state snapshots.
• mongodb.bulk.write.batch.size — How many writes to batch before sending to MongoDB. Larger batches reduce round-trips but increase memory usage.

Handling late arrivals and out-of-order events

In a distributed system, events don't always arrive in order. You might receive an update for a transaction before its creation event. The sink connector alone doesn't solve this—you need logic in your connector configuration or post-processing. One common approach:

• Include a timestamp field in every event.
• Use a custom single message transform (SMT) to check MongoDB for an existing document and only replace if the incoming event's timestamp is newer.
• Store a _last_update_at field in MongoDB so you can safely ignore stale events.

Idempotent writes — Upsert by event key to handle redelivery

Kafka provides at-least-once delivery semantics, meaning a message can be delivered multiple times, especially after broker failures or connector restarts. If you're not careful, idempotent writes, you'll end up with duplicate data in MongoDB or incorrectly applied transactions.

The solution is to design your events and writes so that replaying the same event multiple times produces the same result. This is called idempotency.

Pattern: Upsert with a business key

Rather than always inserting, use updateOne with upsert: true and match on a business key—typically a unique identifier like an order ID or user ID, combined with an event timestamp.

// Consumer logic (pseudocode)
async function handlePaymentEvent(event) {
  const { orderId, amount, status, eventTime } = event;

  const filter = {
    orderId: orderId,
    eventTime: eventTime
  };

  const update = {
    $set: {
      orderId: orderId,
      amount: amount,
      status: status,
      eventTime: eventTime,
      processedAt: new Date()
    }
  };

  const result = await db.collection('payments').updateOne(filter, update, { upsert: true });
  
  // If upsertedId is set, this was the first write; otherwise, it was a duplicate
  console.log(`Upserted: ${result.upsertedId}, Matched: ${result.matchedCount}`);
}

By including both orderId and eventTime in the filter, you ensure that the same event, if delivered twice, updates the same document. If Kafka redelivers an older event after you've processed a newer one, the older document upsert won't overwrite it because it matches a different (orderId, eventTime) pair.

Version fields for causal ordering

For events that represent state transitions (e.g., order status changes), add a version field:

const filter = {
  orderId: orderId,
  version: currentVersion
};

const update = {
  $set: {
    status: newStatus,
    version: currentVersion + 1,
    updatedAt: new Date()
  }
};

const result = await db.collection('orders').updateOne(filter, update, { upsert: false });

if (result.matchedCount === 0) {
  // Version mismatch; event was out of order or already processed
  console.warn(`Order ${orderId} version conflict`);
}

If a Kafka event is replayed or arrives out of order, the version check ensures you only apply it if the document's current version matches what the event expects. This prevents accidentally rolling back to an old state.

Schema evolution — Use a schema registry for event formats

As your systems mature, event schemas change. A payment event might start with just amount and status, but later you add currency, card type, and fraud score. Without schema management, consumers break when they encounter new fields, and tracking which version of a schema an old message used becomes a nightmare.

A schema registry (like Confluent Schema Registry or AWS Glue Schema Registry) stores versioned schemas and lets producers and consumers negotiate compatibility. Events are serialized in a compact format (Avro, Protobuf, or JSON Schema) with a schema ID, so consumers always know how to deserialize them.

Avro schema example

Here's an Avro schema for a payment event, versioned to allow adding new fields without breaking old consumers:

{
  "type": "record",
  "name": "PaymentEvent",
  "namespace": "com.example.payments",
  "fields": [
    {
      "name": "orderId",
      "type": "string"
    },
    {
      "name": "amount",
      "type": "double"
    },
    {
      "name": "currency",
      "type": {
        "type": "string",
        "default": "USD"
      }
    },
    {
      "name": "status",
      "type": {
        "type": "enum",
        "name": "PaymentStatus",
        "symbols": ["PENDING", "COMPLETED", "FAILED", "REFUNDED"]
      }
    },
    {
      "name": "timestamp",
      "type": "long"
    },
    {
      "name": "metadata",
      "type": {
        "type": "map",
        "values": "string"
      },
      "default": {}
    }
  ]
}

Key design decisions:

• Default values — Fields like currency and metadata have defaults, so old messages without them can still be read.
• Enums for status — Constrains values and makes validation upfront.
• Maps for extensibility — The metadata field lets producers add arbitrary key-value pairs without schema changes.

Integration with MongoDB connector

When your MongoDB source connector produces events to Kafka, configure it to use Avro serialization with schema registry:

{
  "name": "mongodb-source-with-schema",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "mongodb.connection.uri": "mongodb+srv://user:pass@cluster.mongodb.net",
    "database": "production",
    "collection": "orders",
    "topic.prefix": "mongodb",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schema.registry.subject.name.strategy": "io.confluent.kafka.connect.storage.subjects.TopicNameStrategy"
  }
}

Now every message produced includes a schema ID, and consumers can retrieve the schema from the registry. When you release version 2 of the schema with new fields, old consumers continue working, and new consumers can access the richer data.

Common mistakes

• Ignoring duplicate handling — Assuming Kafka delivers exactly once. Always make your writes idempotent or accept duplicates gracefully.
• Not monitoring connector lag — Kafka Connect connectors can silently fall behind. Set up metrics to track how far behind your source connector is relative to the MongoDB oplog.
• Over-filtering in the pipeline — Applying complex aggregation logic in the Change Stream pipeline can be expensive. Consider filtering in the Kafka consumer instead.
• Mixing schema versions without a registry — Hardcoding JSON structures and adding fields ad-hoc leads to deserialization errors. Use a schema registry from day one.
• Batch sizes too large — Setting bulk.write.batch.size too high can cause memory pressure and slow restarts. Start with 500–1000 and tune from there.
• No dead-letter topic for errors — If a Kafka message fails to parse or write to MongoDB, the connector stops. Route failures to a DLQ for manual inspection.

Key takeaways

• Use the MongoDB Kafka connector to stream Change Stream events from collections into Kafka topics, decoupling your systems and enabling audit trails.
• The sink connector inverts the flow, pulling events from Kafka and materializing them in MongoDB with configurable write strategies.
• Make writes idempotent by upsert-matching on business keys and event timestamps; include version fields to prevent out-of-order replays from corrupting state.
• Adopt a schema registry (Avro, Protobuf, or JSON Schema) early to manage event evolution and let producers and consumers work with different schema versions safely.
• Monitor connector lag, set up dead-letter queues, and test failover scenarios to ensure your Kafka–MongoDB pipeline is production-ready.