MongoDB Sink

The MongoDB sink connector writes messages from Iggy streams to a MongoDB database. Each message is inserted as a document into the configured collection.

Configuration

Use the broker credentials and runtime setup from the sink guide. From the matching 0.9.0/edge repository root, build the plugin:

cargo build --release -p iggy_connector_mongodb_sink

Save this connector file in the runtime's connector directory and replace the MongoDB URI and destination names for your deployment. Startup parses the URI and pings the target database. max_pool_size, when set, overrides the URI's pool setting; other connection options come from the MongoDB driver and URI.

type = "sink"
key = "mongodb-sink"
enabled = true
version = 1
name = "MongoDB Sink"
path = "target/release/libiggy_connector_mongodb_sink"
verbose = false

[[streams]]
stream = "my-stream"
topics = ["my-topic"]
schema = "json"
batch_length = 100
poll_interval = "1s"
consumer_group = "mongodb-sink-group"

[plugin_config]
connection_uri = "mongodb://localhost:27017"
database = "my_database"
collection = "my_collection"
batch_size = 100
include_metadata = true
include_checksum = true
include_origin_timestamp = true
payload_format = "json"
auto_create_collection = true
verbose_logging = false
max_retries = 3
retry_delay = "1s"

Create the Iggy resources and send a message:

./target/release/iggy --username iggy --password iggy stream create my-stream
./target/release/iggy --username iggy --password iggy topic create my-stream my-topic 1 none 1d
./target/release/iggy --username iggy --password iggy message send --partition-id 0 my-stream my-topic '{"hello":"mongodb","internal_field":"remove me"}'

Plugin config options

Stored Documents

Every document has a generated _id of stream:topic:partition:message_id and a payload field. The offset is not part of the ID. Payload fields remain nested under payload, so a payload's own _id does not replace the connector's ID. Message headers are not stored.

• json parses the payload and converts it to a BSON value under payload, including objects, arrays and scalars. Invalid JSON or an unsupported BSON conversion rejects the whole chunk before insertion.
• string and its text alias require valid UTF-8 and store a BSON string.
• binary stores BSON Binary with the generic subtype. Use schema = "raw" and no payload-changing transform to preserve original bytes.

Format names are case-insensitive. Formatting happens after stream decoding and transforms, so JSON decoding can change the original bytes.

With include_metadata = true, documents contain iggy_offset, iggy_timestamp, iggy_stream, iggy_topic and iggy_partition_id. Offsets above the signed 64-bit range use iggy_offset_str instead. Partition IDs use a BSON 32-bit integer when possible and a 64-bit integer otherwise. iggy_checksum is a 64-bit integer when possible and a decimal string otherwise. Message and origin timestamps become BSON datetimes, truncating microseconds to milliseconds. Checksum and origin timestamp flags are independent of include_metadata; _id and payload are always present.

Delivery Semantics

Each polled batch is split into chunks of at most batch_size messages. Chunks are inserted immediately with unordered insert_many; other documents can succeed when one is rejected. Later chunks are still attempted after a failure, and the last chunk error is returned. There is no accumulation across polls and no transaction covering the batch.

The connector retries transient insert failures up to max_retries total attempts, with linear waits of retry_delay, twice that delay, and so on. An invalid delay silently falls back to 1s. Driver-level retryable writes may add retries independently, according to the URI and MongoDB deployment. Startup has no connector-level retry loop.

An insert error containing only duplicate-key errors (11000) and no write-concern error is treated as success. This covers all unique indexes, not only _id, and does not compare payload contents. Reusing a message ID in one stream/topic/partition preserves the existing document. A conflict on another unique index can also discard a distinct message while runtime reports it processed. Account for this when choosing collection indexes.

The runtime auto-commits while polling, records plugin errors and continues without replaying the failed batch. Earlier or later chunks may already be stored, and a timeout can leave the write outcome uncertain. Duplicate tolerance and retries do not provide an end-to-end at-least-once guarantee.

Transforms

You can apply transforms before writing to MongoDB. Every [transforms.*] section requires enabled = true. See the transforms documentation for details.

[transforms.add_fields]
enabled = true

[[transforms.add_fields.fields]]
key = "source"
value.static = "iggy"

[[transforms.add_fields.fields]]
key = "processed_at"
value.computed = "timestamp_millis"

[[transforms.add_fields.fields]]
key = "trace_id"
value.computed = "uuid_v7"

[transforms.delete_fields]
enabled = true
fields = ["internal_field"]