Skip to main content

​
Overview

The MongoDB Sink Connector consumes records from Kafka topics, converts message keys and values to BSON, and writes them to MongoDB databases and collections. By default, the Connector uses the configured database and uses the Kafka topic name as the collection name. You can also specify a fixed collection, route records dynamically based on record fields, or override write configurations for individual topics. The Connector is suitable for continuously writing business events, state updates, and time-series data from Kafka to MongoDB. The standard write path supports configurable document _id values, replacement or update strategies, field processing, bulk writes, and tombstone deletion. For supported CDC messages, dedicated processors can convert change events into MongoDB write operations.

​
Prerequisites

  • The MongoDB account must have the permissions required to create or write to the target database and collection. When using TLS, client certificates, the Stable API, client-side field level encryption, or custom authentication, prepare the corresponding certificates, keys, key vaults, and server capabilities in advance.
  • When using time-series collections or the Stable API, MongoDB must be version 5.0 or later. For an existing time-series collection, its time field, metadata field, and granularity must match the Connector configuration.

​
License

Uses the Apache License 2.0.

​
Quick Start

Prepare a Connect cluster, Kafka, and MongoDB, and confirm network connectivity and access permissions. For preparation and management operations, see Manage Connectors.
Replace <mongodb-host> with a MongoDB address accessible to the Connect Worker. This configuration writes JSON objects from mongodb-sink-input to inventory.events. value.converter.schemas.enable is a JSON Converter configuration placed in the Connector config.

​
Configuration

​
Connector Identity and Tasks

​
connector.class

Specifies the MongoDB Sink Connector implementation class.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / Notes: Use com.mongodb.kafka.connect.MongoSinkConnector.
  • Required: Yes

​
tasks.max

The maximum number of tasks that the Connector can request.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / Notes: Must be at least 1. Actual parallelism is also limited by the number of topic partitions and Worker assignment.

​
Kafka Topic Selection

​
topics

Specifies the list of Kafka topics to consume.
  • Type: list
  • Default: Empty list
  • Importance: High
  • Valid values / Notes: Exactly one of topics and topics.regex must have a non-empty value.

​
topics.regex

Uses a Java regular expression to select the Kafka topics to consume.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: Must be a valid Java regular expression. Exactly one of topics and topics.regex must have a non-empty value.

​
MongoDB Connection and Authentication

​
connection.uri

The MongoDB Driver connection string, which can include hosts, credentials, and URI options.
  • Type: password
  • Default: mongodb://localhost:27017
  • Importance: High
  • Valid values / Notes: Must be a valid MongoDB Connection String. Do not expose credentials in documentation or logs.

​
mongo.custom.auth.mechanism.enable

Enables a custom MongoDB credential provider.
  • Type: boolean-like string
  • Default: No ConfigDef default
  • Importance: Not declared
  • Valid values / Notes: This public extension is not registered in ConfigDef. Only a value that Boolean.parseBoolean parses as true enables the provider. When enabled, mongo.custom.auth.mechanism.providerClass is required.

​
mongo.custom.auth.mechanism.providerClass

Specifies the custom CustomCredentialProvider implementation class.
  • Type: class-name string
  • Default: No ConfigDef default
  • Importance: Not declared
  • Valid values / Notes: Required when custom authentication is enabled. The class must be on the plugin classpath, implement CustomCredentialProvider, and provide an accessible no-argument constructor.

​
TLS

​
connection.ssl.truststore

The truststore file used for MongoDB TLS connections.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: When non-empty, the file must be readable and loadable using connection.ssl.truststorePassword.

​
connection.ssl.truststorePassword

The truststore password.
  • Type: password
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Used only when connection.ssl.truststore is non-empty.

​
connection.ssl.keystore

The keystore file used for client TLS credentials.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: When non-empty, the file must be readable and loadable using connection.ssl.keystorePassword.

​
connection.ssl.keystorePassword

The password for the keystore and its private key.
  • Type: password
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Used only when connection.ssl.keystore is non-empty.

​
MongoDB Stable API

​
server.api.version

Specifies the MongoDB Stable API version.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: An empty value disables the Stable API. A non-empty value must be a ServerApiVersion supported by the Driver and requires MongoDB 5.0 or later.

​
server.api.deprecation.errors

Treats use of deprecated APIs as an error when the Stable API is enabled.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: Effective only when server.api.version is non-empty.

​
server.api.strict

