Debezium MongoDB

The Debezium MongoDB source connector tracks either a MongoDB replica set or a MongoDB sharded cluster for document changes in databases and collections. It streams those changes as messages to a Pulsar topic.

The connector automatically handles the following:

  • Addition and removal of shards in a sharded cluster.

  • Changes in membership for each replica set.

  • Elections within each replica set.

  • Resolution of communications issues with replica set members.

Compatibility

Astra Streaming supports Apache Pulsar™ 3.1, which uses Debezium 1.7 libraries.

For supported database versions, see Debezium 1.7 compatibility.

Create the connector

  1. Optional: If you are using the pulsar-admin CLI or Pulsar Admin API, set the following commonly-used environment variables:

    export TENANT="TENANT_NAME"
export TOPIC="OUTPUT_TOPIC_NAME"
export NAMESPACE="NAMESPACE_NAME" # or default
export SOURCE_NAME="SOURCE_CONNECTOR_UNIQUE_NAME"
export PULSAR_TOKEN="TENANT_PULSAR_TOKEN" # API only
export WEB_SERVICE_URL="TENANT_PULSAR_WEB_SERVICE_URL" # API only

    SOURCE_NAME is the name for your new source connector. DataStax recommends a memorable, human-readable name that summarizes the connector’s purpose. For example: debezium-mongodb-source-prod-us-east-1.

  2. Create the connector using JSON-formatted connector configuration data. You can pass the configuration directly or with a configuration file.

    pulsar-admin CLI
    ./bin/pulsar-admin sources create \
  --source-type debezium-mongodb \
  --name "$SOURCE_NAME" \
  --destination-topic-name "persistent://$TENANT/$NAMESPACE/$TOPIC" \
  --tenant "$TENANT" \
  --source-config-file configs.json
    Pulsar Admin API
    curl -sS --fail -L -X POST "$WEB_SERVICE_URL/admin/v3/sources/$TENANT/$NAMESPACE/$SOURCE_NAME" \
  --header "Authorization: Bearer $PULSAR_TOKEN" \
  --form "sourceConfig=@mynetty-source-config.json;type=application/json"
    Example configuration data structure
    {
  "tenant": "${TENANT}",
  "namespace": "${NAMESPACE}",
  "topicName": "persistent://${TENANT}/${NAMESPACE}/${TOPIC}",
  "name": "${SOURCE_NAME}",
  "producerConfig": {
    "maxPendingMessages": 0,
    "maxPendingMessagesAcrossPartitions": 0,
    "useThreadLocalProducers": true,
    "cryptoConfig": {
      "cryptoKeyReaderClassName": "string",
      "cryptoKeyReaderConfig": {
        "property1": {},
        "property2": {}
      },
      "encryptionKeys": [
        "string"
      ],
      "producerCryptoFailureAction": "FAIL",
      "consumerCryptoFailureAction": "FAIL"
    },
    "batchBuilder": "string"
  },
  "serdeClassName": "string",
  "schemaType": "string",
  "configs": {
    "database.whitelist": "usersdb",
    "mongodb.hosts": "host:port",
    "mongodb.name": "streaming_prod",
    "mongodb.password": "${PASSWORD}",
    "mongodb.task.id": "123",
    "mongodb.user": "mongodb"
  },
  "secrets": {},
  "parallelism": 1,
  "processingGuarantees": "ATLEAST_ONCE",
  "resources": {
    "cpu": 0.25,
    "disk": 1000000000,
    "ram": 1000000000
  },
  "archive": "builtin://debezium-mongodb",
  "runtimeFlags": "string",
  "customRuntimeOptions": "string",
  "batchSourceConfig": {
    "discoveryTriggererClassName": "string",
    "discoveryTriggererConfig": {
      "property1": {},
      "property2": {}
    }
  },
  "batchBuilder": "string"
}

Edit the connector

To update a connector, pass the new configuration definition to the connector. For example, if you created the connector with a configuration file, you can pass an updated configuration file.

You can include the entire configuration or only the properties that you want to change. Additionally, some properties can be modified with specific arguments, such as --parallelism.

To get the current configuration, see Get source connector configuration data.

pulsar-admin CLI
./bin/pulsar-admin sources update \
  --source-type debezium-mongodb \
  --name "$SOURCE_NAME" \
  --destination-topic-name "persistent://$TENANT/$NAMESPACE/$TOPIC" \
  --tenant "$TENANT" \
  --parallelism 2
Pulsar Admin API
curl -sS --fail -L -X PUT "$WEB_SERVICE_URL/admin/v3/sources/$TENANT/$NAMESPACE/$SOURCE_NAME" \
  --header "Authorization: Bearer $PULSAR_TOKEN" \
  --form "sourceConfig=@mynetty-source-config.json;type=application/json"

Manage the connector

See Create and manage connectors for details on how to manage connectors after you create them, including:

  • Get connector status

  • Get existing connectors

  • Get connector configuration details

  • Start connectors

  • Stop connectors

  • Restart connectors

  • Delete connectors

Connector configuration reference

To configure Astra Streaming Pulsar connectors, you use a combination of common Pulsar properties and provider-specific properties.

Because Astra Streaming and the Astra Streaming Pulsar connectors are based on a specific version of the open-source Apache Pulsar project and connectors, some properties and values might be unsupported or ignored by Astra Streaming.

Pulsar source connector properties

