> ## 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 Source Connector

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

## Overview

The MongoDB Source Connector uses MongoDB Change Streams to read changes from a database, a collection, or an entire deployment and write the events to Kafka topics. It is suitable for data pipelines such as data synchronization, stream processing, search indexing, and auditing. By default, the MongoDB namespace `database.collection` maps to a Kafka topic with the same name, and the message value contains the complete change stream event.

The Connector can capture only changes that occur after startup, or it can first copy existing documents and then continue with subsequent changes. Copied documents are wrapped as Change Stream events shaped like insert operations and pass through the same topic, key, and value mapping process as real-time events.

## Prerequisites

* The MongoDB deployment must support Change Streams. The connection account needs the permissions required to read change streams for the capture scope, and it also needs permission to read the target collections when copying existing data.
* When using MongoDB capabilities such as pre-change documents, expanded events, or the Stable API, the server version and database settings must meet the requirements of the corresponding feature.

## License

Uses Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka, and a MongoDB deployment that supports Change Streams, and confirm network connectivity and access permissions. For preparation and management, see [Manage connectors](../manage-connectors).

```properties theme={null}
connector.class=com.mongodb.kafka.connect.MongoSourceConnector
connection.uri=mongodb://<username>:<password>@<mongodb-host>:27017/?replicaSet=<replica-set>
database=<database-name>
collection=<collection-name>
```

Replace the MongoDB connection information, database name, and collection name. This configuration starts capturing from the current change stream position and writes events to the `<database-name>.<collection-name>` topic by default. Use a Config Provider or protected configuration management for connection credentials.

## Configuration

### Connector and Tasks

#### `connector.class`

Specifies the MongoDB Source Connector implementation class.

* **Type**: `string`
* **Default**: None
* **Importance**: High
* **Valid Values / Notes**: Use `com.mongodb.kafka.connect.MongoSourceConnector`.
* **Required**: Yes

#### `tasks.max`

Sets the maximum number of Tasks that Connect may create.

* **Type**: `int`
* **Default**: `1`
* **Importance**: High
* **Valid Values / Notes**: Must be at least `1`. This Connector always creates only one Source Task, so a larger value does not increase Task parallelism.

### MongoDB Connection and Capture Scope

#### `connection.uri`

The MongoDB connection string, which specifies hosts, authentication, the Replica Set, TLS, and other driver options.

* **Type**: `password`
* **Default**: `mongodb://localhost:27017,localhost:27018,localhost:27019`
* **Importance**: High
* **Valid Values / Notes**: Must be a valid MongoDB URI. The default is generally suitable only for local development. Do not store credentials in plaintext configuration.

#### `database`

Restricts monitoring to a database.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: When empty, monitors the entire MongoDB deployment. Setting a database without setting a collection monitors the collections in that database.

#### `collection`

Restricts monitoring to a single collection.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: This is not a regular expression or a list. When set, `database` must also be set.

#### `offset.partition.name`

Overrides the name the Connector uses to identify the Source Offset partition.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: When empty, the name is generated from the connection hosts, database, and collection. Changing this value starts using a new Offset identity; the existing Resume Token is not migrated automatically.

### MongoDB Stable API

#### `server.api.version`

Enables the specified version of the MongoDB Stable API.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: An empty value disables it. The available value is `1`, which requires MongoDB 5.0 or later.

#### `server.api.deprecation.errors`

Enables Stable API deprecation errors.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Applies only when `server.api.version` is not empty.

#### `server.api.strict`

Enables Stable API strict mode.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Applies only when `server.api.version` is not empty.

### TLS Certificates

#### `connection.ssl.truststore`

Specifies the Truststore path local to the Connect Worker.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: When not empty, it is loaded using the JVM default KeyStore type. TLS must still be enabled through the connection URI or deployment settings.

#### `connection.ssl.truststorePassword`

Specifies the Truststore password.

* **Type**: `password`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Used only when `connection.ssl.truststore` is not empty.

#### `connection.ssl.keystore`