Enables strict MongoDB Stable API version checking.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: Effective only when server.api.version is non-empty.

​
Client-Side Field Level Encryption

​
csfle.enabled

Enables MongoDB client-side field level encryption for Sink writes.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: When set to true, csfle.key.vault.namespace and csfle.local.master.key are required.

​
csfle.key.vault.namespace

Specifies the key vault namespace in database.collection format.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Must be non-empty when csfle.enabled=true.

​
csfle.local.master.key

Specifies a Base64-encoded, 96-byte local KMS master key.
  • Type: password
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Required when csfle.enabled=true. The value is Base64-decoded at runtime.

​
csfle.schema.map

Specifies the JSON Schema Map that controls automatic field encryption.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: When non-empty, it must parse as a BSON document that maps namespaces to schemas.

​
csfle.bypass.query.analysis

Bypasses automatic CS-FLE query analysis.
  • Type: boolean
  • Default: true
  • Importance: Low
  • Valid values / Notes: Relevant only when CS-FLE is enabled. Setting it to false requires automatic encryption support such as mongocryptd or crypt_shared.

​
Target Namespace

​
database

The default target MongoDB database.
  • Type: string
  • Default: None
  • Importance: High
  • Valid values / Notes: Must be a non-empty string. A field path mapper can replace it per record.
  • Required: Yes

​
collection

The default target MongoDB collection.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid values / Notes: With the default namespace mapper, an empty value uses the Kafka topic name.

​
namespace.mapper

Specifies the NamespaceMapper implementation class that selects the target database and collection.
  • Type: string
  • Default: com.mongodb.kafka.connect.sink.namespace.mapping.DefaultNamespaceMapper
  • Importance: High
  • Valid values / Notes: Must be the fully qualified name of a class that implements NamespaceMapper and has a public no-argument constructor. Field path configurations require com.mongodb.kafka.connect.sink.namespace.mapping.FieldPathNamespaceMapper.

​
namespace.mapper.key.database.field

Reads the target database name from a field path in the key document.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Requires FieldPathNamespaceMapper and cannot be configured together with namespace.mapper.value.database.field.

​
namespace.mapper.key.collection.field

Reads the target collection name from a field path in the key document.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Requires FieldPathNamespaceMapper and cannot be configured together with namespace.mapper.value.collection.field.

​
namespace.mapper.value.database.field

Reads the target database name from a field path in the value document.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Requires FieldPathNamespaceMapper and cannot be configured together with namespace.mapper.key.database.field.

​
namespace.mapper.value.collection.field

Reads the target collection name from a field path in the value document.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Requires FieldPathNamespaceMapper and cannot be configured together with namespace.mapper.key.collection.field.

​
namespace.mapper.error.if.invalid

Controls whether a record fails when a namespace field is missing or is not a string.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: Used with FieldPathNamespaceMapper. false falls back to the default database or collection, while true throws a DataException.

​
topic.override.<topic>.<property>

Overrides a topic-level Sink configuration for one Kafka topic.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: Use a concrete key formed from the actual topic and a registered property; do not use the template key directly. Override values are merged over the global configuration and cannot override connection.uri or topics. When using topics.regex, the topic in an override must match the regular expression.

​
Record Conversion

​
key.converter

Overrides the Connector-level message key Converter.
  • Type: class
  • Default: No fixed default
  • Importance: Low
  • Valid values / Notes: Inherits the Worker setting when not configured. It must be an instantiable Converter, and its subconfigurations use the key.converter. prefix.

​
value.converter

Overrides the Connector-level message value Converter.
  • Type: class
  • Default: No fixed default
  • Importance: Low
  • Valid values / Notes: Inherits the Worker setting when not configured. It must be an instantiable Converter, and its subconfigurations use the value.converter. prefix.

​
Document Identity

​
document.id.strategy

Specifies the IdStrategy implementation class used by DocumentIdAdder to generate the MongoDB _id.
  • Type: string
  • Default: com.mongodb.kafka.connect.sink.processor.id.strategy.BsonOidStrategy
  • Importance: High
  • Valid values / Notes: Use the fully qualified name of a class that implements IdStrategy. Although an empty string passes ConfigDef validation, initialization fails. Ignored when a CDC processor is configured.

​
document.id.strategy.overwrite.existing

Allows DocumentIdAdder to overwrite an existing _id in the value document.
  • Type: boolean
  • Default: false
  • Importance: High
  • Valid values / Notes: Relevant only when DocumentIdAdder runs. Ignored when a CDC processor is configured.