Pulsar connectors and functions both use Pulsar functions workers. Therefore, some function configuration properties are also used to configure connectors.

The following table lists Astra Streaming Pulsar source connector configuration properties by JSON field name. For the equivalent command line arguments and more information about these properties, see the following:

Pulsar configuration properties for source connectors
Field name Required Description

archive

Yes

The type of built-in Astra Streaming Pulsar connector that you want to deploy. Formatted as builtin://CONNECTOR, such as builtin://kafka.

batchBuilder

No

The batch construction method: DEFAULT or KEY_BASED.

Default: DEFAULT

batchSourceConfig

No

The batch source configuration key-value as a JSON string.

configs

Yes

A JSON-formatted key-value map containing configuration properties specific to the connector type, including provider-specific authentication and integration settings. For available subproperties, see Debezium MongoDB source connector properties (configs).

Default: {} (Empty map, uses defaults if they exist, fails otherwise)

customRuntimeOptions

No

A string that encodes options to configure the Apache Pulsar function runtime.

name

Yes

The name for the connector. It must start with a lowercase letter, and contain only numbers, hyphens (-), and lowercase letters. DataStax recommends a memorable, human-readable name that summarizes the connector’s purpose. For example: debezium-mongodb-prod-us-east-1.

namespace

Yes

The namespace in your Pulsar tenant where you want to create the connector.

parallelism

Yes

The number of Pulsar function instances to run.

Default: 1

processingGuarantees

Yes

The messaging delivery semantic to use when writing to topics: ATLEAST_ONCE, ATMOST_ONCE, or EFFECTIVELY_ONCE.

Respect for the processing guarantee depends on the connector implementation.

For more information, see the Pulsar documentation on Function processing guarantees and Processing guarantees in I/O connectors.

Default: ATLEAST_ONCE

producerConfig

No

A JSON string containing a custom producer configuration. By default, most values are null or empty. For example:

  "producerConfig": {
    "maxPendingMessages": null,
    "maxPendingMessagesAcrossPartitions": null,
    "useThreadLocalProducers": false,
    "cryptoConfig": null,
    "batchBuilder": ""
  },

resources

No

A JSON string describing the compute resources to allocate to each Pulsar function instance. For example: {"cpu":0.25,"disk":1000000000,"ram":500000000}.

runtimeFlags

No

A string that encodes flags to pass to the Apache Pulsar function runtime. Only applicable to process and Kubernetes runtimes.

schemaType

No

The schema type used to encode messages emitted from a Pulsar source. Accepts either a built-in schema like avro or json, or a custom Schema class name.

secrets

No

If security is enabled on your function workers, you can provide a map of secret names (secretName) to secrets provider configuration objects. The secretName is used by the connector to reference the secret. The mapped object contains the required properties to fetch the secret from the secrets provider.

To get the types for the values in this mapping, use the SecretProviderConfigurator.getSecretObjectType() method.

This is separate from connector-specific security settings in configs and the Pulsar authentication token used to connect to your Pulsar cluster.

serdeClassName

No

The SerDe classname for the source.

tenant

Yes

The Pulsar tenant where you want to create the connector.

topicName

Yes

An existing topic in your Astra Streaming tenant where messages are published. Must be formatted as a full persistent or non-persistent topic name, such as persistent://$TENANT/$NAMESPACE/$TOPIC.

typeClassName

Yes

The class reference for the connector type. Begins with org.apache.pulsar, such as org.apache.pulsar.common.schema.KeyValue or org.apache.pulsar.io.debezium.mysql.DebeziumMysqlSource.

Debezium MongoDB source connector properties (configs)

Set these properties in the configs section of the connector configuration.

The following table lists some commonly used properties. For more information and additional properties, see the documentation for the Debezium connector for MongoDB and the OSS Apache Pulsar Debezium source connector.

Name Required Description

database.whitelist

No if alternative properties are used

A list of databases hosted by the server monitored by the connector.

There are multiple properties you can use to specify (inclusively or exclusively) the databases and tables to monitor.

mongodb.hosts

Yes

A comma-separated list of hosts (either as host or host:port) for the MongoDB servers in the replica set.

If mongodb.members.auto.discover is false, then you must prefix hosts with the replica set name, such as rs0/localhost:27017.

mongodb.name

Yes

A unique name that identifies the connector, MongoDB replica set, or shared cluster that the source connector monitors.

Use only one Debezium MongoDB source connector for each MongoDB server.

mongodb.password

Yes if authentication is enabled

The password to authenticate to the MongoDB server.

mongodb.task.id

Yes

The taskId of the MongoDB connector that attempts to use a separate task for each replica set.

mongodb.user

Yes if authentication is enabled

The name of the database user to authenticate to the MongoDB server.

Was this helpful?

Edit this page

Give Feedback

How can we improve the documentation?

© Copyright IBM Corporation 2026 | Privacy policy | Terms of use |  Manage Privacy Choices

Apache, Apache Cassandra, Cassandra, Apache Tomcat, Tomcat, Apache Lucene, Apache Solr, Apache Hadoop, Hadoop, Apache Pulsar, Pulsar, Apache Spark, Spark, Apache TinkerPop, TinkerPop, Apache Kafka and Kafka are either registered trademarks or trademarks of the Apache Software Foundation or its subsidiaries in Canada, the United States and/or other countries. Kubernetes is the registered trademark of the Linux Foundation.

General Inquiries: Contact IBM