Specifies the Keystore path local to the Connect Worker for providing a client certificate and private key.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: When not empty, it is loaded using the JVM default KeyStore type.

#### `connection.ssl.keystorePassword`

Specifies 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 not empty. The same password is used to load the KeyStore and initialize the private key.

### Change Stream

#### `pipeline`

Sets an aggregation pipeline for the Change Stream.

* **Type**: `string`
* **Default**: `[]`
* **Importance**: Medium
* **Valid Values / Notes**: Must be an array of JSON documents. The pipeline must preserve the `_id`, `ns`, `documentKey`, or `fullDocument` fields required for topic, Offset, key, and value mapping. For efficient filtering while copying existing data, prefer `startup.mode.copy.existing.pipeline`.

#### `batch.size`

Sets the batch size hint for the MongoDB Change Stream Cursor.

* **Type**: `int`
* **Default**: `0`
* **Importance**: Medium
* **Valid Values / Notes**: Must be at least `0`. `0` retains driver and server behavior and is different from `poll.max.batch.size`.

#### `publish.full.document.only`

Publishes only the change event's `fullDocument` as the message value.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: High
* **Valid Values / Notes**: When set to `true`, forces `updateLookup`. Events without `fullDocument` are filtered unless delete Tombstones are also enabled.

#### `publish.full.document.only.tombstone.on.delete`

In full-document-only mode, sends a null-valued Tombstone for events that do not contain a document-valued `fullDocument`; delete events are the most common case.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Applies only when `publish.full.document.only=true`. The implementation checks for a document-valued `fullDocument`; it does not only check whether the event's `operationType` is `delete`.

#### `change.stream.document.key.as.key`

Controls whether non-Schema output uses the Change Stream `documentKey` as the Kafka message key.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: Medium
* **Valid Values / Notes**: When set to `false`, or when the event has no `documentKey`, the key uses the Resume Token document. This setting is ignored when `output.format.key=schema`.

#### `change.stream.full.document.before.change`

Sets how the Change Stream returns the document from before the change.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Available values are `default`, `off`, `whenAvailable`, `required`, or an empty string. Requires MongoDB 6.0 or later and Pre-images enabled for the collection. Does not affect records copied from existing data.

#### `change.stream.full.document`

Sets how the Change Stream returns the full document.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Available values are `default`, `updateLookup`, `whenAvailable`, `required`, or an empty string. `publish.full.document.only=true` forces `updateLookup`.

#### `change.stream.show.expanded.events`

Controls whether expanded Change Stream events are requested.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Expanded DDL events require MongoDB 6.0 or later, and disambiguated update paths require MongoDB 6.1 or later.

#### `collation`

Sets the Collation for the Change Stream.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: An empty value leaves it unset; otherwise, it must be a valid MongoDB Collation JSON document. This setting does not apply to aggregation for copying existing data.

#### `poll.max.batch.size`

Limits the maximum number of records returned by a single `poll`.

* **Type**: `int`
* **Default**: `1000`
* **Importance**: Low
* **Valid Values / Notes**: Must be at least `1`. It limits only the number of records, not the total bytes in a batch.

#### `poll.await.time.ms`

Sets the maximum time that the Change Stream waits for new events.

* **Type**: `long`
* **Default**: `5000`
* **Importance**: Low
* **Valid Values / Notes**: Must be at least `1`, in milliseconds. This value is not the Kafka Connect Poll Timeout.

### Topic Mapping

#### `topic.mapper`

Specifies the implementation class that maps MongoDB namespaces to Kafka topics.

* **Type**: `string`
* **Default**: `com.mongodb.kafka.connect.source.topic.mapping.DefaultTopicMapper`
* **Importance**: High
* **Valid Values / Notes**: Must be a loadable fully qualified Java class name that implements `TopicMapper`. A custom implementation may read additional properties that it defines.

#### `topic.separator`

Specifies the separator between parts of a default topic name.

* **Type**: `string`
* **Default**: `.`
* **Importance**: Low
* **Valid Values / Notes**: Used to join the prefix, database, collection, and suffix. The Connector does not validate whether the final topic name is valid.