​
document.id.strategy.uuid.format

Specifies the BSON representation of UUIDs generated by UuidStrategy.
  • Type: string
  • Default: string
  • Importance: High
  • Valid values / Notes: Case-insensitive. Valid values are string and binary. Used only with UuidStrategy.

​
document.id.strategy.partial.key.projection.type

Specifies the dedicated field projection mode for PartialKeyStrategy.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: An empty value falls back to key.projection.type. Valid values are none, allowlist, and blocklist. The deprecated whitelist and blacklist values remain compatible but should be replaced by allowlist and blocklist, respectively.

​
document.id.strategy.partial.key.projection.list

Specifies the field list used by PartialKeyStrategy.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: An empty value falls back to key.projection.list. Used only with PartialKeyStrategy.

​
document.id.strategy.partial.value.projection.type

Specifies the dedicated field projection mode for PartialValueStrategy.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: An empty value falls back to value.projection.type. Valid values are none, allowlist, and blocklist. The deprecated whitelist and blacklist values remain compatible but should be replaced by allowlist and blocklist, respectively.

​
document.id.strategy.partial.value.projection.list

Specifies the field list used by PartialValueStrategy.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: An empty value falls back to value.projection.list. Used only with PartialValueStrategy.

​
Writes and Deletion

​
writemodel.strategy

Specifies the implementation class that builds MongoDB WriteModels for non-null records.
  • Type: string
  • Default: com.mongodb.kafka.connect.sink.writemodel.strategy.DefaultWriteModelStrategy
  • Importance: Low
  • Valid values / Notes: Must implement WriteModelStrategy. By default, replacement writes are used, while time-series collections use inserts. Ignored when a CDC processor is configured.

​
delete.on.null.values

Converts tombstone records with a null value into MongoDB deletions based on the message key.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: With the default deletion strategy, document.id.strategy must be FullKeyStrategy, PartialKeyStrategy, or ProvidedInKeyStrategy. Ignored when a CDC processor is configured.

​
delete.writemodel.strategy

Specifies the implementation class that builds MongoDB deletion models for tombstone records.
  • Type: string
  • Default: com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneDefaultStrategy
  • Importance: Low
  • Valid values / Notes: Used only when delete.on.null.values=true. A custom class must implement WriteModelStrategy.

​
max.batch.size

Limits the maximum number of records in each MongoDB bulk write.
  • Type: int
  • Default: 0
  • Importance: Medium
  • Valid values / Notes: Must be at least 0. 0 means that record groups for the same topic and namespace are not split by this configuration. This value limits the number of records, not the number of BSON bytes.

​
bulk.write.ordered

Controls whether MongoDB bulk writes execute in ordered mode.
  • Type: boolean
  • Default: true
  • Importance: Medium
  • Valid values / Notes: Ordered mode stops subsequent writes in the current batch after the first write error. Unordered mode can continue other writes in the current batch but does not guarantee record execution order.

​
rate.limiting.timeout

The number of milliseconds the Connector sleeps when rate limiting is triggered.
  • Type: int
  • Default: 0
  • Importance: Low
  • Valid values / Notes: Must be at least 0. Rate limiting is enabled only when both this configuration and rate.limiting.every.n are greater than 0.

​
rate.limiting.every.n

Specifies how many batches are processed before a rate-limiting sleep is triggered.
  • Type: int
  • Default: 0
  • Importance: Low
  • Valid values / Notes: Must be at least 0. Rate limiting is enabled only when both this configuration and rate.limiting.timeout are greater than 0.

​
Post-Processing and Field Transformations

​
post.processor.chain

Specifies the PostProcessor classes executed in sequence on the standard write path.
  • Type: list
  • Default: com.mongodb.kafka.connect.sink.processor.DocumentIdAdder
  • Importance: Low
  • Valid values / Notes: Each item must be a fully qualified implementation class name. If DocumentIdAdder is omitted, the Connector still automatically places it at the beginning of the chain. Ignored when a CDC processor is configured.

​
key.projection.type

Specifies the key field projection mode.
  • Type: string
  • Default: none
  • Importance: Low
  • Valid values / Notes: Valid values are none, allowlist, and blocklist. whitelist and blacklist are deprecated and should be replaced by allowlist and blocklist, respectively. Ignored when a CDC processor is configured.

​
key.projection.list

