> ## Documentation Index
> Fetch the complete documentation index at: https://docs.automq.com/llms.txt
> Use this file to discover all available pages before exploring further.

# MongoDB Sink Connector

> Configure and operate the MongoDB Sink Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting.

## 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](../manage-connectors).

```properties theme={null}
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
topics=mongodb-sink-input
connection.uri=mongodb://<mongodb-host>:27017
database=inventory
collection=events
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
```

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:

```properties theme={null}
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
topics=mongodb-sink-input
connection.uri=mongodb://<mongodb-host>:27017
database=inventory
collection=events
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.KafkaMetaDataStrategy
```

`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:

```properties theme={null}
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
topics=mongodb-sink-input
connection.uri=mongodb://<mongodb-host>:27017
database=inventory
collection=events
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
tasks.max=4
max.batch.size=1000
```

`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:

```properties theme={null}
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
topics=mongodb-sink-input
connection.uri=mongodb://<mongodb-host>:27017
database=inventory
collection=events
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
errors.tolerance=all
mongo.errors.tolerance=all
errors.log.enable=true
errors.deadletterqueue.topic.name=mongodb-sink-dlq
errors.deadletterqueue.context.headers.enable=true
```

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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), 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.