#### `topic.prefix`

Adds a prefix to all topics produced by the default mapping.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Low
* **Valid Values / Notes**: A non-empty prefix is followed by `topic.separator`.

#### `topic.suffix`

Adds a suffix to all topics produced by the default mapping.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Low
* **Valid Values / Notes**: A non-empty suffix is preceded by `topic.separator`.

#### `topic.namespace.map`

Uses a JSON object to map MongoDB namespaces to specified topics.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Supports full namespaces, databases, `/regular expression/` mappings matched in declaration order, and `*` mappings. Prefixes and suffixes are still applied to mapping results. Templates may use the `db`, `sep`, `coll`, `sep_coll`, `coll_sep`, and `sep_coll_sep` variables. You must prevent invalid names and topic conflicts.

### Output Format and Schema

#### `output.format.key`

Sets the output format of message keys.

* **Type**: `string`
* **Default**: `json`
* **Importance**: High
* **Valid Values / Notes**: Available values are `json`, `bson`, and `schema`, case-insensitive. The format should be compatible with `key.converter`.

#### `output.format.value`

Sets the output format of message values.

* **Type**: `string`
* **Default**: `json`
* **Importance**: High
* **Valid Values / Notes**: Available values are `json`, `bson`, and `schema`, case-insensitive. The format should be compatible with `value.converter`.

#### `output.schema.infer.value`

Automatically infers a Connect Schema for each message value.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Applies only when `output.format.value=schema`. Changes in document structure may cause multiple Schemas to appear in the same topic.

#### `output.schema.key`

Defines the Avro Schema used for Schema-format message keys.

* **Type**: `string`
* **Default**: `{  "type": "record",  "name": "keySchema",  "fields" : [{"name": "_id", "type": "string"}]}`
* **Importance**: High
* **Valid Values / Notes**: Must be a valid Connector Avro Schema compatible with conversion of the complete Change Stream document. Read only when `output.format.key=schema`.

#### `output.schema.value`

Defines the Avro Schema used for Schema-format message values.

* **Type**: `string`
* **Default**: Built-in `ChangeStream` Avro Schema
* **Importance**: High
* **Valid Values / Notes**: Must be a valid Connector Avro Schema compatible with the message value. Ignored when `output.schema.infer.value=true` or the output format is not `schema`.

#### `output.json.formatter`

Specifies the BSON-to-JSON formatting implementation class.

* **Type**: `string`
* **Default**: `com.mongodb.kafka.connect.source.json.formatter.DefaultJson`
* **Importance**: Medium
* **Valid Values / Notes**: Must be a loadable fully qualified Java class name that implements `JsonWriterSettingsProvider`. It is used for JSON output and Schema conversion and ignored for raw BSON output.

#### `key.converter`

Overrides the Worker-level message key Converter.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: `null` inherits the Worker configuration. The selected Converter must be compatible with the String, BSON Bytes, or Schemaful Connect Data produced by `output.format.key`.

#### `value.converter`

Overrides the Worker-level message value Converter.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: `null` inherits the Worker configuration. The selected Converter must be compatible with the String, BSON Bytes, or Schemaful Connect Data produced by `output.format.value`.

### Startup Position and Existing Data Copy

#### `startup.mode`

Sets the startup behavior when no usable Source Offset is available.

* **Type**: `string`
* **Default**: Empty string, treated as `latest`
* **Importance**: Medium
* **Valid Values / Notes**: Available values are `latest`, `timestamp`, `copy_existing`, or an empty string. An existing Resume Token takes precedence over this setting. An explicit non-empty value takes precedence over the deprecated `copy.existing`.

#### `startup.mode.timestamp.start.at.operation.time`

Sets the starting operation time for Timestamp startup mode.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Applies only when `startup.mode=timestamp` and no usable Offset is available. Supports decimal Epoch seconds, an ISO-8601 Instant with second precision, or a Canonical Extended JSON BSON Timestamp.

#### `startup.mode.copy.existing.max.threads`