Specifies the comma-separated field paths used for key field projection.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: Interpreted according to key.projection.type. Ignored when the mode is none or when a CDC processor is configured.

​
value.projection.type

Specifies the value field projection mode.
  • Type: string
  • Default: none
  • Importance: Low
  • Valid values / Notes: Valid values are none, allowlist, and blocklist. whitelist and blacklist are deprecated and should be replaced by allowlist and blocklist, respectively. Ignored when a CDC processor is configured.

​
value.projection.list

Specifies the comma-separated field paths used for value field projection.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: Interpreted according to value.projection.type. Ignored when the mode is none or when a CDC processor is configured.

​
field.renamer.mapping

Specifies a JSON array of exact field path rename rules.
  • Type: string
  • Default: []
  • Importance: Low
  • Valid values / Notes: Each enabled object must contain oldName and newName, and the corresponding rename processor must be configured in post.processor.chain.

​
field.renamer.regexp

Specifies a JSON array of regular expression field rename rules.
  • Type: string
  • Default: []
  • Importance: Low
  • Valid values / Notes: Objects must be valid regular-expression rename settings, and the corresponding processor must be configured in post.processor.chain. Do not use the undefined field.renamer.regex.

​
field.value.transformer

Specifies a custom FieldValueTransformer to apply to selected value fields.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: When non-empty, it must be the fully qualified name of an implementation class with a public no-argument constructor. The Connector automatically appends the corresponding post-processor on the standard write path.

​
field.value.transformer.fields

Specifies the comma-separated field names processed by field.value.transformer.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: Meaningful only when a Transformer is configured. Matching is by field name rather than full path and also matches same-named fields in nested documents and document arrays.

​
CDC Processing

​
change.data.capture.handler

Specifies the CdcHandler implementation class that interprets CDC envelopes and builds MongoDB WriteModels.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: A non-empty value switches to the CDC path and ignores the post-processing chain, field renaming, key and value projections, the standard write model, document ID configurations, and delete.on.null.values.

​
change.data.capture.handler.remove.null.values

Removes null fields from CDC replacement documents and the $set content of updates.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / Notes: Effective only when a CDC Handler is configured. It preserves $unset and does not affect deletion operations.

​
Time-Series Collections

​
timeseries.timefield

Specifies the top-level BSON date-time field for a MongoDB time-series collection.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: A non-empty value enables time-series mode and requires MongoDB 5.0 or later. Every record must contain this field, and its value must be a BSON date-time or be automatically convertible.

​
timeseries.metafield

Specifies the optional top-level metadata field for a time-series collection.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: A non-empty value requires timeseries.timefield. It cannot be the same as the time field or _id, and MongoDB does not allow array metadata.

​
timeseries.expire.after.seconds

Specifies the data retention period in seconds for a newly created time-series collection.
  • Type: long
  • Default: 0
  • Importance: Low
  • Valid values / Notes: Must be at least 0. A non-zero value requires timeseries.timefield. 0 means expireAfterSeconds is not set when the collection is created.

​
timeseries.granularity

Specifies the expected time interval granularity for a newly created time-series collection.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: Valid values are seconds, minutes, and hours. A non-empty value requires timeseries.timefield.

​
timeseries.timefield.auto.convert

Converts an epoch-millisecond or string time field to a BSON date-time.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / Notes: Adds the conversion processor only in time-series mode. If conversion fails, the original value is preserved and the subsequent write can still fail.

​
timeseries.timefield.auto.convert.date.format

Specifies the DateTimeFormatter pattern used to parse string time fields.
  • Type: string
  • Default: yyyy-MM-dd[['T'][ ]][HH:mm:ss[[.][SSSSSS][SSS]][ ]VV[ ]'['VV']'][HH:mm:ss[[.][SSSSSS][SSS]][ ]X][HH:mm:ss[[.][SSSSSS][SSS]]]
  • Importance: Low
  • Valid values / Notes: Must be accepted by DateTimeFormatter.ofPattern. Used only when automatically converting string time values.

​
timeseries.timefield.auto.convert.locale.language.tag

Specifies the Locale Language Tag used by the time-field formatter.
  • Type: string
  • Default: Empty string
  • Importance: Low
  • Valid values / Notes: An empty value uses Locale.ROOT. A non-empty value must be accepted by Locale.forLanguageTag.

​
Error Handling and Retries

​
errors.tolerance

Configures Kafka Connect framework fault tolerance and serves as the base value for the Connector’s internal fault tolerance.
  • Type: string
  • Default: none
  • Importance: Medium
  • Valid values / Notes: Kafka Connect 3.9.1 accepts only none or all. Do not use data here even though the Connector ConfigDef accepts it, because the framework rejects it. Use all to enable general continued error processing and the DLQ.

​
mongo.errors.tolerance

Independently overrides the Connector’s internal fault-tolerance mode.
  • Type: string
  • Default: none
  • Importance: Medium
  • Valid values / Notes: Valid values are none, data, and all. To tolerate only Connector data errors on Kafka Connect 3.9.1, configure data here.

​
errors.log.enable

Enables error logging for the Kafka Connect framework and the Connector.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: When set to true, tolerated Connector errors are also logged. Errors that are not tolerated are logged regardless of this value.

​
mongo.errors.log.enable

Independently overrides the Connector’s internal error logging setting.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: When explicitly configured, it overrides errors.log.enable. Connector errors that are not tolerated are still logged.

​
errors.retry.timeout

Configures the total retry duration for retryable processing stages in the Kafka Connect framework, in milliseconds.
  • Type: long
  • Default: 0
  • Importance: Medium
  • Valid values / Notes: 0 disables framework retries, while -1 means retry indefinitely. This setting does not guarantee retries for MongoDB bulk writes and does not restore the removed max.num.retries or retries.defer.timeout behavior. MongoDB Driver retryable writes are controlled by Driver options such as those in connection.uri.

​
errors.retry.delay.max.ms

Configures the maximum delay between consecutive framework retries, in milliseconds.
  • Type: long
  • Default: 60000
  • Importance: Medium
  • Valid values / Notes: Relevant only when errors.retry.timeout allows retries. After reaching the limit, Kafka Connect adds random jitter.

​
Dead Letter Queue

​
errors.deadletterqueue.topic.name

Specifies the Kafka topic that receives erroneous Sink records.
  • Type: string
  • Default: Empty string
  • Importance: Medium
  • Valid values / Notes: An empty value disables the DLQ. This topic cannot be included in topics or match topics.regex. Usually configure errors.tolerance=all together with the corresponding Connector fault-tolerance mode.

​
errors.deadletterqueue.topic.replication.factor

Specifies the replication factor used when Kafka Connect automatically creates the DLQ topic.
  • Type: short
  • Default: 3
  • Importance: Medium
  • Valid values / Notes: Used only when the DLQ topic does not already exist. The Kafka cluster must support the value.

​
errors.deadletterqueue.context.headers.enable

Adds __connect.errors.* context headers to DLQ records.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid values / Notes: Relevant only when a DLQ is configured. Headers can contain runtime metadata and must be handled according to data governance requirements.

​
Best Practices

​
Use Stable Record Identity to Control Write Results After Recovery

Applicable scenario: The Connector is already writing continuously, and the business requires the same Kafka record to target the same MongoDB document after a Worker restart or offset replay, rather than creating another document because of the default random ObjectId. This strategy is suitable when records are stored according to Kafka record identity and different events do not need to be merged into the same business entity. Use the following complete Connector configuration based on the Quick Start configuration:
KafkaMetaDataStrategy generates a stable _id from the topic, partition, and offset. Together with the default replacement-and-upsert write model, this allows a replay of the same record to target the same document. It does not provide end-to-end exactly-once semantics: there is no atomic transaction between MongoDB writes and Kafka offset commits, and triggers, validation, and other external side effects can still run again.

​
Scale Throughput by Partition Count and Write Pressure

Applicable scenario: The basic pipeline is running and must use multiple Kafka partitions to increase parallelism while limiting the number of records in each MongoDB bulk write to balance throughput, request size, latency, and the scope of failures. Use the following complete Connector configuration based on the Quick Start configuration, and adjust the example values according to the actual topic partition count and MongoDB capacity:
tasks.max is only the maximum allowed number of tasks; actual parallelism is limited by topic partition assignment, and multiple tasks can write to the same collection concurrently. max.batch.size splits batches by record count rather than BSON byte size. Increasing the value can usually reduce request count, but it can also increase per-batch latency and the scope of errors. No global ordering or atomicity is provided across tasks, batches, or namespaces.

​
Send Tolerated Errors to a Dead Letter Queue