Sets the number of read threads for copying existing data.

* **Type**: `int`
* **Default**: `Runtime.getRuntime().availableProcessors()` (number of processors available to the runtime)
* **Importance**: Medium
* **Valid Values / Notes**: Must be at least `1` and is used only when `startup.mode=copy_existing`. The actual thread count does not exceed the number of selected namespaces. When explicitly set, it overrides the deprecated `copy.existing.max.threads`.

#### `startup.mode.copy.existing.queue.size`

Sets the capacity of the in-memory queue between existing-data copy threads and the Source Task.

* **Type**: `int`
* **Default**: `16000`
* **Importance**: Medium
* **Valid Values / Notes**: Must be at least `1` and is used only when `startup.mode=copy_existing`. When explicitly set, it overrides the deprecated `copy.existing.queue.size`.

#### `startup.mode.copy.existing.pipeline`

Sets an aggregation pipeline used only to read existing documents.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Must be empty or an array of JSON documents and is used only when `startup.mode=copy_existing`. It runs before synthetic change events are created and before the general `pipeline`. When explicitly set, it overrides the deprecated `copy.existing.pipeline`.

#### `startup.mode.copy.existing.namespace.regex`

Uses a regular expression to filter existing-data copy namespaces discovered at startup.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Matches the full `database.collection` using Java regular expression substring matching semantics and is used only when `startup.mode=copy_existing`. When explicitly set, it overrides the deprecated `copy.existing.namespace.regex`.

#### `startup.mode.copy.existing.allow.disk.use`

Controls whether MongoDB may use disk for aggregation while copying existing data.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: Medium
* **Valid Values / Notes**: Used only when `startup.mode=copy_existing`. When explicitly set, it overrides the deprecated `copy.existing.allow.disk.use`.

#### `copy.existing`

Enables copying existing data with the legacy switch.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Applies only when `startup.mode` is empty.
* **Deprecated**: Yes
* **Replacement**: `startup.mode=copy_existing`

#### `copy.existing.max.threads`

Sets the legacy thread count for copying existing data.

* **Type**: `int`
* **Default**: `Runtime.getRuntime().availableProcessors()` (number of processors available to the runtime)
* **Importance**: Medium
* **Valid Values / Notes**: Must be at least `1`. It is used as a fallback only when the replacement is not explicitly set.
* **Deprecated**: Yes
* **Replacement**: `startup.mode.copy.existing.max.threads`

#### `copy.existing.queue.size`

Sets the legacy in-memory queue capacity for copying existing data.

* **Type**: `int`
* **Default**: `16000`
* **Importance**: Medium
* **Valid Values / Notes**: Must be at least `1`. It is used as a fallback only when the replacement is not explicitly set.
* **Deprecated**: Yes
* **Replacement**: `startup.mode.copy.existing.queue.size`

#### `copy.existing.pipeline`

Sets the legacy aggregation pipeline for copying existing data.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Must be empty or an array of JSON documents. It is used as a fallback only when the replacement is not explicitly set.
* **Deprecated**: Yes
* **Replacement**: `startup.mode.copy.existing.pipeline`

#### `copy.existing.namespace.regex`

Sets the legacy namespace regular expression for copying existing data.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Must be a valid Java regular expression. It is used as a fallback only when the replacement is not explicitly set.
* **Deprecated**: Yes
* **Replacement**: `startup.mode.copy.existing.namespace.regex`

#### `copy.existing.allow.disk.use`

Sets whether legacy existing-data copy aggregation may use disk.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: Medium
* **Valid Values / Notes**: It is used as a fallback only when the replacement is not explicitly set.
* **Deprecated**: Yes
* **Replacement**: `startup.mode.copy.existing.allow.disk.use`

### Error Handling and Heartbeats

#### `errors.tolerance`

Sets the error tolerance mode for the Kafka Connect framework and the MongoDB Connector.