Applicable scenario: The pipeline has entered stable operations. Individual malformed records or failed writes should not immediately stop the entire task, while operators need to retain the erroneous records and context for investigation, correction, and replay. Use the following complete Connector configuration based on the Quick Start configuration, and create a separate DLQ topic in advance:
The DLQ topic cannot also be an input topic. Tolerated records are excluded from MongoDB writes and allow current processing to continue; this is not an automatic repair or retry mechanism. Continuously monitor the DLQ, verify that it does not contain sensitive context, and establish a replay process after correction. The outcome of a write that reports a Write Concern error can be uncertain, so a DLQ record alone does not prove that the target was not written.

​
Monitoring

​
What to Monitor

Monitor Kafka Connect health, Connector and task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals. Monitor DLQ activity only when the corresponding error handling is enabled.

​
Import the Grafana Dashboard

Confirm that Connect metrics are available in a Grafana data source and that the collected labels satisfy the dashboard filters. Download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

​
Limitations

  • There is no atomic transaction between MongoDB writes and Kafka offset commits. If a Worker stops after a write succeeds but before the offset is persisted, it can process the same record again after recovery. Therefore, this behavior cannot be used to claim exactly-once processing, no duplicates, or no data loss.
  • Writes do not preserve atomicity or global ordering across batches, namespaces, or tasks. Increasing tasks.max can cause multiple tasks to write to the same collection concurrently.
  • Enabling change.data.capture.handler bypasses the standard post-processing chain, document ID strategy, standard WriteModel strategy, and tombstone deletion configuration. These paths cannot be combined to compose behavior.
  • InsertOne has no upsert behavior. Replaying a stable _id can cause a duplicate-key error, while replaying a random _id can create additional documents.
  • Tolerated error records are not automatically retried by the Connector. Without an available Errant Record Reporter or a configured DLQ, these records can be skipped without being copied to a DLQ.
  • Time-series mode and the Stable API require MongoDB 5.0 or later, and a time-series collection’s time field must be a BSON date-time or a value that can be converted successfully.

​
FAQ

​
The Connector Reports That a Topic Must Be Configured During Startup

Validation fails when both topics and topics.regex are empty or when both are non-empty. Keep only one selection method: use topics to list topics explicitly, or configure topics.regex with a valid Java regular expression for dynamic matching. Also confirm that the DLQ topic is outside the input range.

​
Records Are Written to the Wrong Database or Collection

The default mapper uses database and uses the Kafka topic name when collection is empty. Check whether topic.override.<topic>.<property> or FieldPathNamespaceMapper is configured. Field routing values must be strings, and the key and value cannot both provide fields for the same namespace component. To fail immediately when a field is invalid, set namespace.mapper.error.if.invalid=true.

​
Additional Documents Appear in MongoDB After a Restart

The default BsonOidStrategy generates a new ObjectId each time a record is processed, so offset replay can create a new document. Select a stable ID strategy based on business identity, such as generating an ID from the Kafka topic, partition, and offset or extracting an ID from a stable message key or value, and evaluate replay results together with the replacement or update WriteModel. A stable ID controls only target document selection and does not provide exactly-once processing.

​
A Record with a Null Value Does Not Delete the Document

By default, delete.on.null.values=false. After it is enabled, the default deletion strategy also requires FullKeyStrategy, PartialKeyStrategy, or ProvidedInKeyStrategy, and the message key must form a valid deletion condition. When a CDC Handler is configured, this setting is ignored; inspect the message according to the deletion event format of the corresponding CDC Handler.

​
Validation Fails When errors.tolerance=data Is Configured

The MongoDB Connector ConfigDef accepts data, but the Kafka Connect 3.9.1 framework configuration accepts only none or all. To tolerate only data-processing errors inside the Connector, configure mongo.errors.tolerance=data in the Connector-specific setting. Depending on whether framework-level continued processing or the DLQ is required, keep errors.tolerance as none or set it to all. mongo.errors.tolerance does not control errors in the key or value converter stage; Kafka Connect handles those errors according to the framework-level errors.tolerance setting.

​
No Error Records Appear in the DLQ After Fault Tolerance Is Configured

Check that errors.deadletterqueue.topic.name is non-empty, that the DLQ topic is separate from the input topics, that errors.tolerance allows the framework to continue processing, and whether mongo.errors.tolerance overrides the Connector’s internal fault tolerance. Also confirm that the runtime provides an available Errant Record Reporter. Otherwise, the Connector can use an empty Reporter and will not write error records to the DLQ.