* **Type**: `string`
* **Default**: `none`
* **Importance**: Medium
* **Valid Values / Notes**: Available values are `none` and `all`, case-insensitive. `all` can skip conversion failures and allow some Change Stream recovery paths to rebuild without a Resume Token. `mongo.errors.tolerance` overrides only the Connector's reading of this value.

#### `mongo.errors.tolerance`

Overrides the error tolerance mode only within the MongoDB Connector.

* **Type**: `string`
* **Default**: `none`
* **Importance**: Medium
* **Valid Values / Notes**: Available values are `none` and `all`. Whenever explicitly set, it takes precedence over the Connector's reading of `errors.tolerance`, but it does not replace the Kafka Connect framework setting with the same name.

#### `errors.log.enable`

Controls whether the Kafka Connect framework and the MongoDB Connector log tolerated errors.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: The Connector still logs conversion errors when the tolerance mode is `none`. `mongo.errors.log.enable` can override only the Connector's behavior.

#### `mongo.errors.log.enable`

Overrides only whether the MongoDB Connector logs tolerated conversion errors.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: When explicitly set, it takes precedence over the Connector's reading of `errors.log.enable` and does not change the Kafka Connect framework logging setting.

#### `errors.deadletterqueue.topic.name`

Sets the DLQ topic to which records with MongoDB conversion failures are written.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Used only when the Connector's effective tolerance mode is `all`. The Connector does not create or validate the topic. `mongo.errors.deadletterqueue.topic.name` can override the Connector's reading of this value.

#### `mongo.errors.deadletterqueue.topic.name`

Overrides only the DLQ topic used by the MongoDB Connector for records with conversion failures.

* **Type**: `string`
* **Default**: Empty string
* **Importance**: Medium
* **Valid Values / Notes**: Requires the Connector's effective error tolerance mode to be `all`. It affects only Connector conversion error handling.

#### `heartbeat.interval.ms`

Sets the minimum interval for generating heartbeat records when there are no business records.

* **Type**: `long`
* **Default**: `0`
* **Importance**: Medium
* **Valid Values / Notes**: Must be at least `0`, in milliseconds. `0` disables heartbeats. A heartbeat is generated only when a new Post-batch Resume Token appears and the current Poll contains no business records.

#### `heartbeat.topic.name`

Specifies the Kafka topic to which heartbeat records are written.

* **Type**: `string`
* **Default**: `__mongodb_heartbeats`
* **Importance**: Medium
* **Valid Values / Notes**: Must be a non-empty string and is used only when `heartbeat.interval.ms` is greater than `0` and a heartbeat is generated. The topic must already meet naming, permission, and availability requirements.

### Custom Credentials

#### `mongo.custom.auth.mechanism.enable`

Enables a custom MongoDB credential provider.

* **Type**: `string`
* **Default**: None
* **Importance**: Not declared
* **Valid Values / Notes**: Only the string value `true`, case-insensitive, enables it. This property is not registered in ConfigDef, so it does not appear in ConfigDef metadata and does not receive ConfigDef type validation.

#### `mongo.custom.auth.mechanism.providerClass`

Specifies the custom MongoDB credential provider implementation class.

* **Type**: `string`
* **Default**: None
* **Importance**: Not declared
* **Valid Values / Notes**: Required when custom authentication is enabled. The class must be loadable, implement `CustomCredentialProvider`, have an accessible no-argument constructor, and be available with its dependencies on every Worker where the Task may run.

## Best Practices

### Initially Load an Existing Collection and Continue Capturing Changes

Applicable scenario: The target collection already contains existing documents, and you need to establish a Kafka data baseline before continuously receiving new changes produced during and after the copy. This mode is suitable for initial onboarding, not as an initialization process to repeat on every restart.

Use the following complete Connector configuration based on the Quick Start configuration:

```properties theme={null}
connector.class=com.mongodb.kafka.connect.MongoSourceConnector
connection.uri=mongodb://<username>:<password>@<mongodb-host>:27017/?replicaSet=<replica-set>
database=<database-name>
collection=<collection-name>
startup.mode=copy_existing
```

Key considerations: At startup, the Connector copies existing documents in the selected scope and then continues with the Change Stream. Restarting after an interrupted copy repeats the entire copy, so consumers should handle duplicates by using stable business keys or another method. The copy is not a consistent snapshot across multiple collections.

### Assign Stable Topic Names to Business Namespaces

Applicable scenario: MongoDB database or collection names are unsuitable as downstream topic contracts, or existing topic names must remain unchanged during a migration. Explicit mapping decouples source naming from the Kafka consumer interface.

Use the following Connector configuration to route one namespace to a fixed topic:

```properties theme={null}
connector.class=com.mongodb.kafka.connect.MongoSourceConnector
connection.uri=mongodb://<username>:<password>@<mongodb-host>:27017/?replicaSet=<replica-set>
database=inventory
collection=orders
topic.namespace.map={"inventory.orders":"commerce.orders"}
```

Key considerations: Mapping occurs before prefix and suffix processing. Changing the mapping does not change the Source Offset identity, but you must confirm in advance the target topic's naming, permissions, partitioning strategy, and potential conflicts with other mapping results.

## 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

Ensure that Connect metrics are available in a Grafana data source and that the collected labels meet the dashboard filter requirements; 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

* Each Connector instance creates only one Source Task. Increasing `tasks.max` cannot horizontally scale Change Stream consumption.
* Restarting during `copy_existing` repeats the copy from the beginning, so existing records already written to Kafka may be duplicated. The process is also not a consistent snapshot across collections.
* Copying multiple namespaces may interleave records, topic mapping may split data streams, and the Connector does not specify the Kafka Partition. Therefore, it does not guarantee global ordering across namespaces, topics, or Partitions.
* `publish.full.document.only=true` filters events without `fullDocument` unless the Tombstone option is also enabled to generate null-valued records.
* When error tolerance is set to `all` without a configured DLQ, records with conversion failures are discarded, and subsequent Offsets may permanently advance past those events.
* When a Resume Token is invalid or its history has been lost, a tolerant recovery path may rebuild the Change Stream without a Token and cannot recover a data interval that has already left the Oplog.
* The Connector does not provide end-to-end Exactly-once, duplicate-free, or lossless guarantees from MongoDB to Kafka. Offset commit windows, copying existing data, and error tolerance can all produce duplicates or data gaps.

## FAQ

### No New Messages Arrive After the Connector Starts

Check whether the MongoDB deployment supports Change Streams, whether the connection account has permissions for the selected scope, and whether `database` and `collection` point to the location receiving writes. Confirm that the default `<database>.<collection>` topic or the target topic from `topic.namespace.map` is correct. If you use `pipeline`, also confirm that it does not filter out all events or remove required fields such as `_id` and `ns`.

### Update Events Do Not Contain the Latest Full Document

By default, a Change Stream event does not necessarily contain the complete document after an update. Set `change.stream.full.document=updateLookup` when the complete document is required. To publish only complete documents, set `publish.full.document.only=true`, but also evaluate the filtering of events without `fullDocument` and whether delete events require Tombstones.

### Capture Starts from an Unexpected Position After a Restart

The Connector prioritizes the committed Resume Token and applies `startup.mode` only when no usable Offset is available. Check whether the connection hosts, `database`, `collection`, or `offset.partition.name` changed, because these changes may create a new Offset identity. Restore the previous configuration, confirm that the original Offset still exists, and then restart the Connector.

### Duplicate Records Appear After Copying Existing Data

Documents read during an existing-data copy are published as insert events. Restarting after an interrupted copy repeats the entire copy, and concurrent updates may also appear as subsequent Change Stream events. Consumers should implement idempotent processing based on stable document keys and determine before restarting whether to continue using the original Offset or establish a new data baseline.

### Conversion Errors Do Not Enter the DLQ

Confirm that the effective Connector error tolerance mode is `all`, that the DLQ topic exists and the Worker has write permission, and whether any `mongo.errors.*` overrides are set. Filtering behavior such as an empty topic mapping or a missing `fullDocument` in full-document-only mode is not a conversion failure and is not written to the DLQ.