Introduction

crossing streams

mq-bridge moves data reliably between message brokers, databases, files, and HTTP — with batching, retries, dead-letter queues, deduplication, and change data capture built in. One Rust engine, one config format, two ways to use it:

Almost everything in this book applies to both: connectors, middleware, delivery guarantees, and tuning share the same settings. A CLI URL’s query parameters are the same fields you write in a library’s YAML config.

Supported integrations include Kafka, RabbitMQ (AMQP), NATS, AWS SQS / SNS, MQTT, IBM MQ, HTTP, WebSocket, gRPC, ZeroMQ, MongoDB, Redis Streams, ClickHouse, Postgres CDC, SQLx (MySQL, MariaDB, PostgreSQL, SQLite), cloud object storage, and filesystem endpoints — plus Pulsar, Meilisearch, and Redpanda Connect through plugins.

A quick taste

In code — the same route runs in Rust, Python, or Node.js:

from mq_bridge import Route
Route.from_file("routes.yaml", "kafka_to_nats").run()

Without code — mqb copy moves data between databases, queues, and files in a single line of bash:

mqb copy \
  --from 'postgres://user:pass@localhost/db?table=src&sslmode=disable' \
  --to   'file://out.jsonl?format=raw' \
  --drain

The scheme selects the endpoint and query parameters configure it, so any source→sink pair is just one URL each. And it’s quick: in benchmarks a 1,000,000-row Postgres → JSONL job sustained 413,907 rows/s at ~55 MiB peak RSS (mq-bridge 0.4.20) — see Performance tuning for the full table and the conditions each row was measured under.

Philosophy

The project has one main bias: move data reliably without forcing the rest of the application to care too much about the transport. Kafka offsets, RabbitMQ nacks, HTTP responses, MongoDB polling, WebSocket frames, and file rows are all different in real life, but route code should still be able to receive a batch, process it, publish it, and commit it.

• Fast by default. Every endpoint is optimized around batch-shaped APIs, and the headless surfaces ship tuned for throughput: the copy CLI and the MCP server default to batch_size: 1024 and concurrency: 4. The library/config primitive defaults to batch_size: 512, concurrency: 1: batches fill opportunistically — a route takes whatever is already queued rather than waiting — so throughput comes for free while parallelism stays a deliberate per-route choice.
• Reliability is built in, not bolted on. Retries, dead-letter queues, deduplication, rate limiting, and cookie/session persistence wrap any endpoint. Ack/nack behaviour and retry/DLQ handling were designed to work with batching, including commit sequencing for cumulative-ack brokers.
• Not a framework. It is not a domain framework, an actor runtime, or a full stream processor. It cares about transport, routing, and delivery behaviour, not about prescribing your domain model.

Where to go next

• Writing code? Start with the library quick start, then Embed the library.
• Moving data without code? Start with Installation and the copy quick start.
• Want the end-to-end walkthroughs? See the Tutorials.
• Looking for a specific task? The Cookbook has short recipes.
• Need exact fields and defaults? The Reference is authoritative.
• Running it in production? See Operations, especially the Performance tuning page.
• Driving it from an AI agent? The same binary is an MCP server — the rows move without entering the model’s context.

Library vs. app

mq-bridge is the engine/library; mq-bridge-app is the application built on it. This book covers both; the engine’s Rust API reference lives on docs.rs, and the language bindings API covers Python and Node.js.

Install

mq-bridge ships in three forms that share one engine and one config format: the library (mq-bridge for Rust, Python, and Node.js), and the two forms of the application mq-bridge-app — the CLI / server and the desktop app (UI). Pick the install path for the form you need. To compile any of them yourself, see Building from source.

Library in one line: cargo add mq-bridge · pip install mq-bridge · npm install mq-bridge — details in Library.

• CLI / server — Homebrew, Conda, cargo binstall, cargo install, Docker
• Desktop app (UI) — Homebrew cask, or a prebuilt bundle
• Library — Rust, Python, Node.js

CLI / server

The CLI is a single headless binary, installed under two names: mq-bridge-app and the short mqb, which is what the docs use and what you’ll normally type. mqb is a small launcher that hands straight over to the real binary, so the two are interchangeable — existing scripts and MCP registrations keep working untouched.

Homebrew (macOS, Linux) — recommended

brew install marcomq/tap/mq-bridge-app
brew upgrade mq-bridge-app     # later, to update

Prebuilt bottles cover Apple Silicon macOS and x86_64 Linux. Homebrew refreshes the tap only on brew update, so if brew upgrade doesn’t pick up a new release yet, run brew update first.

Conda (Windows, macOS, Linux)

Conda is the package-manager install path for Windows and is also available on macOS and Linux. It installs both mq-bridge-app and mqb:

conda install -c marcomq -c conda-forge mq-bridge-app

The marcomq channel carries mq-bridge itself; conda-forge supplies its build and runtime dependencies.

cargo binstall — prebuilt binary

Downloads the prebuilt CLI from Releases instead of compiling from source, so it installs in seconds:

cargo binstall mq-bridge-app

Prebuilt binaries are available for x86_64 Linux, Apple Silicon macOS, and x86_64 Windows. (cargo-binstall is a drop-in cargo install replacement.)

cargo install — from source

Requires a Rust toolchain and compiles all supported endpoint client libraries (except IBM MQ), so it may take a while:

cargo install mq-bridge-app

For IBM MQ, install the client library first and build with --features=ibm-mq.

Docker

The CLI is published as a multi-arch image (amd64 + arm64):

docker run --rm --name mq-bridge -p 9091:9091 ghcr.io/marcomq/mq-bridge-app:latest

To read+tail from input.log and forward its content, mount the working directory at /app and seed the config from one of the templates baked into the image at /config:

touch input.log
docker run --rm --name mq-bridge -p 9091:9091 -v "$(pwd)":/app \
  ghcr.io/marcomq/mq-bridge-app:latest --ui --init-config=/config/file-to-http.yml

Note

The default latest image is a plain multi-arch image for amd64 and arm64. IBM MQ support is published separately as the latest-ibm-mq and ibm-mq tags on amd64 only, since there is no redistributable IBM MQ client library for arm64 yet. Start that image in emulation mode with --platform=linux/amd64, or build mq-bridge-app yourself with cargo build --release --features=ibm-mq.

Desktop app (UI)

The desktop app is a Tauri bundle of the full messaging workbench — the same UI the CLI serves in a browser, only packaged as a native app.

Homebrew cask (macOS, Apple Silicon)

brew install --cask marcomq/tap/mq-bridge

Homebrew quarantines cask apps by default and the desktop binaries are not notarized yet, so macOS blocks the app on first launch. See the note below for how to open it (via System Settings or Terminal).

Prebuilt bundles (macOS, Windows, Linux)

Bundles for every platform are attached to each release on the GitHub Releases page.

• macOS — download the .dmg / .app bundle.
• Windows — download the installer or standalone executable.
• Linux — download the bundle that suits your distribution: AppImage, .deb, .rpm, or the unpacked archive.

Note

The desktop binaries are currently not notarized, so on macOS Gatekeeper blocks the app on first launch — it’s reported as “damaged” or “cannot be opened because the developer cannot be verified”. This applies to both the Homebrew cask and a downloaded bundle. There are two ways to open it:

Option 1 — System Settings (no Terminal). Try to open the app once so the block is triggered, then go to System Settings → Privacy & Security, scroll to the Security section, and click “Open Anyway” next to the mq-bridge message. Authenticate, then launch the app again.

Option 2 — Terminal. Remove the quarantine attribute directly:

# app in /Applications (where the cask and the .dmg install it)
sudo xattr -rd com.apple.quarantine /Applications/mq-bridge.app
# app in a user-owned directory (e.g. ~/Downloads) — no sudo needed
xattr -rd com.apple.quarantine ~/Downloads/mq-bridge.app

If macOS says the app is “damaged”, the “Open Anyway” button may not appear — use the Terminal method in that case.

Library

Embed the core engine in your own code — produce or consume messages with a unified API, one config format across all three bindings:

• Rust — mq-bridge (cargo add mq-bridge)
• Python — pip install mq-bridge
• Node.js — npm install mq-bridge

The core of the library are the MessageConsumer and MessagePublisher traits, found in mq_bridge::traits.

Build from source

Building the CLI/server, the desktop (Tauri) app, or the Docker image from source is covered in BUILD.md. For IBM MQ specifically, see the IBM MQ Setup Guide.

Quick start: library

The engine is a library first. The same route config runs embedded in your Rust, Python, or Node.js service and in mqb — design it once, load it anywhere.

Install

The Rust crate enables connectors through Cargo features (kafka, nats, mongodb, …, or full for all of them); yaml adds YAML config files (JSON works without it). The Python and Node.js packages ship with all of them built in.

Describe a route

A route moves messages from one input endpoint to one output endpoint. This one reads a Kafka topic and publishes to NATS, retrying failed sends:

# routes.yaml
kafka_to_nats:
  input:
    kafka: { url: "localhost:9092", topic: "orders", group_id: "bridge" }
  output:
    middlewares:
      - retry: { max_attempts: 5 }
    nats: { url: "nats://localhost:4222", subject: "orders.processed" }

Run it

Python

from mq_bridge import Route

route = Route.from_file("routes.yaml", "kafka_to_nats")
route.run()  # blocks; use route.start() to keep running your own code

Node.js

import { Route } from "mq-bridge";

Route.fromFile("routes.yaml", "kafka_to_nats").start();

Rust

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    mq_bridge::deploy_file("routes.yaml").await?; // every route in the file, in the background
    tokio::signal::ctrl_c().await?;
    Ok(())
}

Add business logic

Attach a handler to transform, filter, or answer messages in between. Returning a message publishes it to the output; returning nothing acknowledges and drops it.

def enrich(message):
    order = message.json()
    order["processed"] = True
    return message.with_json(order)

Route.from_file("routes.yaml", "kafka_to_nats").with_handler(enrich).run()

use mq_bridge::{CanonicalMessage, Handled, Route};

Route::from_file("routes.yaml", "kafka_to_nats")?
    .with_handler(|message: CanonicalMessage| async move {
        let mut order: serde_json::Value = message.parse().map_err(anyhow::Error::from)?;
        order["processed"] = true.into();
        Ok(Handled::Publish(CanonicalMessage::from(order)))
    })
    .deploy("kafka_to_nats")
    .await?;

Handlers, typed dispatch on the kind field, and request/reply are covered in Embed the library and Core concepts.

Same settings everywhere

The YAML fields above are the same fields the CLI takes as URL query parameters, so everything in the connector pages and the URL parameter reference applies to library configs too:

mqb copy 'kafka://localhost:9092?topic=orders&group_id=bridge' \
         'nats://localhost:4222?subject=orders.processed'

The full config grammar — routes, middleware lists, structural endpoints — is in Configuration grammar and Middleware & structural endpoints.

Quick Start

mqb copy SOURCE TARGET moves data between two endpoints described as URIs. The scheme picks the connector; ?query=params configure it. The existing --from SOURCE --to TARGET form is equivalent and remains supported.

mqb copy \
  'postgres://localhost/app?table=users' \
  'clickhouse://localhost:8123?table=users&database=analytics'

This copies every row currently in app.users (PostgreSQL) into analytics.users (ClickHouse). No config file, no UI — just a source and a destination. Add --drain to exit once the source is empty instead of running as a continuous bridge (see Continuous vs. one-shot below).

The common copy controls are deliberately small:

mqb copy SOURCE TARGET [--filter EXPR] [--resume | --no-resume] [--drain] [--wait SECS]

--filter evaluates a readable expression against each top-level JSON payload, for example amount > 100 or status == "paid". A false result intentionally drops and acknowledges the message; malformed JSON and invalid expressions are errors, while a field that is absent or not a scalar counts as no match and is warned about once. It is an in-process filter and is not translated into a database query. Filtering into cloud object storage also changes how the objects are named — see Filtering.

--resume asks the source to use its native durable position and fails before the route starts when that is not safe. The generated state identity includes the credential-redacted source, destination, and filter, so changing pipeline semantics starts a new checkpoint while rotating a password does not. --no-resume explicitly ignores optional cursor/checkpoint state and its diagnostics for cursor-based sources; it is a no-op for native queue/CDC offsets.

The examples below are complete, working commands. Each links to the full connector page for that endpoint, which lists every available option; the generated URL reference is the authoritative source for every parameter’s type, default, and description.

PostgreSQL → ClickHouse

mqb copy --drain \
  --from 'postgres://user:pass@localhost/app?table=orders' \
  --to 'clickhouse://localhost:8123?table=orders&database=analytics'

Reads all rows from the orders table and bulk-inserts them into ClickHouse’s HTTP interface. For a resumable non-destructive scan, add &cursor_column=id on --from and pass --resume; the CLI supplies the stable cursor id and the SQL source stores the checkpoint in its own database. An explicit cursor_id or checkpoint_store in the URI still takes precedence. See PostgreSQL and ClickHouse.

Filtered, resumable Kafka copy

mqb copy \
  'kafka://localhost:9092?topic=orders' \
  'postgres://localhost/app?table=orders' \
  --filter 'status == "paid"' \
  --resume

The generated Kafka consumer group is stable for this source, destination, and filter. Kafka offsets advance only after the destination succeeds, or after a message is intentionally filtered out.

PostgreSQL CDC → PostgreSQL

mqb copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to 'postgres://user:pass@otherhost/replica?table=orders&auto_create_table=true'

Streams inserts/updates/deletes from a PostgreSQL logical-replication publication into another PostgreSQL table, continuously (CDC is a change stream, so this command doesn’t drain — run it as a long-lived process). See PostgreSQL CDC.

MQTT → Kafka

mqb copy \
  --from mqtt://broker.local:1883?topic=sensors/+/temperature \
  --to kafka://kafka.local:9092?topic=sensor-readings

Subscribes to an MQTT topic (wildcards supported) and republishes every message to a Kafka topic, continuously. See MQTT and Kafka.

RabbitMQ → HTTP

mqb copy \
  --from rabbitmq://guest:guest@localhost:5672/%2f?queue=orders \
  --to http://internal-api.local/ingest?method=POST

Consumes messages from a RabbitMQ queue and POSTs each one to an HTTP endpoint, continuously. See RabbitMQ and HTTP.

File (CSV) → MongoDB

mqb copy --drain \
  --from file:///data/customers.csv?format=csv \
  --to 'mongodb://localhost?database=app&collection=customers'

Reads a CSV file (first row = header) and inserts one document per row into a MongoDB collection, then exits since the source is a finite file. See File and MongoDB.

Continuous vs. one-shot

Without --drain, copy runs as a continuous bridge until Ctrl-C — the right mode for message brokers (MQTT, Kafka, RabbitMQ) and CDC sources, which never “end”. With --drain, copy exits once the source yields an empty batch — the right mode for finite sources (a file, or a full-table read from a database). --concurrency and --batch-size tune throughput on both modes.

--wait SECS covers the case in between: a source another process is still filling. It is long polling over a drain — it implies --drain, waits up to SECS for the first message, and returns as soon as one arrives, so the value is a ceiling rather than a delay. Without it a drain that starts a moment too early sees an empty source and exits reporting nothing to do.

# Blocks until another process writes to the spool, then drains it and exits.
mqb copy --from spool:///var/spool/incoming --to postgres://... --wait 300

--wait waits by retrying the drain, so it lets go of the source between attempts. That is invisible for a file or a table, but a source that allows only one reader at a time — a dir_spool queue — is claimable by someone else in the gap. Where that matters, use agent-listen, which holds its source open for the whole wait.

Each retry is a fresh connection, roughly four a second. That is free for a spool or a file, but a long --wait against a broker or a database reconnects thousands of times; prefer a short wait there, or a continuous bridge.

Escape hatch: driver options and full connection strings

Any query parameter that isn’t a recognised config field (e.g. sslmode, replicaSet) is left on the connection URL untouched, so driver-specific options just work — including object-typed fields like tls, which can never be set from a single scalar query param and so always stays on the URL (e.g. mongodb://host/?tls=true&database=appdb passes tls=true straight through to the MongoDB driver). If you already have a complete connection string (copied from elsewhere, or one whose own options would otherwise be mis-parsed as config), skip decomposition entirely and pass it verbatim with ?url=<url-encoded string>:

mqb copy \
  --from 'mongodb://_/?url=mongodb%3A%2F%2Fuser%3Apass%40host%2Fdb%3Ftls%3Dtrue&database=db&collection=orders' \
  --to null:

See the generated reference for each connector’s recognised field names.

Core concepts

Four terms recur throughout this book.

Route

A route is a named data pipeline that defines a flow from one input to one output. It carries optional middlewares on either side, a handler, and the tuning knobs batch_size, concurrency, and commit_concurrency_limit. In config, the top-level keys are route names:

orders_bridge:                 # <- route name
  input:  { kafka: { topic: "orders", url: "localhost:9092" } }
  output: { nats:  { subject: "orders.processed", url: "nats://localhost:4222" } }

Endpoint

An endpoint is a source or sink for messages, written as a single-key object naming the connector. Transport endpoints talk to a broker or store (kafka, nats, sqlx, mongodb, file, memory, …); structural endpoints (fanout, switch, response, ref, null, …) compose other endpoints or shape routing. The same endpoint type can act as consumer or publisher depending on which side of the route it is on. See the full catalog in Endpoints (connectors).

Middleware

Middleware intercepts and processes messages around an endpoint — retries, dead-letter queues, transformation, deduplication, rate limiting, encryption, metrics. It attaches as a middlewares: list on the input, output, or both. Ordering matters: on an output the last entry is the outermost layer, so dlq goes last. See Middleware.

Handler

A handler is a programmatic component for business logic, used when you embed the library (config-only routes forward messages without one). There are three kinds:

• CommandHandler — a 1-to-1 or 1-to-0 transformation. It takes a message and can return a new message to publish or return as a response.
• EventHandler — a terminal 1-to-N handler for event consumption. It reads messages without removing them for other event handlers, and should not return a response.
• TypeHandler — a strongly-typed handler. It dispatches on the kind metadata field and deserializes payloads into concrete Rust types, so handlers don’t repeat parsing code.

let typed_handler = TypeHandler::new()
    .add("create_user", |cmd: CreateUser| async move {
        // handle create_user; Ok(()) maps to Handled::Ack
    })
    .add("delete_user", |cmd: DeleteUser| async move {
        // handle delete_user
    });
let route = Route::new(input, output).with_handler(typed_handler);

Handlers return a Handled value: Handled::Ack (consume), Handled::Publish(msg) (forward msg down the publisher chain), and so on. Message selection for TypeHandler is driven by the kind metadata field, which msg!(&value, "kind") sets. See the Embed the library tutorial and Language bindings API.

CanonicalMessage

Every handler works with a CanonicalMessage — the unified message type that decouples your code from any specific transport. It carries a payload (bytes), string metadata headers, and a message_id (a UUIDv7 when omitted). Conventional metadata keys: kind (message type / typed routing), correlation_id and reply_to (request/reply). Keys starting with mqb.src. are reserved for provenance and stripped on input.

Delivery model

Endpoints default to a consumer (queue) pattern; subscriber (pub/sub) behaviour is opt-in per backend. Delivery is at-least-once where a durable position or lease exists — so for ETL, pair it with an idempotent write at the sink for effective exactly-once. Ack/nack and retry/DLQ handling are designed to work with batching; see Learn the architecture.

The three ways to run it

mq-bridge is one engine with one config format, exposed three ways: the library and the two forms of mq-bridge-app. Build and test a route in the UI, export the JSON/YAML, then run that config in a config-mode service or from library code. The copy CLI takes the same settings, but expressed as endpoint URIs and flags rather than a config file.

See Installation for every install method and platform.

Desktop app (UI)

The desktop app is a Tauri bundle of the full messaging workbench: manage publishers/consumers/routes, run request/response traffic (like Postman for REST), inspect message history, and import Postman/OpenAPI/AsyncAPI definitions. It is the same UI the CLI serves in a browser — only the packaging differs.

The UI is generated dynamically from the Rust configuration structures: the backend uses schemars to produce a JSON Schema for the AppConfig struct (exposed at /schema.json, also mqb --schema <path>), and the frontend renders a complete config form from that schema. So when a new middleware or option is added to the engine, the schema updates automatically and the UI reflects it with no frontend change.

CLI / server

The CLI (mqb) is a headless binary that runs in three modes. They share the same engine and config format but differ in how you drive them — and only config mode serves the browser UI:

# Config mode: run a long-lived bridge from a file
mqb --config config.yml

# Seed config.yml from a template on first run only
mqb --config config.yml --init-config dev/config/file-to-http.yml

# Start empty, then open the UI to build your config
mqb --ui

In config mode the CLI can also serve the browser UI (the same UI as the desktop app) on the configured port. It is never opened implicitly: set ui_addr in the config, pass --ui, or answer the prompt — an unattended run without --ui stays headless. See the CLI reference for the full rule and every flag, and Configuration grammar for the config format.

What a config file has to contain

Everything a config file defines is started at boot, in both headless and UI runs. A route can be held back with enabled: false; the UI starts it on demand.

The full form is a routes: map alongside the application settings, which is what the UI writes out. Routes may also be written at the top level, so a file that is nothing but a route needs neither the wrapper nor a name — these three are the same bridge:

# 1. the full form
routes:
  proxy:
    input:  { http: { url: "0.0.0.0:8443" } }
    output: { http: { url: "https://upstream.internal/" } }

# 2. named at the top level, as in the Configuration grammar
proxy:
  input:  { http: { url: "0.0.0.0:8443" } }
  output: { http: { url: "https://upstream.internal/" } }

# 3. a single unnamed route — it runs as `route`
input:  { http: { url: "0.0.0.0:8443" } }
output: { http: { url: "https://upstream.internal/" } }

Application settings (ui_addr, metrics_addr, log_level, plugins, …) stay at the top level in every form. In form 3 every other top-level key belongs to the route, so route options such as batch_size and exit_on_empty go there — and a misspelled one is reported by name rather than ignored.

The configuration-first workflow

The point of one shared config format is that you can test connections and dial in a route in the UI, export the JSON/YAML, then run that exact config unchanged — as a config-mode service or loaded from library code. A known-good route shape from the UI drops straight into production.

copy is the exception: it takes no config file, so a route you built in the UI has to be mapped by hand onto --from / --to endpoint URIs and route flags. The settings are the same, only the way you pass them differs.

Library

Beyond running standalone, the core engine is available as a library so you can produce or consume messages with a unified API — no broker-specific SDK, one config format across all three bindings:

• Rust — mq-bridge (cargo add mq-bridge)
• Python — pip install mq-bridge
• Node.js — npm install mq-bridge

The core of the library are the MessageConsumer and MessagePublisher traits in mq_bridge::traits. See the Language bindings API and the Embed the library tutorial.

How the UI differs from API clients

The UI overlaps with API clients like Postman, Bruno, and Insomnia, but its centre of gravity is different: it is designed around message bridging, runtime operation, and long-lived route management rather than just request composition. It adds broker pub/sub workflows, long-lived consumers/routes, bridging traffic between protocols, hex-level payload debugging, replay, local-first git-friendly config, and encrypted config — while leaving scripting and complex request workflows to the dedicated API clients. Use an API client when your main job is crafting and sharing API requests; use mq-bridge-app when you need to connect systems, move messages between protocols, inspect live traffic, and manage bridge-style runtime configuration.

Embed the library (Rust / Python / Node)

The core engine is a Rust library, and the same engine ships as native bindings for Python and Node.js. The Tokio runtime, broker I/O, routing, and batching all stay in Rust; the binding is a thin layer for handlers and configuration. Behaviour and reliability match the Rust engine regardless of which language calls in.

This page picks up where the library quick start ends (install lines are there). The exact constructor and type names are in the Language bindings API.

Design the route as config

The natural way to embed the library is to design the route as config, then load that exact config from your code: hand-write the YAML, or build and test the route in the desktop UI and export it. The running route behaves identically to the mqb --config form, and every field in the connector pages and the middleware reference applies unchanged.

# routes.yaml
routes:
  orders:
    input:
      kafka: { url: "localhost:9092", topic: "orders", group_id: "orders-service" }
    output:
      middlewares:
        - retry: { max_attempts: 5 }
      mongodb: { url: "mongodb://localhost:27017", database: "shop", collection: "orders" }

Handlers

A handler runs your business logic between input and output. Return a message to publish it, return nothing to acknowledge and drop the input, or raise/throw to fail it (retryable or not).

Python — handlers are synchronous; run() blocks, start() returns:

from mq_bridge import Route, NonRetryableError

def handle(message):
    order = message.json()
    if order.get("amount", 0) <= 0:
        raise NonRetryableError("invalid amount")
    return message.with_json({**order, "checked": True})

route = Route.from_file("routes.yaml", "orders").with_handler(handle)
route.start()   # or route.run() to block, or `with route:` to scope it

Node.js

import { Message, Route } from "mq-bridge";

const route = Route.fromFile("routes.yaml", "orders");
route.withHandler(async (message) => {
  const order = message.json();
  return Message.fromJson({ ...order, checked: true });
});
route.start();

Typed dispatch — add_handler / addHandler routes on the message’s kind metadata field and hands you decoded JSON:

route.add_handler("order.created", lambda order: {"seen": order["order_id"]})

Publishing into a route

A Publisher sends to any endpoint — for example the route’s input, to feed it from your own code:

from mq_bridge import Publisher

publisher = Publisher.from_config({"kafka": {"url": "localhost:9092", "topic": "orders"}})
publisher.send_json({"order_id": 42}, {"kind": "order.created"})

Node.js has the same shape with Publisher.fromConfig(...) and await publisher.sendJson(...).

Rust

The Rust crate exposes the full engine: Route, Endpoint, Publisher, the handler types, and the MessageConsumer / MessagePublisher traits. A config file loads into mq_bridge::models::Config (a map of route name → Route), or build routes in code:

use mq_bridge::{models::Endpoint, CanonicalMessage, Handled, Route};

let handler = |mut msg: CanonicalMessage| async move {
    msg.set_payload_str(format!("handled {}", msg.get_payload_str()));
    Ok(Handled::Publish(msg))
};
let route = Route::new(Endpoint::new_memory("in", 200), Endpoint::new_memory("out", 200))
    .with_handler(handler);
route.deploy("my_route").await?;

Typed handlers and sending commands

TypeHandler picks a handler by the message’s kind metadata field and deserializes the payload into your type:

use mq_bridge::type_handler::TypeHandler;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct CreateUser { id: u32, username: String }

let typed_handler = TypeHandler::new()
    .add("create_user", |cmd: CreateUser| async move {
        println!("create_user {} {}", cmd.id, cmd.username);
        // () maps to Handled::Ack
    });
let route = Route::new(input, output).with_handler(typed_handler);

// Publish into the route's input; msg! sets the `kind` field.
let input_publisher = Publisher::new(route.input.clone()).await?;
input_publisher.send(msg!(&CreateUser { id: 1, username: "test".into() }, "create_user")).await?;

CQRS-style flows

Routes and typed handlers can act as a command bus and an event bus without becoming a domain framework. A command route handles the write side and emits an event; downstream routes subscribe to those events to update read models:

// Write side: handle the command, emit an event
let command_bus = TypeHandler::new()
    .add("submit_order", |cmd: SubmitOrder| async move {
        let evt = OrderSubmitted { id: cmd.id };
        Ok(Handled::Publish(msg!(evt, "order_submitted")))
    });

// Read side: project the event
let projection = TypeHandler::new()
    .add("order_submitted", |evt: OrderSubmitted| async move {
        // update read database / cache
        Ok(())
    });

See also

• Language bindings API — the reference for this material.
• Writing endpoints & middleware — plugging your own endpoint or middleware into the engine from Rust, Python, or Node.
• Core concepts and Learn the architecture — the handler and message model.
• The three ways to run it — library vs CLI-server vs desktop.

Request / reply

mq-bridge supports request-response patterns for interactive services such as web APIs. A client sends a request and waits for the matching response, while the bridge keeps the correlation details away from your handler.

The response output is the most direct option and the safest one under concurrency: the response stays in the same execution context as the request, so concurrent requests do not share a reply queue or race on correlation IDs.

How it works

1. An input endpoint that supports request-reply (like http) receives a request.
2. The message passes through the route’s processing chain — this is where you attach a handler to process the request and generate a response payload.
3. The final message is sent to the output.
4. If the output is response: {}, the bridge sends the message back to the original input source, which delivers it as the reply (e.g. the HTTP response body).

Not every backend can do request-reply. It is only supported/tested for endpoints that natively support or emulate it. SQLx, files, AWS, IBM MQ, and Sled do not support request-reply. Check the Request-Reply column in Consumer vs. subscriber, and nack support.

Example: MongoDB request/response

A sender writes a request document to MongoDB and waits for a reply. The bridge reads the document, runs the handler, and writes the result back to the reply collection:

mongo_responder:
  input:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "app_db"
      collection: "requests"
  output:
    # 'response' sends the processed message back to the reply collection
    # (whatever reply_to the sender set).
    response: {}

Attaching the handler (Rust)

The response output only echoes what the pipeline produced — you supply the logic by attaching a handler to the route:

use mq_bridge::models::Handled;
use mq_bridge::CanonicalMessage;

let handler = |mut msg: CanonicalMessage| async move {
    let request_body = String::from_utf8_lossy(&msg.payload);
    let response_body = format!("Handled response for: {}", request_body);
    msg.payload = response_body.into();
    Ok(Handled::Publish(msg))
};
// load the route from YAML, then attach `handler` to its output and run it.

See Embed the library for the full load-and-run scaffolding, and Core concepts for the handler model (CommandHandler / EventHandler / TypeHandler).

Caveats

• If the input does not support responses (e.g. File, SQLx), a message sent to response is dropped.
• Configure timeouts on the requester side: the bridge’s processing time adds latency.
• Middleware that drops metadata (like correlation_id) can break the response chain.
• MongoDB request/reply is emulated by polling for the reply; set its timeouts deliberately.

See also

• request / response structural endpoints — the reference definitions.

Language bindings API

The core engine is a Rust library, but the same engine ships as native bindings for Python and Node.js. The Tokio runtime, broker I/O, routing, and batching all stay in Rust; the binding is a thin layer for handlers and configuration.

The core of the library are the MessageConsumer and MessagePublisher traits, found in mq_bridge::traits.

Config loaders

Constructor names are kept aligned across languages, so a config loader reads the same in either binding (Python uses snake_case, Node uses camelCase):

The name argument is optional in both: pass it to select one entry from a routes:/publishers: document, or omit it to treat the config as a single bare route/endpoint body.

This is the natural companion to the configuration-first workflow: design and test a route in the UI, export the JSON/YAML, then load that exact config from your code. See the Embed the library tutorial for full examples in each language.

Extending from a binding

Custom endpoints and middleware are not Rust-only. Both bindings expose register_endpoint / register_middleware (registerEndpoint / registerMiddleware in Node), with the same batch, ack and request-reply semantics as a Rust CustomEndpointFactory, and both can load a compiled plugin — mq_bridge.load_endpoint_plugin(path) in Python, loadEndpointPlugin(path) in Node — so one Rust implementation serves every language. Registration is process-global, keyed by name, and must happen before a route that names it starts; a duplicate name is an error rather than a silent replacement.

Full examples in Custom endpoints and Native plugins.

Rust API surface

The Rust crate exposes the full engine. The main types:

• Route::new(input, output) — a pipeline from one endpoint to one endpoint, with .with_batch_size(n), .with_handler(h), .add_handler(kind, f), .deploy(name) / .run().
• Endpoint — protocol adapters (Endpoint::new_memory, Endpoint::null, …) plus the serde-configured variants.
• Publisher::new(endpoint) — publish into a route’s input or any endpoint.
• Handlers: CommandHandler (1-to-1/1-to-0), EventHandler (terminal 1-to-N), and TypeHandler (dispatches on the kind metadata field, deserializing payloads).
• CanonicalMessage — the unified message type all handlers work with; msg!(&value, "kind") builds one with a kind.

See Core concepts and Learn the architecture for the handler model, and the Rust docs on docs.rs for the full API.

Notes

• The Python binding holds up under load on the third-party http-arena.com requests-per-second HTTP benchmark (a live leaderboard — rankings shift over time).
• A binding is a thin layer: routing, batching, and broker I/O stay in Rust regardless of which language calls in, so behaviour and reliability match the Rust engine.

The desktop / web UI

mq-bridge-app ships a visual workbench for building, testing, and running message routes. It is the same UI whether you launch the desktop app (a Tauri bundle) or let the CLI serve it in a browser — only the packaging differs. Think of it as Postman for message bridging: build and test a route, export the JSON/YAML, then run that exact config unchanged wherever you deploy.

For how the UI is built (schema-driven forms, the engine acting as its own web server), see How the app is built.

Layout

The window is split into three top-level tabs, a sidebar listing your endpoints, and a top bar:

• Publishers (↑) — endpoints you send to (build/test requests, run request/response traffic).
• Consumers (↓) — endpoints you receive from (live message log, payload inspection).
• App Config (⚙) — application-wide settings, config security, and environment variables.

The top bar shows a runtime status indicator (idle / active consumer(s)), a theme switcher (light / auto / dark), and Save. The sidebar filters endpoints and offers one-click import from Postman, OpenAPI, AsyncAPI, and existing mq-bridge configs.

Publishers

Select or add a publisher on the left, then edit it on the right. The header row holds the endpoint type (HTTP, Kafka, NATS, …), method, and URL, with a Send button to fire a request. Below it are per-endpoint tabs:

• Definition — name, endpoint type and connection settings (with Show advanced options), and a Middlewares list you can add to / remove from (metrics, retries, transforms, …).
• Body / Headers — the outgoing payload (with AUTO / TEXT / JSON / XML / HEX views) and header rows.
• History / Presets — previous sends and saved request presets.

After a Send, the response panel shows status, timing, request/response headers, and the response body (including a HEX view), with copy / json / curl shortcuts. Copy to Consumer, Clone, {} (view raw JSON), and Delete act on the selected endpoint.

Consumers

Consumers receive messages live. The header shows the connection state (Connected), a Capture messages toggle, a Keep last N limit, and Clear / Stop controls. Tabs:

• Definition — connection settings, same shape as a publisher.
• Output — the consumer’s response/output configuration.
• Messages — a live log (time + payload preview). Click a row to inspect it below: Message Headers, Message Body, and any Response Headers / Body, each with AUTO / TEXT / JSON / XML / HEX views plus copy/export.

App Config

Application-wide settings rendered directly from the AppConfig schema:

• AppConfig — default tab, log level, logger, metrics address, and UI address.
• Config Security — the storage mode (unencrypted, balanced, sensitive, durable, …) controlling how secrets and cached message history are stored. See Encryption at rest and Secrets.
• Environment Variables — key/value pairs available for interpolation in endpoint URLs and settings.

Export, Import, Reset, and {?} JSON operate on the whole configuration.

The configuration-first workflow

The point of one shared config format is that you test connections and dial in a route in the UI, export the JSON/YAML, then run that exact config unchanged — as a copy command, a config-mode service, or loaded from library code. A known-good route shape from the UI drops straight into production.

How the UI differs from API clients

The UI overlaps with API clients like Postman, Bruno, and Insomnia, but its centre of gravity is different: it is built around message bridging, runtime operation, and long-lived route management rather than one-off request composition. It adds broker pub/sub workflows, long-lived consumers/routes, protocol-to-protocol bridging, hex-level payload debugging, replay, local-first git-friendly config, and encrypted config — while leaving scripting and complex request chaining to the dedicated API clients.

Status. The UI/Tauri layer was prototyped quickly and does not mirror the mq-bridge / core/CLI standards — treat it as a working demo, not a reference implementation, and test before relying on it in production.

CLI commands

mqb is a single headless binary with three modes: config mode (the default — run a long-lived bridge, optionally serving the browser UI), the copy subcommand (an ad-hoc one-route job), and the mcp subcommand (expose the bridge as MCP tools). The checkpoint subcommand inspects and edits a source’s resume position. The status subcommand shows what every mqb process on the machine is running.

mqb [OPTIONS]                          # config mode
mqb copy SOURCE TARGET [COPY OPTIONS]  # one-route ad-hoc job
mqb mcp  [MCP OPTIONS]                 # MCP server
mqb checkpoint show|reset|set ...      # inspect or edit a resume position
mqb status [--no-watch] [--json]       # what is running on this machine

Config mode (default)

Run with no subcommand to load a config and run a bridge; with no config at all it starts empty and offers to serve the UI so you can build one interactively.

mqb --config config.yml
mqb --config config.yml --init-config dev/config/file-to-http.yml
mqb --ui                           # start empty, define config.yml in the UI

Config is hierarchical (files + environment variables) — see Configuration grammar.

Starting the web UI

The UI is a control surface, so its port is never opened implicitly. What happens in config mode depends on where the address comes from:

The last row is the important one: a run started by a script or a service unit never puts the UI on the network by accident. Nothing about the bridge itself is gated — configured routes run either way.

The UI has no login. --ui therefore listens on loopback, and a start with an address that other machines can reach logs a warning. Two more rules apply to every UI:

• A UI on a loopback address answers only requests whose Host header names localhost or a loopback address, which stops a web page from reaching it through DNS rebinding. Behind a reverse proxy or a port forward that sends another name, list the names in MQB_UI_ALLOWED_HOSTS (comma-separated).
• GET /config shows each secret as a reference to the key it is stored under, such as ${MQB__PUBLISHERS__<id>__HTTP__BASIC_AUTH__1}, except in the unencrypted and temporary_messages storage modes. A URL keeps everything but its password. A save replaces each reference by the stored value, so a cloned or renamed endpoint keeps its secrets; a reference to a key that is not stored is refused with 400.

MQB_UI_DEFAULT_ADDR changes the address that --ui and the prompt use. It does not start the UI.

In a container the calculation is reversed, because nothing is reachable until you publish it. The Docker image’s CMD therefore asks for the UI on your behalf — see Ports in containers.

The metrics endpoint

Metrics are always collected, and always available at /metrics on the web UI when it runs. Separately, config mode serves a standalone Prometheus endpoint on 127.0.0.1:9090 by default.

It defaults to loopback rather than 0.0.0.0 because, while the endpoint is read-only, it still describes the routes and endpoint types in use — a bare run on a workstation shouldn’t publish that to the local network. Scraping from another host is an explicit choice:

mqb --config config.yml --metrics-addr 0.0.0.0:9090   # scrapeable
mqb --config config.yml --no-metrics                  # no separate port at all

metrics_addr in the config does the same thing; the flag overrides it.

copy — ad-hoc one-route job

Builds a single route from two endpoint URIs and runs it headlessly (no web UI). The scheme selects the endpoint and query parameters set its config.

# DB → DB, drain the source table then exit (exit code 0 on success)
mqb copy \
  'postgres://user:pass@localhost/db?table=src' \
  'postgres://user:pass@localhost/db?table=dst' \
  --drain

# Queue → DB as a continuous bridge (runs until Ctrl-C; omit --drain)
mqb copy \
  --from 'nats://localhost:4222?subject=orders' \
  --to   'postgres://user:pass@localhost/db?table=orders'

copy prints the copied … summary on stdout and its log lines on stderr, so a script can read the summary alone. Set MQB_LOG_STDOUT=1 to get the log lines on stdout as before 0.4.18. It exits 1 when it could not deliver every row it read, and when rows could not be decrypted. Rows sent to a dlq are reported apart from the copied ones (copied 0 rows, dead-lettered 1_000) and leave the exit code at 0.

Two copy processes can listen on the same HTTP address: the port is opened with SO_REUSEPORT and the kernel spreads the connections over them. That is how several processes share one port; a second process started by mistake raises no error.

Note: copy’s defaults (--concurrency 4, --batch-size 1024) are higher than the library’s route defaults (concurrency: 1, batch_size: 512), because copy is built for bulk throughput. See Performance tuning.

Pipes: stdin and stdout

- as SOURCE reads stdin, - as TARGET writes stdout. Parameters follow as usual (-?format=csv). The copy ends when the input does, with or without --drain, and the copied … summary goes to stderr so it does not land among the rows.

pg_dump_rows | mqb copy '-?format=raw' 'kafka://localhost:9092?topic=rows'
mqb copy 'postgres://…/shop?table=orders&cursor_column=id' '-?format=raw' --limit 5

When the reader of stdout goes away (mqb copy … - | head -2), the copy ends with one line, Error: stdout was closed; stopped after … rows, and exit code 1.

On a terminal a copy that runs longer than two seconds shows a running row count on stderr; it is not written to a pipe or a log file.

Resumable copies

By default a bounded copy re-reads the whole source every run. --resume derives a stable state identity from the credential-redacted source, destination, and filter, then maps it to the source’s existing mechanism. Changing any of those pipeline semantics starts new state; rotating a password does not.

mqb copy \
  'postgres://user:pass@localhost/app?table=orders&cursor_column=id' \
  'file:///data/orders.jsonl' \
  --resume \
  --drain

Currently supported mappings are Kafka consumer groups, MongoDB capture_all/capture_new cursors, persistent Postgres CDC slots, SQL cursor readers with an explicit monotonic cursor_column, and ClickHouse/object-store cursor readers with their required explicit external checkpoint_store. Explicit group_id, cursor_id, slot_name, and checkpoint_store URI settings remain advanced overrides.

File offsets are deliberately not accepted yet because partial batch failure can advance the current file offset past a failed record. NATS is also rejected because its generated durable consumer name cannot currently include the destination and filter. Other non-replayable sources, including MQTT, fail early instead of silently ignoring --resume. Full checkpoint details: Checkpoints & resumable copies.

When a full re-copy is intentional, --no-resume skips optional cursor_id and checkpoint_store handling for SQL/ClickHouse cursor readers, MongoDB change streams, and object-store sources. A SQL or ClickHouse cursor_column is still required for paging. For native-position sources such as Kafka, NATS, and Postgres CDC, --no-resume is intentionally a no-op; configure their normal group, durable, slot, or start-position settings instead.

Filtering

--filter is evaluated by mq-bridge after the source read and before URI-configured transform middleware. It is not translated to SQL or MongoDB, so connector-native predicates remain a separate optimization and keep their existing behavior.

mqb copy \
  'kafka://localhost:9092?topic=orders' \
  'postgres://localhost/app?table=orders' \
  --filter 'country == "DE" && amount >= 50' \
  --resume

An expression reads payload fields by bare name, including nested paths (order.status), and message metadata under the reserved meta. prefix (meta.kind). Metadata is always text, as are CSV fields and a SQL source’s numeric and timestamp columns, but comparing one against a numeric literal reads it as a number: meta.retry_count < 3 and amount >= 50 both work untouched. A string literal still compares as text (zip == "01234"), and number() remains available.

A true result continues to the destination. A false result is an intentional successful drop and advances the source acknowledgement/checkpoint. Invalid expressions and payloads that are not a JSON object are errors. A referenced field that is absent, null, or holds an array or object counts as no match instead, the way a SQL WHERE treats NULL, so one heterogeneous record does not end a copy that is otherwise running fine. The first such field is logged once as a warning, so a typo in the expression does not simply look like an empty source.

Object naming under a filter

An object_store sink names each object after the contiguous source range it covers whenever name_by resolves to source_position — which is what auto picks for any source that stamps a replay position. A filter leaves holes in every batch, and each hole starts another object: measured at 159,474 objects instead of 977 on a 1M-row copy, a ~220x drop in throughput.

So when a row-dropping middleware is present — --filter, deduplication, weak_join, or a transform with on_error: reject on either endpoint — and the sink is still on name_by: auto, the route resolves it to write_time instead and warns that it did. mq-bridge applies this to every route, so a switch in when mode with no default and a config-defined route get the same treatment. That has two consequences worth knowing:

• Objects are named by uuidv7 at write time, so at the default --concurrency 4 their order is the order batches finish encoding, not source order. Order within an object is unchanged. Use --concurrency 1 if order across objects matters.
• The copy is no longer effectively-once at the sink: a crash mid-batch rewrites those rows under fresh names rather than colliding harmlessly with identical ones.

Pass name_by=source_position explicitly to keep replay-safe names and accept the fragmentation. An explicit setting — including the deprecated idempotency alias — is never overridden.

URI grammar

scheme://…?param=a&next=b: the scheme selects the endpoint and query parameters set its config. Any query key that matches a field of that endpoint’s config becomes endpoint config; every other query param stays on the connection URL, so driver params pass through unchanged (e.g. postgres://…/db?table=src&sslmode=disable).

• Schemes: postgres / postgresql / mysql / mariadb / sqlite → sqlx, nats → NATS, mongodb → MongoDB, redis → Redis streams, file → file, s3 / gs / az / abfs → cloud object storage (credentials from the environment), and the rest by name.
• Common config params: table, insert_query (URL-encoded; supports ${metadata:<key>} / ${payload:<field>} token mapping), delete_after_read, subject, stream, collection, database, format, … — anything on the endpoint’s config struct.
• For nats, the dominant target field can be given as the URL path (nats://localhost:4222/orders ≡ ?subject=orders); the query form wins if both are given. A redis path is the connection’s database number, so a redis stream target must use ?stream=….
• MongoDB sources are non-destructive by default: consume defaults to capture_all, which needs a replica set (a single-node one is enough). On a standalone mongod pass ?consume=snapshot for a one-shot read. ?consume=delete opts into the destructive queue-drain mode.

Middlewares in the URI

Append |-separated middlewares to either URI to wrap that endpoint. They apply in the order written, and each takes its own config struct’s fields as query params:

mqb copy \
  --from 'postgres://user:pass@localhost/db?table=src|retry?max_attempts=5&initial_interval_ms=200' \
  --to   'kafka://broker:9092?topic=orders|buffer?max_messages=500&max_delay_ms=50|metrics' \
  --drain

• Names: retry, metrics, dlq, deduplication, transform, id, filter, delay, timeout, limiter, buffer, weak_join, cookie_jar, lookup, aggregate, otel, random_panic, compression, encryption, pack, unpack, custom (- is accepted for _).
• encryption’s key is a shell-visible argument; prefer ${env:VAR} to keep it out of the process list and shell history: |encryption?key=$%7Benv:MQB_KEY%7D.
• compression and encryption produce binary payloads, so a file sink holding them must use format=normal. format=json/text render the payload as a JSON value and it does not survive the round trip (it comes back as a JSON array, and the reader reports a bogus “unsupported encryption envelope version 91”).
• Middlewares apply in list order on both ends, so a route that reads back what another wrote must list them in the reverse order. Writing with |compression?algorithm=zstd|encryption?key=… reads back with |encryption?key=…|compression?algorithm=zstd.
• A middleware with no params needs no ? — |metrics.
• dlq’s endpoint is itself a URL-encoded endpoint URI: |dlq?endpoint=file%3A%2F%2F%2Ftmp%2Ffailed.jsonl.
• Object/array fields take a JSON literal: |weak-join?group_by=cid&expected_count=2&timeout_ms=1000&required=["a","b"].
• A literal | inside the URI (e.g. in a password) must be written percent-encoded as %7C.

Structural endpoints in the URI

Structural endpoints have no connection of their own — they arrange other endpoints. Their nested endpoints are query params that are themselves endpoint URIs:

A nested URI only needs percent-encoding when it carries &, # or | of its own — fanout:?to=http://prod.internal/?method=PUT is fine as written, while an inner & must be encoded, as in fanout:?to=http%3A%2F%2Fprod.internal%2F%3Fmethod%3DPUT%26timeout_ms%3D1000.

switch has the two modes the engine has, and takes one or the other, never both. Value lookup branches on a metadata value; predicate mode takes when=<expression> / to=<uri> pairs in the order written, first match wins, using the same expression language as --filter:

mqb copy \
  'kafka://localhost:9092?topic=orders' \
  'switch:?when=amount > 1000&to=kafka%3A%2F%2Flocalhost%3A9092%3Ftopic%3Dlarge&when=true&to=file%3A%2F%2F%2Fdata%2Frest.jsonl'

An expression travels as a query value, so its == needs no escaping. A literal & would split the query, so write and / or rather than && / || — the engine accepts both spellings. A message matching no predicate goes to default, and without a default it is dropped.

fanout’s two branch kinds differ in what may come back: a to branch is used as written, while a mirror branch has its response and its failures discarded, so it can neither answer the caller nor fail the message for the other branches. That makes the mirroring proxy one command — serve requests, copy each to staging, answer from production:

mqb copy \
  --from 'http://0.0.0.0:8080' \
  --to   'fanout:?mirror=http://staging.internal/&to=http://prod.internal/'

The caller gets production’s response: a fan-out answers with its first to branch in the order written, and a mirror branch that is down does not change it.

An HTTP source is a listener, so --from http://0.0.0.0:8080 binds that address, and https:// makes it a TLS listener. Its certificate is the tls field, which takes a JSON literal:

mqb copy \
  --from 'https://0.0.0.0:8443?tls={"required":true,"cert_file":"cert.pem","key_file":"key.pem"}' \
  --to   'http://prod.internal/'

Escape hatch: full connection strings

Any query parameter that isn’t a recognised config field (e.g. sslmode, replicaSet) stays on the connection URL, so driver options just work. That includes a name shared with an object-typed config field: ?tls=true reaches the driver, while only an actual JSON literal — ?tls={"required":true,"ca_file":"ca.pem"} — is read as endpoint config. If you already have a complete connection string, pass it verbatim with ?url=<url-encoded>:

mqb copy \
  --from 'mongodb://_/?url=mongodb%3A%2F%2Fuser%3Apass%40host%2Fdb%3Ftls%3Dtrue&database=db&collection=orders' \
  --to null:

See the Quick start for complete, working copy commands.

checkpoint — inspect and edit a resume position

mqb -c config.yaml checkpoint show --route orders_sync     # a configured route or consumer
mqb checkpoint show SOURCE TARGET [--filter EXPR]         # a `copy --resume` job
mqb -c config.yaml checkpoint reset --route orders_sync    # next run starts from the beginning
mqb -c config.yaml checkpoint set --value int:1000 --route orders_sync

--route NAME names a route or consumer from the config, which -c loads (it goes before checkpoint). Otherwise pass the same SOURCE, TARGET and --filter the copy --resume ran with; they resolve to the same generated cursor id. show prints the stored value, the source that wrote it, and when:

value:   int:3
source:  sqlx:orders:id
updated: 2026-09-30T08:12:44.120+00:00

reset prints the previous value before deleting it, so set --value can undo it. The value format is the one show prints. -v logs connection details.

Stop the route or copy before reset or set: a reader that is still running saves its own position over the edit with its next acknowledged batch.

It works for SQL, ClickHouse and object-store cursor readers and for MongoDB change streams. Postgres CDC is refused: its replication slot is the authoritative position. See Checkpoints & resumable copies.

mcp — MCP server

mqb mcp                                    # stdio (local clients)
mqb mcp --transport http --bind 127.0.0.1:9092   # streamable HTTP

mcp install / uninstall / status

Register the running binary with local MCP clients so you don’t write the config by hand:

mqb mcp install                         # every detected client
mqb mcp install --client cursor --local  # one client, project-scoped
mqb mcp install --report-to-ui           # bake --report-to-ui into the entry
mqb mcp status
mqb mcp uninstall

--print-config prints the JSON snippet for a client not written directly. Full tool and message reference is in MCP server.

status — what is running on this machine

mqb status                 # live table, redrawn every second until Ctrl-C
mqb status --watch 5       # redraw every 5 seconds
mqb status --no-watch      # one table, then exit
mqb status --json          # the records as JSON, once

The table redraws by default only on a terminal. Piped output and --json print once, so scripts end; add --watch to keep either going.

One row per route or consumer of every mqb process your user is running: a copy, a headless config run, the web UI, the desktop app or an MCP server. It reads a local status registry those processes write to, so it needs no config, starts nothing and connects to no endpoint.

INSTANCE            NAME    FLOW              STATE    RATE      AVG       TOTAL   PENDING  UPTIME
cli copy [48211]    copy    kafka → postgres  running  8123.4/s  7990.1/s  412034  -        51s
mcp default [4790]  orders  nats → file       running  210.0/s   198.7/s   23844   3        2m00s

A process leaves the table within 7 seconds of exiting. --json prints an array of instance records, each with its consumers, publishers and routes; with --watch it prints one compact array per line. Errors appear as the word error unless MQB_STATUS_SHOW_ERRORS=1 is set for both the reporting process and mqb status; see Runtime route status.

MCP server

data movement as a tool call

mqb mcp runs the bridge as a Model Context Protocol server, so an agent can move data between any two supported endpoints — 15+ connectors covering databases, queues, brokers, HTTP, and files — from natural language.

Nothing is preconfigured. publish and start_route take their endpoints inline as JSON keyed by connector type, so the model picks both ends ad hoc; there is no YAML to write first and no per-connector tool to install. list_routes, route_status, route_messages, and stop_route manage what is already running.

The rows never enter the model’s context. The agent describes the job; the engine moves the bytes. Moving a 116.3 MiB dataset costs three tool calls and ~385 tokens — the same ~385 tokens whether the job is 1,000 rows or 1,000,000.

Quick start

# 1. get the binary (any install method from the Installation page)
brew install marcomq/tap/mq-bridge-app

# 2. register it with every MCP client detected on this machine
mqb mcp install

# 3. restart the client fully — reopening a tab is not enough

Then ask the agent for the job in words:

“Drain the events Redis stream into the events table in Postgres, creating the table if needed.”

Running the server

# stdio — for local clients (Claude Code, Claude Desktop, Cursor)
mqb mcp

# streamable HTTP — for remote/shared clients
mqb mcp --transport http --bind 127.0.0.1:9092

No web UI is started in this mode. Logs go to stderr, because stdio transport owns stdout for the protocol itself.

Tools

Call server_info before quoting any throughput number: a debug binary reports much slower rates, and the figure would be meaningless.

Agent messaging

Two agents on one machine can message each other without a broker. This is off by default: start the server with --agent-bus (or install it that way, see below) and neither tool exists otherwise.

agent_listen opens this server’s inbox — a dir_spool directory under $MQB_AGENTS_DIR (default ~/.mqb-agents/<name>) — and tails it for the session. Nothing can reach an agent that has not called it, so the bus is opt-in twice over: the flag offers the tools, the call opens the mailbox.

mqb mcp --agent-bus                 # without this the two tools are not registered

agent_listen {"name": "claude"}     # opt in; off until called
agent_send   {"to": "gpt-worker", "message": "..."}
route_messages {"name": "agent-inbox"}                # collect mail

To register a client with the bus already enabled:

mqb mcp install --agent-bus

Mail collected this way is held in a ring of the last capture_last messages (default 200), and the inbox file is deleted as it is read. An agent that leaves a session running without calling route_messages therefore loses the overflow — raise capture_last, or collect more often.

server_info reports the bus: who this server listens as, and which peers have an inbox and are currently listening. One message is one atomically renamed file, so a peer that is not running collects its mail when it next listens, and any process that can write a file can take part — no MCP client required:

f=~/.mqb-agents/claude/$(printf '%09d' $SEQ)-$(uuidgen | tr -d - | cut -c1-12)
printf '%s' '{"from":"gpt","message":"hi"}' > "$f.bin.tmp" && mv "$f.bin.tmp" "$f.bin"

Both tools take an optional endpoint (input / output) replacing the local directory, so the same bus works over NATS, Kafka or anything else when the peer is not on this machine.

Being woken by mail

MCP has no way to wake an agent: a server cannot start a turn, so agent_listen can only hold the mail until the agent asks for it. A background job can wake one, and agent-listen is that job — it blocks until mail arrives, writes it out and exits, so its completion is the notification:

mqb agent-listen claude --wait 600    # blocks; exits when mail lands

It is sugar for copy --from spool://<inbox> --wait <secs>, and writes each envelope as one line. Point it somewhere explicit with --to.

Use agent_listen for a mailbox you poll during a session, and agent-listen when you want the arrival itself to interrupt you — but only one of them per inbox. dir_spool allows a single consumer, so while an MCP server is listening as claude, mqb agent-listen claude fails with already held by a consumer.

The refusal works in both directions: agent-listen holds the inbox for its whole wait rather than polling it, so a second listener on the same name fails instead of quietly splitting the mail with the first.

It stops at the first delivery, not at an empty inbox, so a burst that is still arriving may leave messages queued. They are not lost — the next agent-listen picks them up, and the summary line says how many came through this time.

start_route and route_messages are annotated not read-only — starting a route moves real data, and reading captured messages consumes the buffer — so a client should not auto-approve them the way it may auto-approve a true reader.

Endpoints

Every endpoint is a single-key JSON object naming the connector:

{"kafka":   {"url": "localhost:9092", "topic": "orders"}}
{"nats":    {"url": "nats://localhost:4222", "stream": "ORDERS", "subject": "ORDERS.new"}}
{"sqlx":    {"url": "postgres://user:pass@localhost:5432/db", "table": "events"}}
{"file":    {"path": "/tmp/out.jsonl", "format": "json"}}
{"null":    null}

The key is the connector; the object is its configuration:

Structural endpoints (fanout, switch, stream_buffer, request, reader, response, static, ref) are accepted here too and are documented under Middleware & structural endpoints.

A file/object_store source must repeat the compression/encryption its data was written with — neither is auto-detected, and a mismatch ends the route as completed having moved few or no messages rather than failing. Check the moved-message count.

Messages

A message is a payload plus optional string metadata headers and an optional message_id (a UUIDv7 is generated when omitted). A string payload is sent verbatim; any other JSON value is serialized to JSON bytes.

{
  "payload": {"id": 1, "sku": "A-100", "qty": 3},
  "metadata": {"kind": "order", "correlation_id": "abc-123"}
}

Conventional metadata keys: kind (message type, drives type-based routing), correlation_id and reply_to (request/reply). Keys starting with mqb.src. are reserved for provenance and are stripped on input.

Route options

Alongside input and output, a route accepts batch_size, concurrency, exit_on_empty (drain the source, then exit) and capture_last. Without exit_on_empty a route polls indefinitely until stop_route.

batch_size and concurrency may be written inside route or as top-level start_route arguments. They resolve most-specific-first:

1. the top-level argument, if given;
2. otherwise the same key inside route, if given;
3. otherwise the app default — 1024 and 4, rather than the library’s 512/1.

Only a value you never wrote is replaced, so an explicit 1 is honoured either way. start_route echoes the resolved batch_size/concurrency in its result.

capture_last: N keeps the last N messages that flow through the route for route_messages to return. It is off by default: captured payloads are held in memory and handed to the model verbatim, which is exactly what “the rows never enter the context” otherwise avoids. Use it to sample a stream, not to move it.

Middleware

Either endpoint may carry a middlewares array, so delivery behaviour is added without changing the route:

{
  "input":  {"kafka": {"url": "localhost:9092", "topic": "orders", "group_id": "agent"}},
  "output": {
    "sqlx": {"url": "postgres://user:pass@localhost:5432/db", "table": "orders"},
    "middlewares": [
      {"retry": {"max_attempts": 5}},
      {"dlq": {"endpoint": {"file": {"path": "/tmp/failed.jsonl"}}}}
    ]
  }
}

Available: retry, dlq, deduplication, limiter, transform, compression, encryption, pack, unpack, buffer, delay, timeout, weak_join, cookie_jar, metrics. Each is documented with its options in the Cookbook and Middleware reference.

Waiting for a job to finish

For a job started with exit_on_empty, call wait_route rather than polling route_status in a loop:

{"name": "redis-to-postgres", "timeout_ms": 300000}

It returns once the route’s task ends, reporting finished, outcome, messages, elapsed_s and average_messages_per_second. On timeout it returns finished: false and leaves the route running, so the agent can wait again or stop it. The default timeout is 60 s and the ceiling is 5 h; the server polls internally every 50 ms, which the client never pays for.

This is what keeps agent token cost flat: one call covers a job of any duration, where a polling loop costs a call per tick.

Route status

route_status (and list_routes) report two rates, and they answer different questions:

For a route that is running now, read messages_per_second. For one that has finished — anything started with exit_on_empty — read average_messages_per_second: the instantaneous rate of a completed job is ~0 by the time any status call observes it, which says nothing about how fast it was. stop_route returns the same two fields as its parting summary.

elapsed_s and average_messages_per_second are null until the route has been sampled at least once (the sampler runs every 200 ms), so a job that finishes inside one tick reports no average.

Examples

Publish a batch to NATS JetStream:

{
  "publisher": {"nats": {"url": "nats://localhost:4222", "stream": "ORDERS", "subject": "ORDERS.new"}},
  "messages": [
    {"payload": {"id": 1, "sku": "A-100"}, "metadata": {"kind": "order"}},
    {"payload": {"id": 2, "sku": "B-200"}, "metadata": {"kind": "order"}}
  ]
}

Drain a Redis stream into Postgres, then exit:

{
  "name": "redis-to-postgres",
  "route": {
    "input":  {"redis_streams": {"url": "redis://localhost:6379", "stream": "events",
                                 "group": "g1", "read_from_start": true}},
    "output": {"sqlx": {"url": "postgres://user:pass@localhost:5432/db",
                        "table": "events", "auto_create_table": true}},
    "exit_on_empty": true
  }
}

Tail a live Kafka topic and keep the last 20 messages for inspection:

{
  "name": "orders-tail",
  "route": {
    "input":  {"kafka": {"url": "localhost:9092", "topic": "orders", "group_id": "agent"}},
    "output": {"null": null}
  },
  "capture_last": 20
}

Performance

The interface costs one round-trip, not a per-row tax: a route started through start_route moves data at the rate the copy CLI does, and what separates them is a fixed ~72 ms of route startup and completion polling — 0.426 s against the CLI’s 0.354 s on the same 1M-row dataset in the same session. Most of that is the measuring client asking route_status every 50 ms, so completion is noticed up to one poll late; the two tool calls bracketing the job cost ~0.05 ms each at p50. The gap is fixed, so it shrinks to nothing as a proportion as the dataset grows — and it is more visible than it used to be only because the job itself got ~2.6x faster.

Agent token cost is flat in the number of rows moved — three tool calls (start_route, one route_status, stop_route) totalling ~1.5 KB of JSON-RPC move 1M rows or 1,000 alike. Passing the same 116.3 MiB through a context window instead would cost ~30.5M tokens, which no context window holds at any price.

Methodology and the client used to measure it — a real MCP client over stdio, not an in-process harness — are in benches/etl/README.md, scenario 7. Engine-level tuning is on the Performance tuning page.

Registering with a client

mcp install writes the client config for you, registering the absolute path of the binary you just ran — so a target/debug build and an installed release binary each register themselves correctly.

# every client detected on this machine
mqb mcp install

# a single client, project-scoped rather than global
mqb mcp install --client cursor --local

# bake --report-to-ui into the registered command
mqb mcp install --report-to-ui

Where the client ships its own CLI (Claude Code) that CLI is driven, since it stays correct across config-schema changes; otherwise the entry is merged into the client’s JSON, leaving every other registered server untouched. Installing twice is a no-op.

With --client omitted, every client detected on the machine is configured. For anything not listed above, mcp install --print-config prints the snippet:

{
  "mcpServers": {
    "mq-bridge": {
      "command": "mq-bridge-app",
      "args": ["mcp", "--transport", "stdio"]
    }
  }
}

Use an absolute path to the binary if it is not on PATH. With --transport http, point your client at http://127.0.0.1:9092/ instead.

Restart the client fully after installing — reopening a tab is not enough.

Under stdio the client spawns the server process, so a rebuilt binary is only picked up after the client restarts the server — an edit to mcp.rs alone will not change the behaviour of an already-running session.

Errors

Invalid endpoints, duplicate route names, and unknown route names come back as JSON-RPC -32602 (invalid params); a connection failure at publish time reports the underlying transport error:

invalid publisher endpoint: IO error: Connection refused (os error 61)

A partially failed batch returns a result flagged is_error with a {"status": "Partial", "sent": N, "failed": M, "errors": [...]} summary, so a partial send is not mistaken for success.

Known limitations

This originates in the upstream mq-bridge crate, not in the MCP layer.

• Finished routes are not reaped. A route started with exit_on_empty stays in list_routes after it has drained and exited; call stop_route to clear it. It is no longer indistinguishable from a running one, though: since mq-bridge 0.3.6 each entry carries "finished" plus an "outcome". While the route runs these are false / null; once its task ends, "finished": true and "outcome" is completed (drained cleanly) or failed (permanent error — the cause is in status.error).

  stop_route removes the entry from list_routes as it stops the route, so the third outcome, stopped, is not observable through these tools.

MCP Registry (server.json)

mcp install and the registry solve different problems, and neither needs the other:

So install remains the shortest path on a machine that already has the binary — it is not superseded by publishing to the registry.

server.json at the repository root is the registry entry. It publishes the server under io.github.marcomq/mq-bridge-app and offers two package types, both of which run the same mcp --transport=stdio command:

The registry only stores metadata — it never hosts the artifact — so it verifies that each package really belongs to this project by looking for an ownership marker inside the published artifact itself:

• OCI — a LABEL io.modelcontextprotocol.server.name in the Dockerfile final stage.
• Cargo — an mcp-name: line in the crate README, which crates.io renders. It must be visible markdown: crates.io strips HTML comments during rendering, so the hidden <!-- mcp-name: … --> form that works for PyPI and NuGet is silently dropped here.

Both markers must equal the name field in server.json.

Publishing happens in the publish-mcp-registry job of the release workflow. It runs after the ghcr manifest and the crates.io release exist, rewrites the versions in server.json from the release tag, authenticates with mcp-publisher login github-oidc (the io.github.* namespace is proven by the workflow’s own OIDC token, so there is no secret to rotate) and publishes. To do it by hand:

brew install mcp-publisher
mcp-publisher login github
mcp-publisher publish

Testing against local brokers

The compose files in the mq-bridge repo bring up the brokers:

cd ../mq-bridge/tests/integration/docker-compose
docker compose -f nats.yml -f postgres.yml -f redis.yml up -d

Verified end-to-end against these: batch publish to file/NATS/Redis Streams, NATS → file and Redis Streams → Postgres routes (including auto_create_table), the full route lifecycle, and the error paths above.

How the app is built

mq-bridge-app is a demonstration of the mq-bridge engine used on itself: the application serves its own management UI through the engine rather than through a separate web framework. If you want to understand the engine’s request-reply model in practice, this is a worked example. For the engine internals themselves, see Learn the architecture.

Backend: mq-bridge as a web server

Instead of using a traditional web framework like Actix or Axum directly for the management API, the application routes HTTP through the engine:

1. HTTP input — an http input endpoint listens on the configured UI port and converts incoming HTTP requests into CanonicalMessages.
2. WebUiHandler — a custom Handler processes these messages as a router: serving static files (HTML, JS) or handling API requests (e.g. /config, /schema.json).
3. Response output — the handler returns a response message, sent to a response output endpoint, completing the HTTP request-response cycle.

This showcases the library’s ability to handle request-reply patterns and act as a lightweight web server.

Frontend: vanilla-schema-forms

The Web UI is generated dynamically from the Rust configuration structures — no hand-written form code:

1. Schema generation — the backend uses schemars to generate a JSON Schema for the AppConfig struct at runtime, exposed via /schema.json (also available on the CLI: mqb --schema schema.json writes it to a file, mqb --schema - prints it).
2. Dynamic form — the frontend uses vanilla-schema-forms to render a complete configuration form from that schema alone.
3. No UI code changes — when a new feature or config option is added to the Rust code (e.g. a new middleware), the schema updates automatically and the UI reflects it without any frontend changes.

Project status

This project is in active development. It originally served as the primary reference implementation and testbed for the mq-bridge engine.

The UI/Tauri layer was largely prototyped quickly and does not mirror the mq-bridge / mq-bridge-app core/CLI standards — treat it as a working demo, not a reference implementation, and test before relying on it in production.

Use cases

Each page starts from a problem, shows one working solution with mq-bridge, and says where it stops working. Every page has a runnable example in the repository’s examples/ directory that CI executes.

Every page has the same sections: the problem, the solution with a diagram and configuration, a runnable example, failure semantics (restarts, duplicates, deletes, an unavailable sink), when not to use mq-bridge, and a comparison with alternatives.

Related recipes elsewhere in this book: query a stream in DuckDB, Postgres CDC to a file, cross-process IPC.

How to keep a vector index in sync with Postgres (CDC → embeddings → Qdrant)

One mq-bridge route reads a Postgres table through logical replication, calls an embeddings API for each inserted or updated row and upserts the result into Qdrant; a deleted row removes its point. It copies the existing rows first and needs no Kafka in between.

Runnable example: examples/postgres-cdc-to-qdrant.

The problem: a RAG index that drifts from the database

Retrieval-augmented generation (RAG) answers from whatever is in the vector index. When the source rows live in Postgres, the index has to follow them, and the usual approaches each miss something:

• A nightly re-embedding job leaves the index up to a day stale and pays for embedding rows that did not change.
• Dual writes in the application (write the row, then call the embeddings API and the vector database) lose updates when the second call fails after the first committed, and miss every change made by a migration, an admin console or another service.
• Polling an updated_at column never sees deletes, so removed documents keep showing up in search results.

Change data capture reads the write-ahead log, so it sees every committed insert, update and delete regardless of who made it.

The solution: one route from the WAL to the index

flowchart LR
  PG[(Postgres<br/>docs table)] -- logical replication --> B[mq-bridge route]
  B -- "insert / update: row text<br/>POST /v1/embeddings" --> E[Embeddings API]
  E -- "vector" --> B
  B -- "upsert point with vector" --> Q[(Qdrant)]
  B -- "delete:<br/>delete point" --> Q

Postgres needs wal_level = logical and a publication on the table:

CREATE PUBLICATION docs_pub FOR TABLE docs WITH (publish = 'insert, update, delete');

The route:

docs_to_qdrant:
  input:
    postgres_cdc:
      url: "postgres://app:app@postgres:5432/app"
      publication: "docs_pub"
      slot_name: "docs_to_qdrant"
      consume: capture_all                             # copy existing rows, then follow changes
      checkpoint_store: "file:///app/checkpoint.json"  # a restart does not repeat the copy
  output:
    switch:
      metadata_key: "postgres.operation"
      cases:
        # Deletes carry only the primary key: remove the point, call no model.
        delete:
          qdrant: &qdrant
            url: "http://qdrant:6333"
            collection: docs
            operation: "${metadata:postgres.operation}"
      # Inserts and updates. Output middlewares run bottom-up.
      default:
        middlewares:
          # 2. Shape the Qdrant point: {id, vector, payload}.
          - transform:
              mapping:
                id: "$.id"
                vector: "$.vector"
                "payload.title": "$.title"
                "payload.body": "$.body"
          # 1. One embeddings request per batch; the vector lands in `vector`.
          - lookup:
              from:
                openai_embeddings:
                  url: "http://ollama:11434"
                  model: all-minilm
                  input: "${payload:body}"
              into: vector
          - retry: { max_attempts: 5 }
        qdrant: *qdrant

How it fits together:

• postgres_cdc emits one message per changed row. The payload is the row as flat JSON; the operation is in the postgres.operation metadata key.
• switch sends deletes straight to Qdrant. Everything else goes through the embedding step first.
• openai_embeddings sends the texts of a whole batch in one request to an API that speaks the OpenAI wire format, which Ollama also serves. A hosted provider needs another url, a model name and an api_key.
• qdrant batches points into Qdrant’s upsert and delete requests. The row’s primary key is the point id, so an update overwrites the point of that row.

This route was run with a file as its input against Ollama and Qdrant 1.19; with postgres_cdc as its input it was not.

The collection must exist with the vector size of the model (384 for all-minilm); the example’s compose file creates it.

The same route by hand

qdrant and openai_embeddings are requests filled in for you. Written out with the generic http_bulk and http endpoints, the route looks like this. It is the form to start from for a vector store or an embeddings API with another request shape, and it is what the runnable example uses (routes.yaml), since its container image predates the two endpoints.

docs_to_qdrant:
  input:
    postgres_cdc:
      url: "postgres://app:app@postgres:5432/app"
      publication: "docs_pub"
      slot_name: "docs_to_qdrant"
      consume: capture_all                             # copy existing rows, then follow changes
      checkpoint_store: "file:///app/checkpoint.json"  # a restart does not repeat the copy
  output:
    switch:
      metadata_key: "postgres.operation"
      cases:
        # Deletes carry only the primary key: remove the point, call no model.
        delete:
          http_bulk: &qdrant
            url: "http://qdrant:6333"
            operation: "${metadata:postgres.operation}"
            upsert:
              method: PUT
              path: /collections/docs/points?wait=true
              format: json_array
              envelope: '{"points": {documents}}'
            delete:
              path: /collections/docs/points/delete?wait=true
              envelope: '{"points": {ids}}'
      # Inserts and updates. Output middlewares run bottom-up.
      default:
        middlewares:
          # 3. Shape the Qdrant point: {id, vector, payload}.
          - transform:
              mapping:
                id: "$.id"
                vector: "$.embedding.data[0].embedding"
                "payload.title": "$.title"
                "payload.body": "$.body"
          # 2. Call the embeddings API; the response lands in `embedding`.
          - lookup:
              from:
                http: { url: "http://ollama:11434" }
              metadata:
                http_path: "/v1/embeddings"
                http_method: POST
              into: embedding
          # 1. Add the fields the embeddings API expects: `model` and `input`.
          - transform:
              mapping:
                id: "$.id"
                title: "$.title"
                body: "$.body"
                input: "$.body"
                model: { path: "$.model", default: "all-minilm" }
          - retry: { max_attempts: 5 }
        http_bulk: *qdrant

Here the first transform builds the request the embeddings API expects, the lookup stores the whole response under embedding, and the last transform picks the vector out of it. One request is sent per row.

Run the route from Rust

The same file runs embedded in a service. Enable the features the route uses; the route written by hand also needs http:

[dependencies]
mq-bridge = { version = "1", features = ["postgres-cdc", "http-bulk", "yaml"] }
tokio = { version = "1", features = ["full"] }
anyhow = "1"

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    mq_bridge::deploy_file("routes.yaml").await?;
    tokio::signal::ctrl_c().await?;
    Ok(())
}

Run the route from Python

The PyPI package includes every connector:

from mq_bridge import Route

Route.from_file("routes.yaml", "docs_to_qdrant").run()

Run it without code

The example uses the mq-bridge-app container: mq-bridge-app --config routes.yaml.

Runnable example

git clone https://github.com/marcomq/mq-bridge && cd mq-bridge/examples/postgres-cdc-to-qdrant
docker compose up -d
./search.sh "when do I get my money back"

The stack is Postgres 16, Qdrant 1.19.0, Ollama 0.12.3 with a local 384-dimension model, and mq-bridge-app 0.4.18. ./test.sh checks the backfill, an insert, an update, a delete, a restart and a semantic search; it runs in CI.

Failure semantics

Limits of this recipe:

• One vector per row. There is no chunking step. To split long documents, do it in a handler and emit one point per chunk, or store chunks as rows.
• Every update re-embeds the row, including updates that did not change the embedded column.
• Large unchanged columns are omitted from update events. Postgres does not write an unchanged TOASTed value (roughly: text over 2 kB) to the WAL, and mq-bridge leaves that field out of the payload. An update that changes only title on a row with a long body then arrives without body. The example’s rows are short and do not hit this; the recipe has not been tested with TOASTed columns.
• Point ids are the unsigned integer primary key here. Qdrant accepts unsigned integers and UUIDs as point ids, so a table with another key type needs a transform that derives one.
• An empty text fails its batch. OpenAI refuses a request with an empty input, and one request covers the whole batch. Drop such rows with a filter first.

When not to use mq-bridge for this

• The vectors can live in Postgres. With pgvector the embedding is a column next to the row, and a trigger or the application can keep it current in the same transaction. There is no second system to sync. (Writing to a vector column through mq-bridge’s SQL output has not been tested.)
• You need chunking, re-ranking or multi-step document processing. That is application logic; a framework or your own worker that consumes the CDC stream fits better than a config file.
• The source is not Postgres or MongoDB. mq-bridge has native CDC for those two only. For MySQL, SQL Server or Oracle see CDC via Debezium.
• You already run Kafka Connect with Debezium. Adding a sink to that platform is less to operate than a second CDC reader on the same database.

Alternatives

Why mq-bridge can be the better fit here

• One process covers the whole pipeline. Capture, the embeddings call and the Qdrant write are one route in one file. With Debezium the capture and the embedding step run in Kafka Connect or Debezium Server, and writing to Qdrant is a further component.
• No JVM and no Kafka to operate. mq-bridge is a native binary (the 0.4.18 macOS arm64 build of mq-bridge-app is about 100 MB with every connector) or a library inside a Rust, Python or Node.js service you already run.
• Postgres CDC is part of the open-source core. mq-bridge is licensed MIT OR Apache-2.0, CDC included. Redpanda Connect’s postgres_cdc input states that it requires an enterprise license (linked above).
• Backfill and streaming are one setting. consume: capture_all copies the existing rows and then follows the WAL, so a new index needs no separate initial load.
• The route is testable without infrastructure. The same route and handler code runs against in-memory endpoints; see Test without a broker.

Where the alternatives are the better fit: Debezium covers more databases and embedding providers, and Redpanda Connect has embedding processors for providers that do not speak the OpenAI wire format.

No throughput comparison against these tools has been measured, so none is claimed. mq-bridge’s own numbers are on the benchmark dashboard and in the tuning page.

How to implement the transactional outbox pattern with Postgres and Kafka

The application inserts an event row into an outbox table in the same transaction as its business change, and one mq-bridge route relays those inserts from the Postgres write-ahead log to Kafka. Events are published if and only if the transaction committed; delivery is at-least-once.

Runnable example: examples/transactional-outbox.

The problem: the dual write

A service that updates its database and then publishes an event performs two writes that cannot be committed together:

• Commit, then publish. If the process dies or the broker is unreachable between the two, the order exists and the event never does. Other services never hear about it.
• Publish, then commit. If the transaction rolls back, subscribers have acted on an order that does not exist.
• Publish inside the transaction. The broker call still succeeds or fails independently of the commit, and it holds the transaction open across a network call.

The transactional outbox pattern replaces the second write with an insert into a table in the same database. A separate relay publishes the rows. Reading them from the WAL instead of polling the table means no polling query, no published flag to update and no lock contention with the application.

The solution: relay the outbox table from the WAL

flowchart LR
  A[Application] -- "one transaction:<br/>orders + outbox" --> PG[(Postgres)]
  PG -- logical replication --> B[mq-bridge route]
  B -- "key = aggregate_id" --> K[[Kafka topic]]
  K --> C[Consumers<br/>dedupe on event id]

The outbox table and a publication that carries only its inserts (Postgres needs wal_level = logical):

CREATE TABLE outbox (
  id           bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  aggregate_id text  NOT NULL,
  event_type   text  NOT NULL,
  payload      jsonb NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now()
);

-- Only inserts are relayed, so the application can delete old outbox rows freely.
CREATE PUBLICATION outbox_pub FOR TABLE outbox WITH (publish = 'insert');

The application writes both rows in one transaction:

BEGIN;
INSERT INTO orders (sku, qty) VALUES ('book-42', 2);
INSERT INTO outbox (aggregate_id, event_type, payload)
VALUES ('order-1', 'order.placed', '{"order_id": 1, "sku": "book-42", "qty": 2}');
COMMIT;

The relay (routes.yaml):

outbox_to_kafka:
  input:
    postgres_cdc:
      url: "postgres://app:app@postgres:5432/app"
      publication: "outbox_pub"   # publishes inserts on the outbox table only
      slot_name: "outbox_relay"
    middlewares:
      # Copies the aggregate id into the `mqb.id` metadata key ...
      - id: "${payload:aggregate_id}"
  output:
    middlewares:
      - retry: { max_attempts: 10 }
    kafka:
      url: "kafka:9092"
      topic: "order-events"
      partition_key: "mqb.id"     # ... so events of one aggregate share a partition, in order

How it fits together:

• postgres_cdc emits one message per inserted outbox row. The Kafka record value is that row as JSON, with the jsonb column as a nested object.
• The id middleware puts the aggregate id into metadata, and the Kafka output’s partition_key uses that metadata field as the record key.
• A rolled-back transaction never reaches the WAL’s committed stream, so nothing is published.

For NATS, replace the output with a nats endpoint. The runnable example and its test cover Kafka only.

Run the relay from Rust

[dependencies]
mq-bridge = { version = "1", features = ["postgres-cdc", "kafka", "yaml"] }
tokio = { version = "1", features = ["full"] }
anyhow = "1"

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    mq_bridge::deploy_file("routes.yaml").await?;
    tokio::signal::ctrl_c().await?;
    Ok(())
}

Running the relay inside the service that owns the outbox is possible, but then every replica of the service would start a relay on the same replication slot, and a slot serves one reader. Run it as one separate process or one designated instance.

Run the relay from Python

from mq_bridge import Route

Route.from_file("routes.yaml", "outbox_to_kafka").run()

Run it without code

The example uses the mq-bridge-app container: mq-bridge-app --config routes.yaml.

Runnable example

git clone https://github.com/marcomq/mq-bridge && cd mq-bridge/examples/transactional-outbox
docker compose up -d --wait
./test.sh

The stack is Postgres 16, Apache Kafka 3.9.0 and mq-bridge-app 0.4.18. ./test.sh commits one transaction, rolls one back, deletes the outbox rows, restarts the relay and checks what is on the topic; it runs in CI. The example’s README shows the same steps by hand.

Failure semantics

When not to use mq-bridge for this

• You need exactly-once delivery to Kafka. This relay is at-least-once. If consumers cannot deduplicate, this pattern with any CDC relay is the wrong tool.
• You cannot enable logical replication. Some managed databases or permission models rule out wal_level = logical or a replication role. A polling relay that selects unpublished rows works with plain SQL access.
• The database is not Postgres. mq-bridge reads the WAL of Postgres and the change streams of MongoDB. For MySQL, SQL Server or Oracle see CDC via Debezium.
• You want events routed to a topic per aggregate type from one config. The example publishes every event to one topic. Debezium’s outbox router does this routing by default.

Alternatives

Why mq-bridge can be the better fit here

• No Kafka Connect cluster. The relay is one native binary or a library call in a Rust, Python or Node.js process. Debezium’s outbox router is a transformation that runs inside Kafka Connect.
• Your outbox schema, not a prescribed one. Any columns work; the key is whichever field you name in the id middleware. Debezium’s router expects its default column names unless reconfigured.
• The broker is a config line. Replacing the kafka output with nats, amqp, mqtt or aws relays the same outbox to another broker, with the same retry and dead-letter middleware.
• Postgres CDC is part of the open-source core. mq-bridge is licensed MIT OR Apache-2.0, CDC included. Redpanda Connect’s postgres_cdc input states that it requires an enterprise license (linked above).
• The consumers of the events can be tested without a broker, with the same library; see Test without a broker.

Where the alternatives are the better fit: Debezium’s router implements the outbox conventions (topic per aggregate type, event id header, key) out of the box and supports more databases. A polling publisher (a job that selects new outbox rows and marks them as sent) needs no replication privileges and no extra component, at the cost of polling load, added latency and write contention on the outbox table.

Measured: memory of the relay

Relaying a backlog of 500,000 outbox rows to Kafka, mq-bridge-app peaked at 51 to 142 MiB of container memory and Debezium Server at 562 to 603 MiB (two passes each, the same Kafka producer settings, Docker Desktop on an Apple M1, 2026-10-05).

Other measurements of mq-bridge are on the benchmark dashboard and in the tuning page.

How to test Kafka consumers and message handlers without a broker

Build the route from endpoints passed in by the caller: Kafka (or NATS, RabbitMQ, MQTT) in production, mq-bridge’s in-process memory endpoint in tests. The handler, the route and the middleware under test are the production code; only the transport is swapped, so the tests need no Docker and wait for no broker to start.

Runnable example: examples/test-without-a-broker.

The problem: slow, flaky or shallow tests for messaging code

Code that consumes from a broker is awkward to test:

• A real broker in every test run (Docker Compose or Testcontainers) costs seconds to minutes of startup, needs Docker on every developer machine and CI runner, and fails for reasons that have nothing to do with the change under test.
• Mocking the client library tests the mock. Hand-written fakes of a Kafka consumer drift from the real API and skip the code between the client and the handler: deserialization, routing by message type, acknowledgement.
• Testing the handler function alone is fast but leaves that same glue untested.

The solution: make the transport a parameter

flowchart LR
  subgraph Production
    K1[[Kafka topic]] --> R1[route + handler] --> K2[[Kafka topic]]
  end
  subgraph Test
    M1[memory channel] --> R2[same route + handler] --> M2[memory channel]
  end

A route is an input endpoint, an output endpoint and an optional handler. The handler works with mq-bridge’s message type and never sees a Kafka record, so the same route runs over any endpoint. In production the two endpoints are a broker:

orders:
  input:
    kafka: { url: "localhost:9092", topic: "orders.in", group_id: "orders" }
  output:
    kafka: { url: "localhost:9092", topic: "orders.out" }

In a test they are in-process channels, which need no Cargo feature and no service:

orders:
  input:
    memory: { topic: "orders.in", capacity: 100 }
  output:
    memory: { topic: "orders.out", capacity: 100 }

Rust

The route takes its endpoints as arguments:

use mq_bridge::{models::Endpoint, CanonicalMessage, Handled, HandlerError, Route};
use serde_json::{json, Value};

/// Business logic: reject orders without a positive amount, mark the rest as checked.
async fn check_order(msg: CanonicalMessage) -> Result<Handled, HandlerError> {
    let mut order: Value = msg
        .parse()
        .map_err(|e| HandlerError::NonRetryable(e.into()))?;
    if order["amount"].as_f64().unwrap_or(0.0) <= 0.0 {
        return Ok(Handled::Ack);
    }
    order["checked"] = json!(true);
    let out = CanonicalMessage::from_json(order).map_err(|e| HandlerError::NonRetryable(e.into()))?;
    Ok(Handled::Publish(out))
}

/// The route used in production and in tests. Only the endpoints differ.
pub fn orders_route(input: Endpoint, output: Endpoint) -> Route {
    Route::new(input, output).with_handler(check_order)
}

The test passes memory endpoints, sends messages into one channel and reads the other:

use std::time::Duration;

#[tokio::test(flavor = "multi_thread")]
async fn valid_orders_are_checked_and_invalid_ones_dropped() {
    let input = Endpoint::new_memory("orders.in", 100);
    let output = Endpoint::new_memory("orders.out", 100);
    let (orders_in, orders_out) = (input.channel().unwrap(), output.channel().unwrap());

    orders_route(input, output).deploy("orders").await.unwrap();

    for order in [json!({"id": 1, "amount": 25.0}), json!({"id": 2, "amount": 0})] {
        let msg = CanonicalMessage::from_json(order).unwrap();
        orders_in.send_message(msg).await.unwrap();
    }

    let mut received = Vec::new();
    for _ in 0..100 {
        received.extend(orders_out.drain_messages());
        if !received.is_empty() && orders_in.is_empty() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    Route::stop("orders").await;

    assert_eq!(received.len(), 1);
    let order: Value = received[0].parse().unwrap();
    assert_eq!(order, json!({"id": 1, "amount": 25.0, "checked": true}));
}

[dependencies]
mq-bridge = { version = "=0.4.18", default-features = false }   # production adds e.g. features = ["kafka"]
serde_json = "1"

[dev-dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] }

Python

The route is built from a config dict, so the test passes a different dict:

from mq_bridge import Route


def check_order(order):
    """Business logic: drop orders without a positive amount, mark the rest as checked."""
    if order.get("amount", 0) <= 0:
        return None
    return {**order, "checked": True}


def orders_route(config: dict) -> Route:
    """The route used in production and in tests. Only `config` differs."""
    return Route.from_config(config).add_handler("order.created", check_order)

from mq_bridge import Consumer, Publisher

from orders import orders_route

INPUT = {"memory": {"topic": "orders.in", "capacity": 100}}
OUTPUT = {"memory": {"topic": "orders.out", "capacity": 100}}


def test_valid_orders_are_checked_and_invalid_ones_dropped():
    publisher = Publisher.from_config(INPUT)
    results = Consumer.from_config(OUTPUT)

    with orders_route({"input": INPUT, "output": OUTPUT}):
        publisher.send_json({"id": 1, "amount": 25.0}, {"kind": "order.created"})
        publisher.send_json({"id": 2, "amount": 0}, {"kind": "order.created"})
        publisher.send_json({"id": 3, "amount": 5.0}, {"kind": "order.created"})
        received = []
        while len(received) < 2:
            batch = results.poll(max=10, timeout_ms=5000)
            assert batch, "timed out waiting for the route"
            received.extend(batch)

    assert [m.json() for m in received] == [
        {"id": 1, "amount": 25.0, "checked": True},
        {"id": 3, "amount": 5.0, "checked": True},
    ]

The Node.js binding has the same memory endpoint; this page’s example covers Rust and Python.

Runnable example

This one needs no Docker:

git clone https://github.com/marcomq/mq-bridge && cd mq-bridge/examples/test-without-a-broker
cargo test                                             # Rust
cd python && pip install -r requirements.txt && pytest  # Python

Both pin mq-bridge 0.4.18 and run in CI.

Failure semantics: what an in-memory test does and does not prove

A test over memory channels shares global state by topic name within one process. Give each test its own topic names (or route name) when tests run in parallel.

When not to use this

• You are testing the broker integration itself: consumer-group rebalancing, offset commits, partition assignment, authentication, schema-registry interaction. Use a real broker.
• Your code uses a broker client directly and you do not want mq-bridge in production. The memory endpoint replaces a transport behind mq-bridge’s route; it is not a fake Kafka server that other client libraries can connect to.
• You need wire compatibility checks between producers and consumers written in different stacks. Contract tests or a shared broker environment cover that.

Alternatives

Why mq-bridge can be the better fit here

• The test runs the production route. Nothing is faked: the handler, the dispatch on the message type, the middleware and the acknowledgement path are the same objects, with a different endpoint underneath.
• No Docker in the inner loop. The tests start no container, so they run wherever cargo test or pytest runs.
• The transport stays a configuration choice afterwards. The same handler can move from Kafka to NATS or RabbitMQ by changing the endpoint, with the tests unchanged.

Measured with the example on an Apple M1 (2026-10-05, three runs each): the Rust test finished in 0.02 s (cargo test 0.16 s wall time after the build) and the Python test in 0.07 s. An apache/kafka:3.9.0 container with the image already pulled took 4.3 to 4.8 s to pass its health check, before any test could run.

Where the alternatives are the better fit: Testcontainers is independent of the library you use and is the only option in this table that exercises a real broker. The two combine well: most tests on memory endpoints, a few integration tests against a container.

Postgres CDC → JSONL

Stream every insert/update/delete from a PostgreSQL table into a newline-delimited JSON file, continuously. This is change-data-capture (CDC): instead of reading a table snapshot, mq-bridge follows the write-ahead log via logical replication, so new changes keep flowing until you stop it.

Prerequisites

• PostgreSQL with logical replication enabled (wal_level = logical).

• A publication on the source table:

  CREATE PUBLICATION mqb_pub FOR TABLE orders;
  

  publication must already exist; slot_name is created automatically if missing (it is a permanent, resumable replication slot).

The one-liner

mqb copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to file:///data/orders.jsonl?format=json

--from uses the postgres-cdc:// scheme (alias pgcdc://); the URL underneath is a plain Postgres URL. --to writes each change as one JSON object per line (format=json); the path comes from the URI path itself, not a query param.

Each change arrives as a CanonicalMessage whose payload is the flat row and whose postgres.operation metadata marks the operation (insert/update/delete) — the same convention as MongoDB CDC, so typed handlers work identically across both.

Equivalent config file

The same thing as a route in mq-bridge.yaml, for the config-driven run forms:

orders_cdc:
  input:
    postgres_cdc:
      url: "postgres://user:pass@localhost:5432/app"
      publication: "mqb_pub"     # CREATE PUBLICATION mqb_pub FOR TABLE orders;
      slot_name: "mqb_slot"      # created if missing (permanent slot, resumable)
  output:
    file:
      path: "/data/orders.jsonl"
      format: "json"

Notes

• Resumability. The named replication slot persists progress on the server, so a restart resumes from the last confirmed LSN rather than re-copying from the start.
• This runs forever. CDC is a continuous stream; the process stays up tailing new changes. For a one-shot bounded copy of an existing table, use a plain postgres://…?table=… source instead (see Quick start).
• Performance. For sustained CDC/ETL throughput and how batching interacts with file sinks, see Performance tuning.

See also

• PostgreSQL CDC connector — URL format and examples.
• postgres-cdc parameter reference — every recognised field.
• File connector — formats and options for the sink.

HTTP webhook → MongoDB

Accept incoming HTTP webhooks on a local port and persist each request body as a document in MongoDB, with automatic retries on transient write failures. This is a config-driven route rather than a one-liner, because it wires middleware into the pipeline.

The route

mq-bridge.yaml:

webhook_to_mongo:
  input:
    http:
      url: "127.0.0.1:8080"
      # Force the normal route pipeline instead of the inline HTTP response fast path.
      inline_response_fast_path: false
  output:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "app_db"
      collection: "webhooks"
      format: "json"   # a bit slower, but stores readable documents
    middlewares:
      - retry:
          max_attempts: 3
          initial_interval_ms: 500

Run it in config mode:

mqb --config mq-bridge.yaml

Then POST to it:

curl -X POST http://localhost:8080 \
  -H 'content-type: application/json' \
  -d '{"event":"signup","user":"alice"}'

The document lands in app_db.webhooks.

What each piece does

• http input listens on 127.0.0.1:8080, i.e. localhost only. The endpoint is unauthenticated, so bind it to 0.0.0.0 only behind an authenticating proxy or a network you control. Setting inline_response_fast_path: false forces requests through the full route pipeline (input → middleware → output) instead of the fast inline-response path, so the retry middleware and the MongoDB write both run before the HTTP response is sent.
• retry middleware re-attempts a failed write up to max_attempts times, starting at initial_interval_ms and backing off. See the Retries & backoff recipe for the full knob set and its interaction with a dead-letter queue.

Note

Delivery here is at-least-once: a write that actually succeeded but whose acknowledgement was lost gets retried and inserts a second document. For an idempotent sink, set id_field on the mongodb output to a stable business key so the value becomes the document _id: a retry of an already-written message is then rejected by the unique _id instead of appending a new document — see Upserts.

• mongodb output with format: "json" stores each payload as a readable JSON document rather than the default full-message serialization.

Variations

• Want a synchronous reply to the caller? Attach a handler and use a response output — see the Request / reply tutorial.
• High request rates? The http input scales with connection concurrency; see Performance tuning.

See also

• Configuration grammar — this is Route 2 of the annotated example.
• MongoDB connector and parameter reference.
• HTTP connector and parameter reference.

Cross-process IPC bridge

The memory endpoint is more than an in-process channel: its url (alias topic) is a transport URL, so the same endpoint can bridge separate processes on one machine over a Unix socket or Windows named pipe — no broker required.

Transport URLs

The url field selects the transport:

The consumer side binds/listens and must be started before the publisher connects. IPC does not support subscribe_mode or request_reply. enable_nack defaults to true, but redelivery is consumer-local: a nacked message is retried inside the consumer and is lost if the consumer process dies.

Example: ingest process → Kafka

One process reads from a Unix socket and forwards to Kafka. Start this consumer first so the socket exists before any producer connects:

ipc_ingest:
  input:
    memory:
      url: "ipc://orders"
      capacity: 256
  output:
    kafka:
      topic: "orders"
      url: "localhost:9092"

mqb --config ipc_ingest.yaml

The named form is used rather than an explicit path so no privileged directory has to exist up front: /run/mq-bridge/ needs root, and ipc://orders falls back to $XDG_RUNTIME_DIR/mq-bridge or /tmp/mq-bridge. Give an absolute ipc:///abs/path.sock only once you have created and permissioned that directory.

Any other process on the same host can now publish into ipc://orders — via its own mq-bridge route, or programmatically through a Publisher built on a memory endpoint (see the embedding tutorial) — and the messages flow through to Kafka.

When to use it

• Decouple a producer and a sink into separate processes without standing up a broker — e.g. an app writes to the socket, a long-lived bridge process owns the Kafka/DB connections and batching.
• Language boundary. A Python or Node process publishes into the socket; a Rust bridge process does the heavy I/O.

capacity bounds the in-flight channel; tune it alongside the guidance in Performance tuning.

When not to use it

IPC is at-most-once across a consumer crash. A send is acknowledged once its bytes are in the kernel socket buffer, so messages the consumer had not yet processed are lost when it dies, and the publisher is never told. It also serves one publisher at a time: a second one connects without an error and then waits until the first disconnects.

If the hand-over must survive a crash of either process, use a directory spool between the two processes, or a broker.

See also

• Configuration grammar — this is Route 9 of the annotated example, with the full URL grammar in comments.
• Endpoints (concepts).

Upserts & insert-if-absent

For ETL, at-least-once delivery plus an idempotent write at the sink gives you effective exactly-once: a replayed or retried record must not create a duplicate row. The most robust place to enforce this is the sink’s own unique constraint — it’s already shared across every writer, so no extra state store is needed.

MongoDB — id_field

Point id_field at a top-level payload field and its value becomes the document _id. Re-inserting the same business key then hits the unique _id index and is treated as an idempotent success (the duplicate is skipped, not errored):

orders_to_mongo:
  input:  { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "shop"
      collection: "orders"
      format: json
      id_field: "order_id"   # payload {"order_id": "A-1", ...} → _id = "A-1"

The field’s JSON type is preserved (a number stays a BSON integer). The payload must contain the field, otherwise the message is dead-lettered rather than written with a random _id. Use id_field on sink collections only — a business-key _id is incompatible with the delete competing-consumer mode, which requires a UUID _id.

SQL (sqlx) — ON CONFLICT / ON DUPLICATE KEY

For a table whose columns are the record’s fields, columns: auto with key: id generates the upsert for you; see Writing by column name. The rest of this section is for a hand-written insert_query.

insert_query is user-supplied, so you write the dialect’s upsert directly. This requires a pre-existing UNIQUE/PRIMARY KEY on the key column, and is incompatible with bulk_copy (COPY cannot express ON CONFLICT) — so you trade peak throughput for deduplication.

-- PostgreSQL — insert if absent (drop duplicates):
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING

-- PostgreSQL — upsert (last write wins):
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body})
  ON CONFLICT (id) DO UPDATE SET body = EXCLUDED.body

-- MySQL / MariaDB:
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body})
  ON DUPLICATE KEY UPDATE body = VALUES(body)

-- SQLite:
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING

A plain INSERT without a conflict clause instead fails the row as a non-retryable error (captured by a dlq, or logged and dropped without one). ${payload:field} binds a typed value from the JSON payload; ${metadata:key} binds a metadata string.

ClickHouse — ReplacingMergeTree

ClickHouse has no unique constraints; dedup is a table-engine property. Create the target as ReplacingMergeTree(version) keyed by your business key via ORDER BY, using a monotonic column as the version (an ingest timestamp, or postgres.lsn from a CDC source):

CREATE TABLE orders (id UInt64, body String, version UInt64)
ENGINE = ReplacingMergeTree(version) ORDER BY id;
-- mq-bridge just inserts rows; duplicates for the same id collapse on merge.
SELECT * FROM orders FINAL;

Read with FINAL (or argMax) to see the deduplicated result. ClickHouse also natively dedupes identical re-inserted blocks (default one-hour window) — treat that only as retry-safety, and rely on ReplacingMergeTree for logical dedup.

Branch on insert vs. duplicate (MongoDB report_outcome)

To act on whether a record was newly inserted or already existed, set report_outcome: true; the Mongo publisher tags the returned message with mongodb.outcome = inserted or existed. Wrap it in a request endpoint to forward that tagged message into a switch that routes on the outcome:

orders_upsert_branch:
  input: { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    request:
      to:
        mongodb:
          url: "mongodb://localhost:27017"
          database: "shop"
          collection: "orders"
          format: json
          id_field: "order_id"     # deterministic _id → insert-if-absent
          report_outcome: true     # → mongodb.outcome = inserted | existed
      forward_to:
        switch:
          metadata_key: "mongodb.outcome"
          cases:
            inserted: { ref: "enrich_new_order" }
            existed:  { ref: "handle_duplicate" }

report_outcome is sink-only and pairs with id_field. See also Deduplication for the middleware-based complement, the Postgres CDC tutorial for CDC-specific idempotency. When the same key can change more than once in a transaction, enable source_metadata: true and order by both mqb.src.postgres_lsn and mqb.src.postgres_ordinal; an LSN-only version cannot order those changes. The linked deduplication guide shows the complete predicate. Also see Delivery guarantees for which source/sink pairings add up to effectively-once.

Deduplication

Two complementary ways to keep replayed or retried records from creating duplicates: the deduplication middleware (filters before the sink), and the sink’s own unique constraint (the robust choice for multi-writer ETL, covered in Upserts & insert-if-absent).

The deduplication middleware

Drops messages whose ID was already seen within a TTL. Input only. Requires the dedup feature:

input:
  middlewares:
    - deduplication: { store: "sled:///var/lib/mq-bridge/dedup", ttl_seconds: 3600 }
  kafka: { topic: "orders", url: "localhost:9092" }

Picking a store

store selects the backend by URL scheme, and the scheme decides whether deduplication is process-local or shared across every instance of the route:

The collection/table defaults to mqb_dedup_<route>. Under mqb copy, whose route name is new on every run, it defaults to mqb_dedup_copy_<id>: the id is derived from source, destination and --filter without credentials, so the same command finds its keys again. A sqlite:// URL has no place for a table name and keeps the per-run default. Point a shared store at the deployment your sink already uses rather than standing up extra infrastructure. sled_path is the legacy spelling of a local sled store, equivalent to store: "sled://<path>". memory:// is the fastest store (~3x sled) but forgets every key on restart, so redeliveries after a restart go through again; it holds at most max_keys keys (default 1,000,000), evicting the oldest first.

# Shared across every instance of this route.
- deduplication: { store: "mongodb://localhost:27017/etl", ttl_seconds: 3600 }

A sled store is per-process, not cluster-wide, so for multi-writer pipelines use either a shared store above or the sink constraint (below). Even with a shared store, the sink’s own unique constraint remains the more robust choice when the sink has one — it is already the authority, with no second write.

Sink-side dedup (the robust path)

The most robust place to dedup is the sink’s own unique constraint — it’s already shared across every writer:

• MongoDB — id_field maps a business key to the unique _id; a duplicate is an idempotent skip.
• SQL — ON CONFLICT (key) DO NOTHING / ON DUPLICATE KEY UPDATE.
• ClickHouse — ReplacingMergeTree(version) collapses duplicates by sort key at merge time.
• File / object storage — no constraint, so the file name does the job: name_by: source_position names each part after the source range it holds, and a restart skips ranges already written. Needs a replayable input (Kafka, Postgres CDC, SQL cursor, MongoDB change stream, file in consume mode); object_store picks it by default for one.

output:
  object_store:
    url: "s3://my-bucket/orders"
    name_by: source_position   # part-orders-<partition>-<start>-<end>.jsonl

Full examples in Upserts & insert-if-absent and, for files, Files & object storage.

Deduplicating CDC replays

A postgres_cdc change event’s message_id is a stable hash that includes the table, key, operation, commit LSN, and intra-transaction ordinal. Replayed changes therefore deduplicate through the deduplication middleware while distinct changes in one transaction remain distinct — but only within that middleware’s reach: a replay arriving more than ttl_seconds after the original is no longer remembered, and a local sled store only remembers what this route instance saw. The sink’s own constraint (id_field / ON CONFLICT) is the durable guarantee; the middleware only saves the sink the write.

An LSN-only sink predicate is not enough when one transaction changes the same key more than once: those changes share a commit LSN, so the first accepted row can block a later row. Enable source_metadata: true on the postgres_cdc source and order sink versions by the pair (mqb.src.postgres_lsn, mqb.src.postgres_ordinal) instead. Persist both metadata values and compare the pair lexicographically in the upsert predicate:

Both columns must be typed and NOT NULL. lsn as text sorts 0/9… above 0/16…, which is backwards, and a NULL on either side makes the whole WHERE predicate NULL, so the update is skipped and the row silently stops advancing:

ALTER TABLE orders
  ADD COLUMN lsn pg_lsn NOT NULL DEFAULT '0/0',
  ADD COLUMN ordinal bigint NOT NULL DEFAULT 0;

On a table that already has these columns as text or as nullable, backfill before switching the predicate on — UPDATE orders SET lsn = '0/0' WHERE lsn IS NULL (same for ordinal), then ALTER COLUMN … TYPE pg_lsn USING lsn::pg_lsn and SET NOT NULL. A '0/0' floor means the first change event for each existing row wins, which is what a re-snapshot should do.

INSERT INTO orders (id, body, lsn, ordinal)
VALUES (${payload:id}, ${payload:body}, ${metadata:mqb.src.postgres_lsn}::pg_lsn, ${metadata:mqb.src.postgres_ordinal}::bigint)
ON CONFLICT (id) DO UPDATE
SET body = EXCLUDED.body, lsn = EXCLUDED.lsn, ordinal = EXCLUDED.ordinal
WHERE (EXCLUDED.lsn, EXCLUDED.ordinal) > (orders.lsn, orders.ordinal)

The row comparison decides on lsn and only consults ordinal when the two LSNs are equal.

See the Postgres CDC → JSONL tutorial for the full CDC idempotency picture, and Delivery guarantees for what identity each source provides and which sinks absorb a duplicate write.

Checkpoints & resumable copies

A checkpoint is a durable record of how far a source has been read, so a restart continues where the last run stopped instead of re-copying from the beginning. It is what turns a one-shot copy into a repeatable incremental sync you can put on a timer.

Two settings control it, on the source endpoint:

Positions are namespaced as <source>:<cursor_id>, so several routes may share one store without colliding. checkpoint_store may embed credentials and is treated as a secret.

Which sources checkpoint

An SQLx source with no cursor_column is a destructive work queue, not a resumable read — progress is the deletion of claimed rows, so there is nothing to checkpoint. See Endpoints.

Picking a store

checkpoint_store selects the backend by URL scheme; a value with no scheme is a plain table/collection name in the source datastore.

Notes:

• ClickHouse requires an external store. It cannot cheaply upsert cursor rows, so a source-datastore checkpoint is rejected; cursor_id without a checkpoint_store disables resume with a warning. Use --resume when a missing resumable configuration must be fatal, or --no-resume when a full re-copy is intentional and resume diagnostics should be ignored.
• Object-store sources must point checkpoint_store at a different bucket or prefix than they read — a cursor object written under the source prefix would be listed and re-read as data. The source rejects an overlapping location.
• The default name carries a hash for a source name with other characters than letters, digits and _: public.orders gets mqb_cursors_public_orders_<hash>, so it cannot share a table with a source named public_orders. A table that already exists as mqb_cursors_public_orders keeps being used.
• A file store is written atomically (temp file + rename) and concurrent writers in one process are serialized, so several routes may share one file. Use one process per file: two processes that write the same file overwrite each other’s cursors.
• Cloud object-store checkpoints need the object-store feature compiled in.

Using it with copy

Pass --no-resume to make an intentional full copy without checkpoint diagnostics. For SQL and ClickHouse, cursor_column is still required to page through the source; the flag only ignores optional cursor_id / checkpoint_store state. It currently applies to cursor-based SQL and ClickHouse sources, MongoDB change streams, and object-store sources. It does not change native queue or CDC offsets and cannot be combined with --resume; for sources such as Kafka, NATS, and Postgres CDC, the flag is intentionally a no-op.

cursor_id, cursor_column, and checkpoint_store are ordinary endpoint config fields, so on the CLI they are just query parameters on --from:

# Incremental table → table sync. Re-run it as often as you like: each run copies
# only rows whose `id` is greater than the last successfully written row.
mqb copy \
  --from 'postgres://user:pass@localhost/app?table=orders&cursor_column=id&cursor_id=orders_sync' \
  --to   'clickhouse://localhost:8123?table=orders&database=analytics' \
  --drain

# Read-only source: keep the cursor next to the job instead of in the source DB.
mqb copy \
  --from 'mysql://ro_user:pass@reporting/app?table=events&cursor_column=event_id&cursor_id=events_export&checkpoint_store=file%3A%2F%2F%2Fvar%2Flib%2Fmqb%2Fcursors.json' \
  --to   'file:///data/events.jsonl' \
  --drain

# MongoDB bulk read that survives a restart mid-copy.
mqb copy \
  --from 'mongodb://localhost:27017?database=app&collection=orders&consume=capture_all&cursor_id=orders_dump' \
  --to   'file:///data/orders.jsonl'

A checkpoint_store URL inside a URI must be percent-encoded (:// → %3A%2F%2F), since it is a query-parameter value. In a YAML config it is written plainly:

orders_sync:
  input:
    sqlx:
      url: "postgres://user:pass@localhost/app"
      table: orders
      cursor_column: id
      cursor_id: orders_sync
      checkpoint_store: "file:///var/lib/mqb/cursors.json"
  output:
    clickhouse: { url: "http://localhost:8123", table: orders, database: analytics }

Translating a copy command into a config file

Most of a copy command maps field for field: URI query parameters become endpoint config keys, --drain becomes exit_on_empty: true, and --filter becomes a filter middleware on the input. mqb to-cli goes the other way — it prints a command for a loaded config — but there is no converter in this direction, so three differences are worth knowing.

--no-resume is not a field to copy across. It does not remove cursor_id from the config; it makes the consumer ignore it. Translating the fields literally therefore reverses the behaviour — the CLI run re-copies everything, the config route resumes and copies nothing. The correct translation is to omit cursor_id, which is how a config route says the same thing:

mqb copy 'sqlite:///data/app.db?table=orders&cursor_column=id&cursor_id=nightly' \
         'file:///data/orders.jsonl' --drain --no-resume

orders_dump:
  exit_on_empty: true
  input:
    sqlx: { url: "sqlite:///data/app.db", table: orders, cursor_column: id }  # no cursor_id
  output:
    file: { path: /data/orders.jsonl }

--resume cannot be reproduced. It generates a copy-… cursor id from the credential-redacted source, destination, and filter. Re-running the same command resumes correctly, and mqb checkpoint show SOURCE TARGET finds its position, but a config route has no way to name that id, so it starts a fresh checkpoint under whatever cursor_id you choose.

The defaults differ. copy defaults to concurrency: 4 and batch_size: 1024; a route defaults to 1 and 512. Write both out explicitly, or the translated route runs at a quarter of the concurrency for no visible reason.

--drain and checkpoints

--drain exits when the source yields an empty batch — i.e. when the checkpoint has caught up with the table. That is exactly the shape you want for a cron/systemd-timer job: each invocation drains the backlog since last time and exits 0. Without --drain, the same command runs forever, polling for new rows every polling_interval_ms (100 ms by default) — or backing off exponentially up to max_polling_interval_ms while drained, if you set it.

MongoDB capture_all follows a change stream once its initial read is done, so a --drain run ends when that stream goes quiet rather than at a known end of data. capture_new only ever emits changes made after it starts and is continuous by nature. For a one-shot read with a real end, use consume: snapshot — non-destructive, no replica set, and not resumable (it rejects cursor_id).

An object-store source ends a drain run when it reaches the end of the objects it listed, which can be before the prefix is exhausted — the checkpoint makes this safe rather than lossy: each run resumes at the last fully-acked object key, so repeated runs advance until one reports zero messages. Loop the job until it moves nothing if you need a single pass to cover everything.

Inspecting and editing a checkpoint

mqb checkpoint reads and changes a stored position without touching the store by hand:

mqb -c config.yaml checkpoint show --route orders_sync     # a configured route or consumer
mqb checkpoint show 'sqlite:///data/app.db?table=orders&cursor_column=id' \
  file:///data/orders.jsonl                                 # a `copy --resume` job

value:   int:3
source:  sqlx:orders:id
updated: 2026-09-30T08:12:44.120+00:00

• reset deletes the position, so the next run starts from the beginning. It prints the old value first.
• set --value V overwrites it, e.g. to replay from a known row (int:1000) or to undo a reset. Use the format show prints. For a SQL cursor_column source a value in another form than int:<number> or str:<text> is refused.
• Stop the route or copy first. reset and set do not coordinate with a running reader: its next acknowledged batch saves its own position over the edit.
• A copy --resume job is named by the same SOURCE, TARGET and --filter it ran with.
• Postgres CDC is refused: manage its replication slot instead.

Each position is stored with the source that wrote it and a timestamp. On SQL and ClickHouse the source is the table and cursor_column, on MongoDB the database and collection. A run refuses a position written by a different source instead of resuming from a value that means something else, e.g. after cursor_column changed under the same cursor_id. show marks it MISMATCH. Reset it, or pick a new cursor_id. Positions written by mq-bridge 0.4.16 or older still load and are upgraded on the next save; an older mq-bridge cannot read the new format and starts from the beginning.

Delivery semantics

Checkpoints are at-least-once, never at-most-once:

• The position advances only after the sink acknowledges, and only across the contiguous run of acks from the front of the batch. The first nack stops the advance.
• On a partial failure the in-memory read cursor rolls back to the committed boundary, so nacked rows are re-read on the next poll rather than skipped until a restart.
• A crash between “rows written” and “checkpoint flushed” replays that batch. Make the sink idempotent — see Upserts & insert-if-absent and Deduplication; a file or object-store sink does this with name_by: source_position.
• If persisting the cursor fails, the route logs a warning and keeps running; rows may be reprocessed on restart.

Gotchas

• cursor_column must be monotonic and non-decreasing for new rows (WHERE col > $last ORDER BY col ASC). An autoincrement id or an append-only timestamp works; a mutable updated_at does not give you deletes, and a column that can go backwards loses rows.
• Cursor polling captures appends only. Updates and deletes to already-copied rows are not observed. For those, use CDC (postgres_cdc, MongoDB capture_*).
• Equal-value groups must fit in a batch. If more rows share one cursor_column value than batch_size, the reader refuses to advance rather than skipping the remainder. It reports cursor_column '…' has a group of equal values larger than batch_size and retries that poll indefinitely rather than exiting, so a --drain job hangs instead of failing — watch for the repeating log line. Raise batch_size above the largest group, or pick a more unique column.
• The cursor column must be integer or text, with more types on Postgres. Other types the SQL Any driver can’t decode fail permanently; expose the column as BIGINT/TEXT through a view. On Postgres a timestamp, date, numeric or uuid column works too.
• cursor_column and delete_after_read are mutually exclusive — one is non-destructive, the other consumes.
• Changing cursor_id or checkpoint_store starts over. The position is keyed by both; a new key means a full re-copy. Reuse the same pair to continue a sync, and give unrelated jobs distinct cursor_ids. Changing the table or cursor_column under the same key fails the run instead; see Inspecting and editing a checkpoint.
• capture_all’s initial snapshot is not incrementally checkpointed. What gets persisted is the change-stream resume token, written once streaming begins; a run interrupted during the snapshot re-snapshots from the beginning. Size the sink’s idempotency accordingly.
• ClickHouse and SQL Server: ClickHouse is polling-only and needs an external store; cursor_column mode is not supported on Microsoft SQL Server at all.

See also

• Endpoints (concepts) — read modes and CDC
• PostgreSQL parameters · MongoDB · ClickHouse · Postgres CDC
• CLI commands — copy and checkpoint flags, URI grammar

Retries & backoff

The retry middleware retries failed sends with exponential backoff. Output only.

output:
  middlewares:
    - retry: { max_attempts: 5, initial_interval_ms: 200, max_interval_ms: 10000, multiplier: 2.0 }
  kafka: { topic: "orders", url: "localhost:9092" }

From the copy CLI:

Attach the chain to --to, since retry is output-only:

--to 'kafka://localhost:9092?topic=orders|retry?max_attempts=5&initial_interval_ms=200'

What gets retried

Only Retryable and connection errors are retried; NonRetryable failures pass straight through. Once attempts are exhausted, the error is marked so a following dlq treats it as permanent.

Always pair retry with dlq

retry does not retain a message it gives up on — it hands the still-failing message on to be dropped, or to a following dlq. Put dlq after retry so it’s the outermost layer and captures the exhausted failures:

output:
  middlewares:
    - retry: { max_attempts: 3 }
    - dlq: { endpoint: { file: { path: "rejected.jsonl" } } }

Don’t over-tune

Several endpoints already retry connection/timeout errors internally, and the retry middleware adds backoff on top — so you rarely need a large max_attempts. Cap max_interval_ms to match your latency budget so retries never stall a route indefinitely. See Performance tuning → Retry & backoff.

Dead-letter queues

By default, a message that fails permanently is logged at error level and dropped — the route tolerates it and keeps going. A dlq middleware is the only mechanism that retains those failures for inspection or replay.

Basic DLQ

Add dlq to the output middleware list. The DLQ endpoint is a full endpoint, so failed messages can land anywhere — a file, another queue, a database:

output:
  middlewares:
    - dlq:
        endpoint:
          file: { path: "dead-letters.jsonl" }
  kafka: { topic: "orders", url: "localhost:9092" }

From the copy CLI, dlq’s endpoint is a URL-encoded endpoint URI:

--to 'kafka://broker:9092?topic=orders|dlq?endpoint=file%3A%2F%2F%2Ftmp%2Ffailed.jsonl'

Pair it with retry — and mind the order

retry alone does not retain a failed message; once its attempts are exhausted it hands the still-failing message on to be dropped (or to a following dlq). Put retry before dlq so the dlq is the outermost layer and captures what retry gives up on:

output:
  middlewares:
    - retry: { max_attempts: 3 }
    - dlq: { endpoint: { file: { path: "rejected.jsonl" } } }   # last = outermost

On an output, publisher middlewares wrap in list order, so the last entry is outermost. See the ordering rule.

What gets dead-lettered

• ✅ NonRetryable failures (data/type errors the sink rejects, poison payloads).
• ✅ Retryable failures whose retries are exhausted.
• ❌ Connection errors are not dead-lettered — they propagate so the route can reconnect. (If the DLQ send itself fails with a connection error, that error propagates rather than silently dropping the message.)

Catching transform rejections

A transform failure on an output is non-retryable and flows to a following dlq, so invalid records are captured with the reason:

output:
  middlewares:
    - transform: { schema_file: "schemas/user.json" }
    - dlq: { endpoint: { file: { path: "rejected.jsonl" } } }
  kafka: { topic: "users", url: "localhost:9092" }

No DLQ? Watch status and logs

Without a dlq, a systematic failure (e.g. every row hitting a column-type mismatch) drains the input while delivering nothing to the sink and still ends completed. Route status retains the drop count and last rejection cause, and the logs contain a burst of Dropping message … due to non-retryable error. Do not use the outcome alone as a delivery check. See Troubleshooting.

MySQL, SQL Server & Oracle CDC via Debezium

mq-bridge captures changes natively from PostgreSQL and MongoDB. For MySQL, MariaDB, SQL Server, Oracle and Db2, run Debezium Server next to it: Debezium reads the database’s change log and hands each change to mq-bridge, which routes, transforms and writes it like any other message.

Debezium Server is a standalone process. It needs no Kafka and no Kafka Connect.

This recipe has not been run end to end. Verify the Debezium properties against the current Debezium Server documentation.

Hand over by HTTP (no broker)

Debezium’s HTTP sink POSTs one change per request, and mq-bridge’s HTTP input receives it.

# Debezium Server: application.properties
debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.database.hostname=localhost
debezium.source.database.port=3306
debezium.source.database.user=debezium
debezium.source.database.password=${MYSQL_PASSWORD}
debezium.source.database.server.id=184054
debezium.source.topic.prefix=shop
debezium.source.table.include.list=shop.orders
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory
debezium.source.schema.history.internal.file.filename=data/schema-history.dat

debezium.sink.type=http
debezium.sink.http.url=http://localhost:8080

# Plain JSON rows instead of the schema + before/after envelope
debezium.format.key.schemas.enable=false
debezium.format.value.schemas.enable=false
debezium.transforms=unwrap
debezium.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
debezium.transforms.unwrap.delete.tombstone.handling.mode=rewrite

mysql_orders_to_postgres:
  input:
    http: { url: "127.0.0.1:8080" }
  output:
    sqlx:
      url: "postgres://user:pass@localhost:5432/warehouse"
      table: "orders"
      insert_query: >-
        INSERT INTO orders (id, status, amount)
        VALUES (${payload:id}, ${payload:status}, ${payload:amount})
        ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, amount = EXCLUDED.amount

• The unwrap transform makes each payload the row itself ({"id": 7, "status": "open", …}). Without it the row is under after, inside Debezium’s change envelope.
• A failed write answers the request with HTTP 500 and Debezium sends the change again, so delivery is at-least-once. The upsert makes the repeat harmless; see Upserts & insert-if-absent.
• Debezium keeps the log position in its offset file. Keep that file and the schema history on durable storage, or a restart re-reads the initial snapshot.
• For another database, change connector.class and the database.* properties. The mq-bridge route stays the same.

Hand over through a broker

If Debezium already publishes to Kafka, or mq-bridge should be able to stop without holding Debezium up, read the change topic instead. Debezium names it <topic.prefix>.<database>.<table>:

input:
  kafka: { url: "localhost:9092", topic: "shop.shop.orders", group_id: "mqb-orders-sync" }

Debezium Server also sinks to NATS JetStream, Redis Streams, Pulsar and RabbitMQ, each of which mq-bridge reads with its native connector.

Deletes

With delete.tombstone.handling.mode=rewrite, a delete arrives as the last known row with "__deleted": "true"; every other change carries "__deleted": "false". The upsert above would write a deleted row back, so send deletes to their own statement with a switch, or drop them with a filter.

Transform & schema mapping

The transform middleware reshapes JSON payloads declaratively, so renaming fields and fixing types doesn’t need a custom handler. It runs two optional stages over a single parse: mapping (rename, move, nest) then schema (coerce, apply defaults, validate). Input and output.

Rename + type-fix a CSV feed

csv_to_kafka:
  input:
    file: { path: "users.csv", format: csv }
  output:
    middlewares:
      - transform:
          mapping:
            firstName: "$.first_name"
            lastName: "$.last_name"
            id: "$.user_id"
            "address.city": { path: "$.city", default: "unknown" }
          schema_file: "schemas/user.json"
      - dlq:
          endpoint: { file: { path: "rejected.jsonl" } }
    kafka: { topic: "users", url: "localhost:9092" }

With the CSV producing {"first_name":"John","last_name":"Smith","user_id":"42"}, the mapping yields {"firstName":"John","lastName":"Smith","id":"42","address":{"city":"unknown"}} ($.city is absent, so address.city falls back to its default) and the schema then coerces id to the integer 42.

• Paths accept $.field, $.a.b, $.items[0] (the $. prefix is optional). Dots in the output key nest the result. An absent optional source field is omitted, not emitted as null.
• Paths have no quoting: a CSV column whose header contains . or [ (e.g. price.usd) cannot be addressed by a mapping.
• Coercions are the lossless ones only: string → integer, string → number, string → boolean, number → string. string → number keeps integers exact within the i64/u64 range ("42" → 42); larger integers and fractional text go through f64 (~17 significant digits) and may lose precision.

Decode an embedded JSON string

A field carrying a JSON document as a string is decoded via contentMediaType (JSON Schema 2020-12) — opt-in per field, never done by plain coercion:

- transform:
    schema:
      type: object
      properties:
        payload:
          type: string
          contentMediaType: application/json
          contentSchema:
            type: object
            properties:
              qty: { type: integer }

The string is replaced by the parsed document, and contentSchema (if given) is applied with the same coercion/defaults/validation — so an inner qty: "7" arrives as 7.

Handling failures

Failures are non-retryable and name the field, e.g. transform failed at $.items[1].qty [coercion]: cannot coerce string "oops" to integer.

• On an output, the message is failed so a following dlq captures it.
• On an input, it’s dropped from the batch and acknowledged, keeping invalid data out.
• on_error: pass_through instead forwards the original payload with the reason in the mqb.transform_error metadata key — which a switch can route on.

Full option list, the supported schema subset, and the exact coercion rules are in the middleware reference → transform.

Content-based routing (switch)

The switch structural endpoint picks one destination per message. Output only. It has two modes and uses exactly one of them: value lookup on a metadata key, or when predicates over the payload. Naming both is a startup error.

output:
  switch:
    metadata_key: "country_code"
    cases:
      US: { kafka: { topic: "us_orders", url: "kafka-us:9092" } }
      EU: { nats:  { subject: "eu_orders", url: "nats-eu:4222" } }
    default: { file: { path: "/var/data/unroutable_orders.log" } }

A message whose key is missing or unmatched goes to default; without a default it is dropped.

From the CLI, the same endpoint is a URI: switch:?metadata_key=country_code&case.US=<uri>&default=<uri> — see Structural endpoints in the URI.

Route on the payload (when)

when takes an ordered list of predicates and sends the message to the first one that matches. The expression language is the one --filter uses: payload fields by bare name including nested paths, metadata under meta., and and / or.

output:
  switch:
    when:
      - if: "amount > 10000"
        to: { kafka: { topic: "large_orders", url: "kafka:9092" } }
      - if: "order.status == 'refunded'"
        to: { nats: { subject: "refunds", url: "nats://localhost:4222" } }
    default: { file: { path: "/var/data/orders.jsonl" } }

A predicate parses the payload, while value lookup is a HashMap get on metadata — which is why the modes cannot be mixed in one endpoint, and why a metadata key you already have is the cheaper branch. A message matching no predicate goes to default, and without a default it is dropped, exactly as in value-lookup mode.

The CLI spells this as when=<expression> / to=<uri> pairs: switch:?when=amount > 10000&to=<uri>&default=<uri>.

Promote a payload value into metadata

Value-lookup mode matches on metadata, not payload fields. When you want that mode — an exact-match table rather than predicates — promote the value into metadata first. Two common ways:

• An endpoint that already emits a status key — e.g. http_status_code from an HTTP request, or mongodb.outcome from a Mongo upsert (see Upserts).
• transform with on_error: pass_through, which sets mqb.transform_error on failed records so you can shunt them aside:

output:
  middlewares:
    - transform: { schema_file: "schemas/order.json", on_error: pass_through }
  switch:
    metadata_key: "mqb.transform_error"
    cases: {}                                  # (no exact-match cases)
    default: { kafka: { topic: "orders", url: "localhost:9092" } }

Split HTTP responses by status

Pair switch with request, which forwards a response (or, on error/timeout, the original message) tagged with a status key:

output:
  request:
    to: { http: { url: "https://api.internal/score" } }
    forward_to:
      switch:
        metadata_key: "http_status_code"
        cases:
          "200": { nats: { subject: "ok", url: "nats://localhost:4222" } }
          "404": { file: { path: "not-found.jsonl" } }
        default: { file: { path: "other.jsonl" } }

See also Fan-out to send to all destinations instead of picking one.

Enrichment & lookups

The lookup middleware asks another endpoint for each message and writes the answer into the payload. Use it to join an event with master data, or to add features such as a previous value or a moving average (for counters and averages that need no database, see running aggregates). The endpoint asked must answer: http, static, mongodb with find, sqlx or clickhouse with lookup_query, nats / memory with request_reply: true, or grpc to an mq-bridge grpc input whose route replies. For a model’s answer or an embedding, see LLM calls in a pipeline.

Enrich, then route on the result

On an output, the enriched message goes on to the next endpoint. lookup.found tells a switch whether the record exists:

orders_enrich:
  input: { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    middlewares:
      - lookup:
          from:
            sqlx:
              url: "postgres://localhost/crm"
              table: "customers"
              lookup_query: "SELECT id, name, tier FROM customers WHERE id IN (${payload:customer_id}::int)"
          into: customer
    switch:
      metadata_key: "lookup.found"
      cases:
        "true":  { kafka: { topic: "orders.enriched", url: "localhost:9092" } }
        "false": { kafka: { topic: "orders.unknown_customer", url: "localhost:9092" } }

Because the key sits inside IN (…), one query answers the whole batch and rows are matched back to messages by id. Write WHERE id = ${payload:customer_id} instead for one query per message. A failed lookup fails only its message; list retry / dlq after lookup to handle it.

Several lookups, before the handler

On an input, the handler already sees the enriched message. Lookups in entries run in parallel, so a batch waits for the slowest one, not the sum:

input:
  kafka: { topic: "payments", url: "localhost:9092" }
  middlewares:
    - lookup:
        concurrency: 64
        entries:
          - into: features.user
            from:
              mongodb:
                url: "mongodb://localhost:27017"
                database: "features"
                collection: "users"
                find: '{"_id": "${payload:user_id}"}'
          - into: features.card_avg
            from:
              sqlx:
                url: "postgres://localhost/payments"
                table: "payments"
                lookup_query: >-
                  SELECT avg(amount)::text AS avg20 FROM (SELECT amount FROM payments
                  WHERE card_id = ${payload:card_id} ORDER BY ts DESC LIMIT 20) t

Each entry sets lookup.<into>.found; lookup.found is true only when all of them found something. On an input a temporary error nacks the whole batch, so the source redelivers it, and reconnects the route. A permanent error (non-JSON payload, HTTP 4xx, a query error) is logged and drops only its message.

Postgres gotchas

• Cast placeholders to the column type: WHERE id = ${payload:author_id}::int. A field missing from the payload, e.g. on a CDC delete, is otherwise bound as text and fails.
• Cast NUMERIC, TIMESTAMPTZ and similar result columns to ::text. They arrive as JSON strings.
• A per-message lookup_query returns the first row only; use ORDER BY … LIMIT 1 for “the latest”. A batched (IN) query takes no plain LIMIT.

LLM calls in a pipeline

Two endpoints call an OpenAI-compatible API: openai_chat sends one chat completion per message, openai_embeddings one embeddings request per batch. Both are used as the from of a lookup, which writes the answer into the message. The message then goes on to any output, with the retries, dead-letter queue and back-pressure of an ordinary route.

Chat, embeddings and the Qdrant upsert were run against Ollama (qwen2.5:0.5b, all-minilm) and Qdrant 1.19, with files in place of Kafka and Postgres. OpenAI itself was not called.

Classify, then route on the answer

With json: true the answer is parsed, so a switch can read its fields:

tickets_triage:
  input: { kafka: { topic: "tickets", url: "localhost:9092" } }
  output:
    middlewares:
      - lookup:
          concurrency: 8
          from:
            openai_chat:
              api_key: "${env:OPENAI_API_KEY}"
              model: gpt-4o-mini
              json: true
              system: 'Answer as JSON: {"urgent": boolean, "topic": string}'
              prompt: "${payload:subject}\n\n${payload:body}"
              options: { temperature: 0 }
          into: triage
      - retry: { max_attempts: 5 }
      - dlq:
          endpoint: { kafka: { topic: "tickets.failed", url: "localhost:9092" } }
    switch:
      when:
        - if: "triage.urgent == true"
          to: { kafka: { topic: "tickets.urgent", url: "localhost:9092" } }
      default: { kafka: { topic: "tickets.normal", url: "localhost:9092" } }

retry repeats only the messages whose request failed, so answers already paid for are kept. An answer that is not JSON fails its message for good and goes to the dlq.

Summarise, and keep only the answer

As the to of a request endpoint, the model’s answer replaces the message and is forwarded:

articles_summarise:
  input: { file: { path: "articles.jsonl", format: raw } }
  output:
    request:
      to:
        openai_chat:
          url: "http://localhost:11434"
          model: llama3.2
          prompt: "Summarise in one sentence: ${payload:text}"
      forward_to: { file: { path: "summaries.jsonl", format: raw } }

A request endpoint sends one message at a time, so this form makes one API request per message. For messages_per_request and batched embeddings use a lookup, or put a buffer on the to endpoint (middlewares: [{ buffer: { max_messages: 500, max_delay_ms: 0 } }] next to openai_chat), which hands the sends of a batch to the endpoint together.

On the command line the endpoint is openai://host?model=…&prompt=…. openai://host is plain HTTP, for a local server such as Ollama; openai+https://host is HTTPS. Without a url the endpoint calls https://api.openai.com/v1.

Bundle messages into one request

A short classification repeats the same instructions for every message. messages_per_request sends several messages as numbered inputs of one completion and gives each message its own answer, so the system message is paid for once per request:

reviews_scored:
  batch_size: 100
  input: { kafka: { topic: "reviews", url: "localhost:9092" } }
  output:
    middlewares:
      - lookup:
          from:
            openai_chat:
              api_key: "${env:OPENAI_API_KEY}"
              model: gpt-4o-mini
              system: "Answer with one word: positive, negative or neutral."
              prompt: "${payload:text}"
              messages_per_request: 20
          into: sentiment
      - retry: { max_attempts: 5 }
    kafka: { topic: "reviews.scored", url: "localhost:9092" }

Each answer carries the number of its input, so the order the model answers in does not matter. Answers that cannot be matched are asked again in halves, so a model that loses count costs more, not less. Put the task into system and only the data into prompt. What to expect is on the connector page.

Embed a table into a vector store

One embeddings request covers a whole batch. The vector lands in the message, and a transform shapes the point the store expects. Output middlewares run bottom-up.

docs_to_qdrant:
  input:
    sqlx:
      url: "postgres://user:pass@localhost/app"
      table: docs
      cursor_column: id
  output:
    middlewares:
      - transform:
          mapping:
            id: "$.id"
            vector: "$.vector"
            "payload.title": "$.title"
      - lookup:
          from:
            openai_embeddings:
              url: "http://localhost:11434"
              model: all-minilm
              input: "${payload:title}: ${payload:body}"
          into: vector
      - retry: { max_attempts: 5 }
    qdrant: { url: "http://localhost:6333", collection: docs }

To follow changes and deletes as well, read the table with postgres_cdc; the Qdrant page has that route.

From Rust

Each endpoint has a typed configuration that converts into an Endpoint:

use mq_bridge::endpoints::http_bulk::OpenAiChatConfig;
use mq_bridge::models::Endpoint;

let llm: Endpoint = OpenAiChatConfig {
    model: "gpt-4o-mini".into(),
    prompt: "Summarise: ${payload:text}".into(),
    api_key: std::env::var("OPENAI_API_KEY").ok(),
    ..Default::default()
}
.into();

It needs the http-bulk feature and no registration.

Cost and limits

• Every message is a request. For chat, the lookup’s concurrency (default 16) is the number of requests in flight. Lower it, or add a limiter, to stay under a provider’s rate limit. messages_per_request puts several messages into one request instead.
• A 429 is retryable. With retry the route slows down instead of losing messages; the input is not acknowledged until the answer is in.
• Nothing is cached. A message that is delivered twice is sent, and billed, twice. Put deduplication on the input if that matters.
• Prompts are escaped. Text taken from a message cannot break the request body. It can still contain instructions to the model; treat the answer as untrusted input.

Stream tokens

These endpoints wait for the whole answer. For a token stream, call the API with the http output and stream_response_to, which publishes each server-sent event as its own message.

Other providers

Of these, only Ollama was called.

Running aggregates

The aggregate middleware keeps one state per key inside the process and updates it with every message: counters, sums, moving averages. The message goes on with the state written into its payload. Use it for values that depend on earlier messages, such as “how many readings did this sensor send so far” or “what is this sensor’s usual reading”. No database is asked, so it is much faster than a lookup that lets a database keep the counter.

Where it fits

mq-bridge moves messages, and a few middlewares keep something about them between messages: deduplication the keys it has seen, aggregate a value computed from them. It is meant for a value the pipeline needs while the message is on its way: to filter, route or alert on it, to hand it to the receiver, or to thin out what reaches an expensive sink. It is kept small on purpose: one state per key, updated message by message, in the order the messages arrive.

Letting the database keep the counter is the obvious alternative, and the slow one: it costs a read and a write per message and key. In our measurements a lookup that updated five keys per message reached about 17k msg/s on PostgreSQL and about 2k on MongoDB. With single_writer, aggregate writes a changed state once per flush instead of once per message and reaches 240k on the same PostgreSQL (see Performance).

Windows, event time and joins are out of scope. They need the state and the input position committed together, which a bridge between arbitrary endpoints cannot promise; see How this compares to stream processors.

A counter and a sum per key

sensor_stats:
  input:
    kafka: { topic: "readings", url: "localhost:9092" }
    middlewares:
      - aggregate:
          consistency: single_writer
          key: "${payload:sensor_id}"
          into: sensor
          expression: "{ total: (state.total ?? 0) + reading, count: (state.count ?? 0) + 1 }"
  output:
    kafka: { topic: "readings.enriched", url: "localhost:9092" }

{"sensor_id": 7, "reading": 20} leaves as {"sensor_id": 7, "reading": 20, "sensor": {"total": 20, "count": 1}}. The next message for sensor 7 continues from that state.

consistency: single_writer without a store says that the states may live in this process only: they are gone after a restart and one instance runs the route. Leaving out both is a startup error, so that states in memory are never the result of a forgotten store.

A CSV sink writes that nested result as the columns sensor.total and sensor.count, not as JSON text in one cell; see CSV.

expression returns the new state. It reads payload fields by name, metadata as meta.<key>, and state, the value it returned for the previous message with the same key. For a new key state is null, so state.total ?? 0 gives the starting value. The language is the one transform and filter use.

The usual form avg * 0.99 + reading * 0.01 starts at the first reading and needs hundreds of messages to forget it. Keep the weighted sum and the weight, and divide when writing:

- aggregate:
    consistency: single_writer
    key: "${payload:sensor_id}"
    into: sensor.avg_reading
    expression: "{ s: (state.s ?? 0) * 0.99 + reading, w: (state.w ?? 0) * 0.99 + 1 }"
    output: "state.s / state.w"

output shapes what the message carries; the stored state stays {s, w}. After a first reading of 500 followed by readings of 50, this reads 52.6 after 100 messages. The usual form still reads 216.

Compare a message with what came before

With emit: previous the message carries the state from before its own update. That is what a rule such as “this reading is five times the sensor’s average” needs:

- aggregate:
    consistency: single_writer
    key: "${payload:sensor_id}"
    into: sensor.avg_before
    emit: previous
    expression: "{ s: (state.s ?? 0) * 0.99 + reading, w: (state.w ?? 0) * 0.99 + 1 }"
    output: "state.s / state.w"

For the first message of a sensor sensor.avg_before is null.

A fixed factor is a rough rule. With the built-in aggregates (see below) the message can carry the average and the standard deviation from before its own update, and a filter or switch behind it can ask how many deviations the reading is away:

- aggregate:
    consistency: single_writer
    key: "${payload:sensor_id}"
    into: before
    emit: previous
    fields:
      avg: ema(reading, 0.01)
      sd: ema_stddev(reading, 0.01)

stddev and variance cover all readings of the key; the ema_ forms weight them like ema, so old readings fade and the result follows a sensor that drifts. Both are sample values (divided by n − 1) and null for the first message of a key, which has no spread yet.

Several keys from one message

entries updates one state per key, each with its own expression:

- aggregate:
    consistency: single_writer
    entries:
      - key: "${payload:sensor_id}"
        into: stats.sensor
        expression: "{ sum: (state.sum ?? 0) + reading, n: (state.n ?? 0) + 1 }"
      - key: "${payload:site_id}"
        into: stats.site
        expression: "{ max: max([state.max ?? reading, reading]) }"

Built-in aggregates: the fast path

For counters, sums and averages you do not need an expression. fields names the results and what each one computes:

- aggregate:
    consistency: single_writer
    entries:
      - key: "${payload:sensor_id}"
        into: sensor
        fields:
          n: count
          total: sum(reading)
          avg: ema(reading, 0.01)
      - key: "${payload:site_id}"
        into: site
        fields: { high: max(reading), usual: mean(reading) }

Available are count, sum, min, max, last, mean, stddev, variance, and ema(path, alpha), ema_stddev(path, alpha), ema_variance(path, alpha), and the time-based decayed_count(half_life) and decayed_sum(path, half_life). ema is the moving average from above: it has no start bias, so the first message reads its own value. The value may be a number or a numeric string, as a CSV source delivers it. fields is several times faster than an expression; use an expression where you need logic that the built-ins do not cover.

A moving average by time

ema(reading, 0.01) counts messages: a sensor that reports every second forgets a value a thousand times faster than one that reports every quarter of an hour. Give a half-life instead of alpha and the weight follows the time between two messages of a key:

- aggregate:
    consistency: single_writer
    key: "${payload:sensor_id}"
    into: sensor
    time: measured_at
    fields:
      avg: ema(reading, 5m)
      spread: ema_stddev(reading, 1h)

A reading counts half after five minutes, a quarter after ten. time names the payload field with the message’s time: epoch seconds or milliseconds, or an RFC 3339 text. Without time the clock of the process is used, and a replay then computes different values.

A message that is older than the newest one its state has seen counts with the weight it would have had by then: half per half-life it is late. The averages and decayed values therefore come out the same in any arrival order. last keeps the newest value. The message carries the metadata mqb.aggregate.late, which names the entries concerned. Filter or route on it where late data must not count at all:

- filter: 'meta["mqb.aggregate.late"] == null'

A wrong time far in the future makes every later message late. Send one message with the metadata mqb.aggregate.overwrite: true to set the state’s clock back to its time.

Rising or falling activity

decayed_count(half_life) counts messages that fade: one counts half after a half-life. At a steady rate it settles at about the messages of the last 1.44 half-lives, without a window. decayed_sum(path, half_life) does the same for a value. A short and a long half-life side by side compare the recent activity of a key with its own usual level:

- aggregate:
    consistency: single_writer
    key: "${payload:customer_id}"
    into: orders
    time: created_at
    fields:
      last_hour: decayed_count(1h)
      last_week: decayed_count(7d)
      amount_hour: decayed_sum(amount, 1h)
      amount_week: decayed_sum(amount, 7d)

last_hour × 168 well above last_week means far more orders than this customer usually places.

Messages with missing fields

A message that lacks a field an entry reads, or a key, cannot be folded. on_error says what happens then:

- aggregate:
    consistency: single_writer
    on_error: skip
    entries:
      - { key: "${payload:sensor_id}", into: sensor, fields: { avg: "ema(reading, 0.01)" } }
      - { key: "${payload:site_id}", into: site, fields: { high: max(reading) } }

A message without site_id leaves with sensor filled, without site, and with the metadata mqb.aggregate.skipped: site.

Try a configuration without changing the states

read_only: true reads the states from the store, computes and writes the result into the message, and stores nothing. Use it for a dry run against production states, or for a second route that only reads what another one maintains. With emit: previous the message carries the stored state as it is; with emit: updated what it would become.

- aggregate:
    store: "postgres://localhost/telemetry/sensor_states"
    read_only: true
    key: "${payload:sensor_id}"
    into: sensor
    fields: { n: count, avg: "ema(reading, 0.01)" }

Keep the states across restarts

Without store the states are gone after a restart. Give the middleware a database and they are kept there:

- aggregate:
    store: "postgres://localhost/telemetry"
    key: "${payload:sensor_id}"
    into: sensor
    fields: { n: count, avg: "ema(reading, 0.01)" }

PostgreSQL, SQLite (sqlite:///var/lib/mqb/agg.db?mode=rwc) and MongoDB (mongodb://localhost/telemetry) work. The table or collection is created on start and is named mqb_aggregate_<route> unless the URL names one.

By default (consistency: shared) several instances of the route can run against the same store: every batch loads its states, updates them and writes them back, and a state that another instance changed in between is computed again. That is safe and bound by the database, about two round trips per batch, so use a large batch_size (4096 to 8192).

If exactly one instance runs the route, consistency: single_writer is several times faster: the states stay in memory and are written to the store continuously. A batch is still only acked once its states are stored, so no update is lost on a crash.

How it works internally

Folding a batch

The middleware works on whole batches. For each message, in order, and for each entry:

1. The key template is rendered from the message.
2. The state of that key is looked up in a hash map; every entry has its own map.
3. The new state is computed, by the expression or by the built-in aggregates.
4. The state (or the previous one, or what output returns) is written into the payload at into. Fields the middleware does not write are copied byte for byte.

A message that fails in any step changes no state. Without a store this is everything: the maps live in the process and the message goes on at once.

What a store holds

One row per key, in one table for all entries of the middleware:

MongoDB holds the same as a document {_id, state, v}. Every write names the version the state was computed from and only takes effect while the row still has that version. This is optimistic concurrency: nobody locks a state while computing, and a writer that was overtaken notices it when it writes.

shared: the store owns the states

Per batch:

1. The distinct keys of the batch are collected and loaded with one query (WHERE agg_key IN (…), 1000 keys per statement), each with its version.
2. The batch is folded in memory, as above.
3. Every changed state is written back with the version it was loaded with. PostgreSQL and SQLite use one transaction of upserts, so the states of a batch land completely or not at all. MongoDB uses one bulk command; each state is atomic, the batch is not.
4. If a write is refused, another instance changed that key in between. The states are loaded again and the affected messages are folded again from them. MongoDB repeats only the refused keys; PostgreSQL and SQLite roll back and repeat the whole batch. After 32 attempts the batch fails and is redelivered.
5. Only then do the messages go on. Nothing stays in memory for the next batch.

One instance runs its batches against the store one after the other, so it does not collide with itself. The cost is two round trips per batch whatever its size, which is why large batches pay off and why fields versus expression hardly matters here.

Between instances there is no order: of two batches that change the same key, the one that writes first wins and the other computes again on top of it. Counters and sums come out right either way; a moving average depends on which batch was first.

single_writer: memory owns the states

1. A key is loaded from the store the first time a batch needs it. After that its state stays in memory, until max_keys pushes it out.
2. The batch is folded in memory, the changed keys are marked, and the batch gets a number.
3. The messages go on at once; they do not wait for the store.
4. A background task writes the marked states. As soon as one write returns it takes everything changed in the meantime, so under load one write covers many batches, and a key that changed a hundred times is written once. This is where the speed comes from.
5. The ack of a batch (on an output: the return of send) waits until a write has covered its number.

The writes carry versions here too. A refused write can only mean that somebody else writes the same states, and is treated like a failed write: the states in memory are dropped, every batch not yet stored is nacked, and the redelivered batches load their states again from the store. So a second instance is noticed, but not prevented; nothing fences it off.

What happens on a crash or a retry

Storing a state and acking a message are two steps, not one transaction. A message is never acked before its state is stored, so no update is lost. The other way round is possible: the state is stored, then the process dies or a later step fails, and the message is delivered again. It then counts twice. A deduplication in front narrows that window, but it has a store of its own and does not close it completely.

How this compares to stream processors

The building blocks are common ones: keyed state folded per message, as in Kafka Streams or Flink; versioned conditional writes for shared; write-behind with one write for many batches and an ack that waits for it for single_writer. Three things those systems do are not done here, and they decide whether the middleware fits:

• No exactly-once. Kafka Streams and Flink commit the state and the input position together. Here a redelivered message counts twice, as described above.
• No key ownership. They assign each key to exactly one worker and move the state when workers come and go. Here you choose: shared lets every instance write every key and pays a database round trip per batch, single_writer is fast and leaves it to you to run one instance.
• No windows and no expiry by time. There is no “sum of the last hour” and no state that ends with a window or a time to live. A moving average stands in for a window, by message count or by time. States leave memory only when max_keys is reached, and rows in a store are kept for good. For fixed windows you can put the window into the key (key: "${payload:sensor_id}:${payload:hour}") and use a store with shared, which keeps nothing in memory; the rows of past windows stay until you delete them.

If you need one of these, compute the aggregate in the database with a lookup, or use a stream processor for that step.

What to know before relying on it

• Without store, state lives in memory only. A restart starts from empty states and two instances do not share them. That is why it has to be asked for with consistency: single_writer.
• Memory is bounded. An entry keeps at most max_keys states in memory, one million unless you set it. See Memory for what happens beyond that.
• single_writer with two instances is wrong. The second writer is noticed as a failed write and its batches are redelivered, but do not rely on that: run one instance.
• A replayed message counts twice. Where that matters, let deduplication see the message first: on an input, list it after aggregate, because a message passes the last middleware of an input first.
• The table is named after the route. Renaming a route, or a consumer that has no id, starts with empty states. Name the table in the URL (postgres://host/db/sensor_states) to keep them.
• Order matters for a moving average. Messages of a batch are folded in order. With a route concurrency above 1 the order across batches is not guaranteed. With time, a message older than its state is weighted by its lateness and marked mqb.aggregate.late, not reordered; ema(path, alpha) and expressions still depend on the order.
• A message that cannot be folded fails alone and changes no state: its payload is not a JSON object, its key has no value, a field is missing, or an expression fails. On an input it is logged and dropped, and the source is told it succeeded; on an output only that message fails, so a following dlq can keep it. on_error: fail or skip changes that, see Messages with missing fields.
• Field order. The fields of a written state object are not in the order the expression lists them.

Performance and tuning

Measured on one Apple M1 core, one route worker, states in memory unless a store is named. Read the numbers as proportions; absolute values depend on the machine and payload.

The update itself costs about 0.3 µs per message with fields and about 2 µs with an expression. The same counter, sum and moving average built from Redpanda Connect’s branch, cache and mapping processors cost about 8 µs per message on the same machine.

What to change, in the order it pays off:

1. Use fields where the built-ins cover it. It is four to seven times faster than an expression, because no expression runs and the payload is not parsed into a tree. Keep expressions for the entries that need logic.
2. With a store, pick the consistency first. single_writer is about 14 times faster than shared, but only correct with one instance. Under shared the database sets the pace, and fields versus expression hardly matters.
3. With a store, raise batch_size to 4096 or 8192. A batch costs about two round trips under shared whatever its size. Without a store the batch size has little effect.
4. Update fewer keys per message. The cost grows with the number of entries, not with the number of aggregates in one entry: five keys cost about three times as much as one.
5. Do not raise concurrency for speed if the result depends on order, as a moving average does. Counters and sums stay correct; under shared, workers that update the same keys collide and compute those states again.
6. Mind the neighbours. A deduplication keyed on a payload field adds about 0.5 µs per message, on message_id about 0.2 µs. A transform before or after parses the payload again; let aggregate write the fields you need instead of reshaping afterwards.

Memory

An entry keeps at most max_keys states in memory, one million unless you set it. When it is full, the half that was used least recently is dropped, so a key that keeps coming stays and a key that went quiet goes. What “dropped” means depends on where the states live:

Without a store a warning is logged the first time states are forgotten. If the number of keys is bounded, such as sensors or customers, and below the limit, nothing is ever dropped. If it is unbounded, such as a session id, the limit is what keeps the process alive for weeks: set max_keys to what you can afford, or use a store so that no state is lost.

- aggregate:
    consistency: single_writer
    key: "${payload:session_id}"
    into: session
    max_keys: 200000
    fields: { n: count }

A fields state takes roughly 100 to 300 bytes, so the default limit stays within a few hundred megabytes per entry; an expression state with a large object takes more. max_keys: 0 removes the limit. Rows in a store are never removed; delete them yourself if the key is unbounded.

Fan-out

The fanout structural endpoint publishes each message to every listed endpoint. Output only. The value is a plain list — each branch may have its own middleware and may itself be structural.

output:
  fanout:
    - kafka: { topic: "audit", url: "localhost:9092" }
    - file: { path: "audit.jsonl" }
    - nats: { subject: "audit", url: "nats://localhost:4222" }

All branches receive the same message. Use it to mirror a stream to an archive while it also flows to its primary sink, to tee traffic to an audit log, or to emulate a static subscriber set (one branch per subscriber).

Fan-out vs. switch

• fanout → every destination gets the message.
• switch → one destination, chosen by a metadata key.

Give one branch its own reliability

Because each branch is a full endpoint, you can wrap just the fragile one with retry/dlq while the others stay plain:

output:
  fanout:
    - kafka: { topic: "orders", url: "localhost:9092" }
    - middlewares:
        - retry: { max_attempts: 5 }
        - dlq: { endpoint: { file: { path: "audit-failed.jsonl" } } }
      http: { url: "https://audit.internal/ingest", method: "POST" }

From the CLI

copy can build a fan-out too — fanout:?to=<uri>&mirror=<uri>, where a mirror branch’s response and failures are discarded. See Structural endpoints in the URI.

SMS & email notifications

mq-bridge has no SMS or email endpoint of its own. It sends notifications through the service that delivers them, using one of three outputs:

Whichever you pick, the message goes to the provider with the retries, dead-letter queue and back-pressure of an ordinary route.

http or http_bulk

Both send to a fixed URL, so the recipient goes in the body, not in the path. transform builds objects but not arrays: an API that wants "to": [{"email": …}] from http needs the payload in that shape already, or an http_bulk document template. Neither output sends form-encoded bodies, so an API that takes only those (Twilio’s own) is reached through a gateway.

Through a notification gateway

Apprise API takes one JSON request and passes it to email (SMTP, SES, SendGrid, …), SMS (Twilio, Vonage, SNS, …) or chat. Which services and recipients a message reaches is configured in Apprise, so the route stays the same when the provider changes.

With the recipients stored in Apprise under the key alerts, each with a tag such as ops or billing, the message picks its audience by tag:

alerts_notify:
  input: { kafka: { topic: "alerts", url: "localhost:9092" } }
  output:
    middlewares:
      - transform:
          mapping:
            tag: "$.team"
            title: "$.subject"
            body: "$.text"
      - retry: { max_attempts: 5, initial_interval_ms: 500 }
      - dlq:
          endpoint: { file: { path: "undelivered-alerts.jsonl" } }
    http:
      url: "http://apprise:8000/notify/alerts"
      custom_headers:
        Content-Type: "application/json"

A message {"team": "ops", "subject": "Disk full", "text": "db-1 at 97%"} becomes {"tag": "ops", "title": "Disk full", "body": "db-1 at 97%"}. Set Content-Type yourself: the http output sends none of its own.

ntfy is the lighter choice for push notifications to phones: the payload is the message text, and the topic is the path of url.

Through the provider’s API

One notification per request: http

An API Gateway of your own in front of SES, SNS or a Lambda, or a mail API such as Postmark’s /email, takes one notification per request. Put the key in a header and shape the body with a transform:

receipts_mail:
  input: { kafka: { topic: "receipts", url: "localhost:9092" } }
  output:
    middlewares:
      - transform:
          mapping:
            To: "$.email"
            Subject: "$.subject"
            TextBody: "$.text"
            From: { path: "$.from", default: "shop@example.com" }
      - retry: { max_attempts: 5, initial_interval_ms: 500 }
      - dlq:
          endpoint: { file: { path: "unsent-receipts.jsonl" } }
    http:
      url: "https://api.postmarkapp.com/email"
      custom_headers:
        Content-Type: "application/json"
        Accept: "application/json"
        X-Postmark-Server-Token: "${env:POSTMARK_TOKEN}"

From falls back to the default when the message has none. For an API Gateway the same shape applies, with its x-api-key header in place of the token. A gateway that checks IAM signatures cannot be called this way: http does not sign requests; use http_bulk or put an API key or a Lambda authorizer on the gateway.

Many notifications per request: http_bulk

SendGrid sends one template to up to 1,000 recipients per request, each with data of its own. With http_bulk each message becomes one recipient, and the message itself is the template data:

welcome_mails:
  batch_size: 1000
  input: { kafka: { topic: "signups", url: "localhost:9092" } }
  output:
    middlewares:
      - retry: { max_attempts: 5, initial_interval_ms: 1000 }
      - dlq:
          endpoint: { file: { path: "unsent-welcome.jsonl" } }
    http_bulk:
      url: https://api.sendgrid.com
      headers:
        Authorization: "Bearer ${env:SENDGRID_API_KEY}"
      upsert:
        path: /v3/mail/send
        format: json_array
        envelope: '{"from": {"email": "hello@example.com"}, "template_id": "d-0123456789abcdef", "personalizations": {documents}}'
        document: '{"to": [{"email": "${payload:email}"}], "dynamic_template_data": {document}}'

A batch of signups {"email": "ada@example.com", "name": "Ada"} is one request whose personalizations hold one entry per signup. SendGrid answers the request as a whole, so a refused request fails every message in it; the route’s batch_size must stay within the API’s limit, since http_bulk has no limit of its own on documents per request.

Amazon SES is called the same way, with requests signed by the AWS credentials of the environment instead of a header. SendBulkEmail takes up to 50 recipients and reports each one, so result.items fails only the recipients SES refused:

    http_bulk:
      url: https://email.eu-central-1.amazonaws.com
      auth:
        aws_sigv4: { region: eu-central-1, service: ses }
      upsert:
        path: /v2/email/outbound-bulk-emails
        format: json_array
        envelope: '{"FromEmailAddress": "hello@example.com", "DefaultContent": {"Template": {"TemplateName": "welcome", "TemplateData": "{}"}}, "BulkEmailEntries": {documents}}'
        result:
          items:
            path: /BulkEmailEntryResults
            error: /Error
            success: { path: /Status, equals: SUCCESS }

SES may leave out Error for a refused recipient, so success fails every entry whose Status is not SUCCESS.

Here every payload must already be one SES entry, such as {"Destination": {"ToAddresses": ["ada@example.com"]}, "ReplacementEmailContent": {"ReplacementTemplate": {"ReplacementTemplateData": "{\"name\":\"Ada\"}"}}}: SES wants the template data as a JSON string, which a document template cannot produce from the payload. Set the route’s batch_size to 50 at most.

The request bodies of the Apprise and SendGrid routes were checked against a local HTTP server; the SES example for its configuration only. None of the services themselves was called.

Through an AWS SNS topic

An aws output with a topic_arn publishes every message to an SNS topic. Phone numbers (SMS) and email addresses subscribed to the topic in AWS each receive it:

alerts_sns:
  input: { kafka: { topic: "alerts", url: "localhost:9092" } }
  output:
    aws:
      topic_arn: "arn:aws:sns:eu-central-1:123456789012:ops-alerts"
      region: "eu-central-1"

On the command line: --to 'aws://?topic_arn=arn:aws:sns:eu-central-1:123456789012:ops-alerts&region=eu-central-1'.

• A subscriber without a filter policy gets every message; one with a filter policy gets only the messages it matches. To reach different people, use one topic per audience and a switch to choose between them.
• The payload is the text that is sent. Emails arrive as plain text with the subject “AWS Notification”, and each address must confirm its subscription before it receives anything.
• Message metadata is sent as SNS message attributes, which subscription filter policies can match.

This example was checked for its configuration only.

Replies and delivery reports

Notifications have answers too: delivery reports, bounces, SMS replies.

• Providers that call a webhook (SendGrid, Postmark, Twilio) reach an http input, which writes each call on to any output.
• SES reports bounces and complaints to an SNS topic. Subscribe an SQS queue to it and read the queue with an aws input with queue_url.

Weak join / correlation

The weak_join middleware correlates messages by a metadata key and emits them as one joined message. Input only. It has two modes.

Count mode

Wait for any N messages sharing a correlation key, then emit them as a JSON array:

input:
  middlewares:
    - weak_join: { group_by: "correlation_id", expected_count: 3, timeout_ms: 5000 }
  kafka: { topic: "fragments", url: "localhost:9092" }

Branch mode

Set branch_by to wait for named branches (e.g. one message from inventory and one from pricing), then emit a branch-keyed JSON object. required overrides expected_count:

- weak_join:
    group_by: "correlation_id"
    expected_count: 2
    timeout_ms: 5000
    branch_by: "source"
    required: ["inventory", "pricing"]
    on_timeout: discard

From the copy CLI, object/array fields take a JSON literal:

--from '...|weak-join?group_by=cid&expected_count=2&timeout_ms=1000&required=["inventory","pricing"]'

On timeout

An incomplete group is either emitted partially (on_timeout: fire, the default) or dropped (on_timeout: discard).

Acknowledgement. By default (ack: on_join) a buffered message is acknowledged only once the joined message is committed, so a crash before the group completes redelivers its members. The source must allow that many unacknowledged messages: with a prefetch limit (e.g. AMQP prefetch_count) a group can stall until its timeout. ack: on_receive acknowledges on receipt instead and loses the buffered members on a crash.

Full field list in the middleware reference → weak_join.

Secrets & interpolation

Keep credentials out of committed config, and template message bodies from request data. There are two related interpolation systems: config-value env references (resolved when the config loads) and ${namespace:selector} message templating (resolved per message, in static bodies and a few endpoint fields).

Config values from the environment

Reference environment variables anywhere in JSON/YAML/UI values with ${ENV_VARIABLE_NAME:-default_if_not_found}:

orders_out:
  output:
    kafka:
      topic: "orders"
      url: "${KAFKA_URL:-localhost:9092}"
      # a password sourced from the environment, never committed
      username: "svc-orders"
      password: "${KAFKA_PASSWORD}"

For local development, drop a .env file in the working directory — it is loaded automatically. In containers/Kubernetes, set the vars in the environment (and override any config field with MQB__{ROUTE}__{PATH}; see Configuration grammar). This is the recommended way to keep secrets out of source — see the deploying security checklist.

Custom endpoints and plugins

When the app stores secrets outside the config file (the balanced config security mode), it also takes them out of a custom endpoint’s or middleware’s config, such as a Connect component:

• a string field whose name contains password, passphrase, secret or credentials, or ends in token, api_key, apikey, access_key, account_key, private_key or connection_string, at any depth;
• a URL with user:pass@ in it, also inside a list such as urls, and a dsn containing @.

Left in the file, because an environment variable cannot carry them back unchanged:

• a field whose name is not lowercase snake_case (apiKey, X-Api-Key), and everything below it;
• a value that reads as a number or boolean ("12345", "true");
• a value that already is a reference ("${DB_PASSWORD}");
• a Connect yaml document.

For those, reference the secret from the environment as above.

Encryption keys from the environment

The encryption middleware and the file endpoints read their key with the ${env:VAR} form:

- encryption: { key: "${env:MQB_ENC_KEY}" }

Message templating with ${namespace:selector}

The static endpoint’s body is a template compiled once at startup. Tokens use the ${namespace:selector} form:

output:
  static:
    body: '{"error":"not found","id":"${message:id}","at":"${gen:now}"}'
    raw: true
    metadata: { content-type: "application/json" }

• payload/metadata/message read the request, so they’re the useful ones on an output (e.g. an error reply echoing the request); on an input (a load-test source) only gen/env produce values.
• When the body’s content-type is a JSON type, interpolated request values are JSON-escaped by default so external data can’t break the structure — append | raw to a token to splice it verbatim.
• To emit a literal ${…}, write $${…}. Any ${…} with an unknown namespace is left untouched.

SQL / query token mapping

The sqlx insert_query supports the same request-binding tokens so a typed value goes into the statement: ${payload:field} binds a typed JSON value, ${metadata:key} binds a metadata string. See Upserts & insert-if-absent.

Encryption at rest

Two ways to encrypt data, depending on whether you want it encrypted in transit between endpoints or at rest in a file / object.

The encryption middleware — payloads in transit

The encryption middleware seals each message payload into a self-describing AEAD envelope on the output side and decrypts it on the input side. Metadata and routing keys stay in the clear. Requires the encryption feature.

output:
  middlewares:
    - encryption: { key: "${env:MQB_ENC_KEY}" }
  nats: { subject: "secure.orders", url: "nats://localhost:4222" }

The key is a base64-encoded 32-byte key; ${env:VAR} reads it from the environment (see Secrets & interpolation). Cipher defaults to xchacha20poly1305 (aes256gcm also available).

A sealed payload is binary, so a file/object_store sink that stores it must use format=normal — format=json/text render the payload as a JSON value and it cannot be read back (the reader reports “unsupported encryption envelope version 91”). Stacking it with compression works, but the reading route must list the two middlewares in the reverse order, since middlewares apply in list order on both ends.

From the copy CLI it is an inline middleware; keep the key in ${env:VAR} so it never reaches the process list or shell history ({/} percent-encode to %7B/%7D):

MQB_ENC_KEY=… mqb copy \
  --from 'file:///tmp/orders.jsonl' \
  --to   'nats://localhost:4222?stream=secure&subject=secure.orders|encryption?key=$%7Benv:MQB_ENC_KEY%7D'

Key rotation: seal with a new key_id/key while listing the old key under decrypt_keys on the consuming side. Each payload is authenticated independently — tampering, a torn frame, or a wrong/missing key fails decryption. By default (on_error: drop) such a message is logged, acknowledged and skipped, so a reader with the wrong key drains its input; set on_error: fail to stop the route instead.

The AEAD binds only the payload (empty associated data), so a sealed payload can be replayed under different metadata. Use deduplication or a sink uniqueness constraint if that matters.

File / object encryption at rest

To store data compressed and encrypted at rest, use the file / object_store endpoints’ own compression + encryption fields — they apply compress-then-encrypt per batch, which is what you want (ciphertext does not compress):

output:
  file:
    path: "data.enc"
    format: raw
    compression: lz4          # none | gzip | lz4 | zstd  (`compression` feature)
    encryption: { key: "${env:MQB_ENC_KEY}" }

Do not stack the encryption middleware on top of a sink’s batch compression on the same route — ciphertext won’t compress. Use the endpoint fields above instead.

An encrypted file is written as length-prefixed sealed frames (one per batch) and is only readable through a matching consumer. object_store derives its object extension from these fields (e.g. .jsonl.gz / .jsonl.lz4, plus a trailing .enc when encryption is on).

Reading it back

A file source must declare the same compression/encryption the data was written with. A mismatch (wrong key, wrong codec, or a missing field) is a permanent decode failure — the route ends failed with the error in its status, rather than completing as if the file were empty. See Troubleshooting and the Compression recipe.

Compression

There are two kinds of compression, both behind the compression feature:

• the batch compression field on the file / object_store endpoints — compresses whole write batches, so the output stays readable with zcat / lz4 -d;
• the compression middleware — compresses each message payload, so it works over any transport (Kafka, NATS, HTTP, …), not just files.

Endpoint batch compression

The file and object_store endpoints can compress each batch on write with the compression field:

output:
  file:
    path: "data.jsonl"
    format: json
    compression: lz4        # none | gzip | lz4 | zstd

object_store derives its default object extension from the codec (e.g. .jsonl.gz / .jsonl.lz4).

Reading it back

A file source must declare the same compression the data was written with:

input:
  file:
    path: "data.jsonl"
    format: json
    compression: lz4

A mismatch — wrong codec, or a missing field — is a permanent decode failure: the route ends failed with the error in its status rather than silently completing as if the file were empty. Reading a compressed file with no compression set is likewise rejected up front by sniffing the leading magic bytes, so raw compressed bytes are never emitted as messages.

File compression supports only the default consume mode. csv works too: the header row is written into the first member, so the decoded stream is a normal CSV file.

The compression middleware

To compress payloads over the wire — a Kafka topic, a NATS subject, an HTTP body — attach the compression middleware instead. It compresses each message payload on the output side and decompresses it on the input side; metadata and routing keys are untouched. It works on input and output:

orders_bridge:
  input:
    middlewares:
      - compression: { algorithm: zstd }
    kafka: { topic: "orders", url: "localhost:9092" }
  output:
    middlewares:
      - compression: { algorithm: zstd }
    nats: { subject: "orders.out", url: "nats://localhost:4222" }

Put the same algorithm on both sides of a route. Each payload is framed independently, so unlike the endpoint field this is only readable through a matching consumer — a truncated or corrupt frame is a permanent consumer error rather than an endlessly re-read poison message.

From the copy CLI it is an inline middleware like any other:

mqb copy \
  --from 'nats://localhost:4222?stream=orders&subject=orders.in|compression?algorithm=zstd' \
  --to   'file:///tmp/orders.jsonl'

A compressed payload is binary, so a file/object_store sink that stores it must use format=normal. With format=json or text the payload is rendered as a JSON value and cannot be read back. Verified round trip:

mqb copy --drain \
  --from 'file:///tmp/in.jsonl?format=json' \
  --to   'file:///tmp/packed.bin?format=normal|compression?algorithm=zstd'
mqb copy --drain \
  --from 'file:///tmp/packed.bin?format=normal|compression?algorithm=zstd' \
  --to   'file:///tmp/out.jsonl?format=json'

Compression and encryption

Do not stack the encryption middleware on top of a sink’s batch compression on the same route — ciphertext does not compress. For compressed and encrypted data at rest, use the endpoints’ own fields, which apply compress-then-encrypt per batch:

output:
  file:
    path: "data.enc"
    format: raw
    compression: lz4
    encryption: { key: "${env:MQB_ENC_KEY}" }

An encrypted file is written as length-prefixed sealed frames (one per batch) and is only readable through a matching consumer; object_store adds a trailing .enc since the object is ciphertext, not a directly decompressible .gz.

See also

• Encryption at rest — the encryption side and key handling.
• Middleware & structural endpoints — the authoritative compression/encryption field docs.
• Performance tuning → Compression & encryption cost.

Query a stream in DuckDB

There is no DuckDB connector, and for analytics you rarely need one: the object_store sink writes Parquet, and DuckDB queries Parquet files where they lie, whether in a local directory or a bucket. The same works in reverse: Parquet that DuckDB exports can be read back into a route.

format: parquet needs the parquet feature, which the app’s default build includes.

Export a stream as Parquet

orders_to_lake:
  batch_size: 10000
  input:
    kafka: { topic: "orders", url: "localhost:9092" }
  output:
    object_store:
      url: "file:///var/lib/mqb/lake/orders"   # or s3://bucket/orders, gs://…, az://…
      format: parquet
      compression: zstd                        # the Parquet column codec

The same export from the CLI:

mqb copy --drain \
  'file:///data/orders.jsonl?format=raw' \
  'local-store:///var/lib/mqb/lake/orders?format=parquet&compression=zstd'

• Each message payload must be a JSON object; its top-level fields become the columns. Message metadata is not written.
• The sink writes one Parquet file per flushed batch, so batch_size sets the file size. Thousands of tiny files slow every DuckDB scan, so keep batches large.
• The schema is inferred per batch. A field whose values disagree in type within one batch is written as JSON text in that file rather than failing the batch.

Query it

SELECT country, sum(amount)
FROM read_parquet('/var/lib/mqb/lake/orders/*.parquet', union_by_name = true)
GROUP BY country;

Always pass union_by_name = true. Because each file has its own inferred schema, a field that first appears in a later batch is missing from the earlier files. Without the option DuckDB takes the columns from the first file and silently drops the new one; with it, the columns are merged by name, missing values read as NULL, and a column that is BIGINT in one file and VARCHAR in another becomes VARCHAR.

For a bucket, point the same glob at s3://… and give DuckDB credentials through its httpfs extension.

Choose the file layout

Where the input carries a replay position (Kafka, Postgres CDC, an SQL cursor, a MongoDB change stream, or a file in consume mode), the default name_by: auto names each file after the source range it holds: part-orders-0000000003-…-….parquet. A restart or replay then skips the ranges already written, so re-running the export never duplicates rows. These names are written flat, without date folders. A route with a middleware that drops messages (filter, deduplication, …) falls back to write-time names under auto, to avoid one file per gap; set name_by: source_position to keep replay safety anyway. See Files & object storage for the details.

To partition by date instead, ask for write-time names and the Hive layout:

    object_store:
      url: "file:///var/lib/mqb/lake/orders"
      format: parquet
      name_by: write_time
      date_partition_style: hive   # year=YYYY/month=MM/day=DD/<uuidv7>.parquet

SELECT year, month, day, count(*)
FROM read_parquet('/var/lib/mqb/lake/orders/**/*.parquet',
                  hive_partitioning = true, union_by_name = true)
GROUP BY ALL;

DuckDB reads year as a number but keeps the zero-padded month and day as text, so filter with month = '09'. The dates are the write time (UTC), not a field of the record. The two layouts exclude each other: write-time names do not recognise a replay.

Feed DuckDB results back into a route

Export a query result to Parquet in a directory of its own, then read it with an object_store source. Each row arrives as one JSON-object message:

COPY (SELECT id, country, amount * 2 AS doubled FROM read_parquet('lake/*.parquet', union_by_name = true))
TO '/var/lib/mqb/outbox/2026-09-28.parquet' (FORMAT parquet);

duckdb_results_to_nats:
  input:
    object_store:
      url: "file:///var/lib/mqb/outbox"
      format: parquet
      cursor_id: "duckdb-outbox"
      checkpoint_store: "file:///var/lib/mqb/checkpoints/outbox.json"
  output:
    nats: { subject: "orders.scored", url: "nats://localhost:4222" }

The source reads files in name order and checkpoints the last one it finished, so give each export a name that sorts after the previous one, such as a date or timestamp.

Not covered

Writing into a .duckdb database file (tables, upserts, transactions) needs a native DuckDB endpoint, which does not exist. Parquet covers the common case of making a stream queryable.

Load into Snowflake

There is no Snowflake connector. The object_store sink writes Parquet to S3, GCS or Azure, and Snowflake loads it from an external stage with COPY INTO or Snowpipe.

format: parquet needs the parquet feature, which the app’s default build includes.

Write Parquet to the stage location

orders_to_snowflake_stage:
  batch_size: 10000
  input:
    kafka: { topic: "orders", url: "localhost:9092", group_id: "snowflake-export" }
  output:
    middlewares:
      - transform:
          schema:
            type: object
            required: [id]
            properties:
              id: { type: integer }
              country: { type: string }
              amount: { type: number }
              note: { type: string, default: "" }
    object_store:
      url: "s3://my-bucket/orders"
      format: parquet
      compression: zstd

• Each payload must be a JSON object; its top-level fields become the columns.
• The sink writes one Parquet file per flushed batch, so batch_size sets the file size.
• The transform schema coerces types ("42" becomes 42) and rejects rows that do not fit, so a column does not flip between string and number across files. A number column is still written as a 64-bit integer when a batch holds only whole numbers, so declare it as a float column in the warehouse.
• A rejected row is dropped and logged. Add a dlq after the transform to keep it.
• null takes the field’s default if it has one and is rejected otherwise. Mark a column the source can leave empty as nullable: true to keep the null.
• S3 credentials are read from the environment (AWS_ACCESS_KEY_ID, …).

Schema drift between files

The Parquet schema is inferred per batch. An optional field that no row of a batch carries is absent from that file, not written as a null column. Give such a field a default in the schema, as note has above, or match columns by name on the Snowflake side. Never load by column position.

Load it

Verify this SQL against the current Snowflake documentation.

CREATE STAGE orders_stage
  URL = 's3://my-bucket/orders/'
  STORAGE_INTEGRATION = my_s3_integration
  FILE_FORMAT = (TYPE = PARQUET);

COPY INTO orders
  FROM @orders_stage
  FILE_FORMAT = (TYPE = PARQUET)
  MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

MATCH_BY_COLUMN_NAME loads each file by column name, so a file that lacks a column loads NULL for it.

For continuous loading, wrap the same statement in a pipe and point the bucket’s event notifications at it:

CREATE PIPE orders_pipe AUTO_INGEST = TRUE AS
  COPY INTO orders
    FROM @orders_stage
    FILE_FORMAT = (TYPE = PARQUET)
    MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

Replays

With a replayable input such as Kafka, the default name_by: auto names each file after the source range it holds, and a restart skips ranges already written. See Choose the file layout for the naming options and date partitions.

Load into BigQuery

There is no native BigQuery connector. The object_store sink writes Parquet to Google Cloud Storage, and BigQuery either loads the files or queries them in place through an external table.

format: parquet needs the parquet feature, which the app’s default build includes.

To insert rows into a table directly instead, the Connect plugin has a gcp_bigquery output.

Write Parquet to GCS

orders_to_gcs:
  batch_size: 10000
  input:
    postgres_cdc:
      url: "postgres://user:pass@localhost:5432/app"
      publication: "mqb_pub"
      slot_name: "mqb_bigquery"
  output:
    middlewares:
      - transform:
          schema:
            type: object
            required: [id]
            properties:
              id: { type: integer }
              country: { type: string, default: "" }
              amount: { type: number, default: 0 }
              note: { type: string, default: "" }
    object_store:
      url: "gs://my-bucket/orders"
      format: parquet
      compression: zstd

• Each payload must be a JSON object; its top-level fields become the columns.
• The sink writes one Parquet file per flushed batch, so batch_size sets the file size.
• The transform schema coerces types ("42" becomes 42) and rejects rows that do not fit, so a column does not flip between string and number across files. A number column is still written as a 64-bit integer when a batch holds only whole numbers, so declare it as a float column in the warehouse.
• A rejected row is dropped and logged. Add a dlq after the transform to keep it.
• null takes the field’s default if it has one and is rejected otherwise. Mark a column the source can leave empty as nullable: true to keep the null.
• postgres_cdc emits one message per change, so the files are a change log, not the table’s current state. A delete carries the key with every other column null (default replica identity); it is rejected by the schema above unless those columns are nullable. The operation is in the postgres.operation metadata, not in the payload.
• GCS credentials are read from the environment (GOOGLE_SERVICE_ACCOUNT, …).

Schema drift between files

The Parquet schema is inferred per batch. An optional field that no row of a batch carries is absent from that file. Give such a field a default in the schema, as the optional columns have above, or create the BigQuery table with every column up front so a file that lacks one loads NULL. BigQuery matches Parquet columns by name.

Load it

Verify this SQL against the current BigQuery documentation.

A load job copies the data into a native table:

LOAD DATA INTO mydataset.orders
FROM FILES (
  format = 'PARQUET',
  uris = ['gs://my-bucket/orders/*.parquet']
);

A wildcard load requires every matched file to have the same schema. A default on every optional column keeps the column set the same, but not the types: a number column is INT64 in a file whose batch held only whole numbers and FLOAT64 otherwise, and the destination table’s schema does not reconcile the files. Load files with differing schemas in separate jobs.

A load job appends whatever the URI matches. Running it twice over the same files loads them twice, so load each file once (for example by date prefix) or use an external table instead.

An external table leaves the data in GCS and reads the current set of files on every query:

CREATE EXTERNAL TABLE mydataset.orders_ext
OPTIONS (
  format = 'PARQUET',
  uris = ['gs://my-bucket/orders/*.parquet']
);

Replays and layout

With a replayable input such as Postgres CDC, the default name_by: auto names each file after the source range it holds, and a restart skips ranges already written. For date folders that a load job can address by prefix, see Choose the file layout.

Load into Databricks / Spark

There is no Databricks or Spark connector. The object_store sink writes Parquet to S3, GCS or Azure, and Spark reads Parquet from all three.

format: parquet needs the parquet feature, which the app’s default build includes.

Write Parquet to the lake

orders_to_lake:
  batch_size: 10000
  input:
    kafka: { topic: "orders", url: "localhost:9092", group_id: "lake-export" }
  output:
    middlewares:
      - transform:
          schema:
            type: object
            required: [id]
            properties:
              id: { type: integer }
              country: { type: string }
              amount: { type: number }
              note: { type: string, default: "" }
    object_store:
      url: "s3://my-bucket/orders"        # or gs://…, az://container/…
      format: parquet
      compression: zstd

• Each payload must be a JSON object; its top-level fields become the columns.
• The sink writes one Parquet file per flushed batch, so batch_size sets the file size. Keep batches large: many small files slow every scan.
• The transform schema coerces types ("42" becomes 42) and rejects rows that do not fit, so a column does not flip between string and number across files. A number column is still written as a 64-bit integer when a batch holds only whole numbers, so declare it as a float column in the warehouse.
• A rejected row is dropped and logged. Add a dlq after the transform to keep it.
• null takes the field’s default if it has one and is rejected otherwise. Mark a column the source can leave empty as nullable: true to keep the null.

Schema drift between files

The Parquet schema is inferred per batch. An optional field that no row of a batch carries is absent from that file. Give such a field a default in the schema, as note has above, or ask Spark to merge the file schemas by name with mergeSchema. Without it Spark takes the schema from one file and drops columns that file lacks.

Load it

Verify this code against the current Databricks and Spark documentation.

Read the files directly:

df = spark.read.option("mergeSchema", "true").parquet("s3://my-bucket/orders/")

Or load them into a Delta table. COPY INTO skips files it has already loaded:

COPY INTO orders
  FROM 's3://my-bucket/orders/'
  FILEFORMAT = PARQUET
  FORMAT_OPTIONS ('mergeSchema' = 'true')
  COPY_OPTIONS ('mergeSchema' = 'true');

For continuous ingestion, Databricks Auto Loader (cloudFiles with cloudFiles.format = parquet) picks up new files as they arrive.

Replays and layout

With a replayable input such as Kafka, the default name_by: auto names each file after the source range it holds, and a restart skips ranges already written. For Hive-style date partitions (year=…/month=…/day=…), which Spark discovers as columns, see Choose the file layout.

Query from Athena / Trino

There is no Athena or Trino connector, and none is needed: both query Parquet files where they lie. The object_store sink writes the files, and an external table makes them queryable.

format: parquet needs the parquet feature, which the app’s default build includes.

Write Parquet to S3

orders_to_s3:
  batch_size: 10000
  input:
    kafka: { topic: "orders", url: "localhost:9092", group_id: "athena-export" }
  output:
    middlewares:
      - transform:
          schema:
            type: object
            required: [id]
            properties:
              id: { type: integer }
              country: { type: string }
              amount: { type: number }
              note: { type: string, default: "" }
    object_store:
      url: "s3://my-bucket/orders"
      format: parquet
      compression: zstd
      name_by: write_time
      date_partition_style: hive   # year=YYYY/month=MM/day=DD/<uuidv7>.parquet

• Each payload must be a JSON object; its top-level fields become the columns.
• The sink writes one Parquet file per flushed batch, so batch_size sets the file size. Athena bills by data scanned and slows down on many small files, so keep batches large.
• The transform schema coerces types ("42" becomes 42) and rejects rows that do not fit, so a column does not flip between string and number across files. A number column is still written as a 64-bit integer when a batch holds only whole numbers, so declare it as a float column in the warehouse.
• A rejected row is dropped and logged. Add a dlq after the transform to keep it.
• null takes the field’s default if it has one and is rejected otherwise. Mark a column the source can leave empty as nullable: true to keep the null.
• The date folders are the write time (UTC), not a field of the record. Write-time names do not recognise a replay; drop the last two lines to get replay-safe names in a flat layout instead. See Choose the file layout.

Schema drift between files

The Parquet schema is inferred per batch. An optional field that no row of a batch carries is absent from that file. Declare every column in the table definition and let the engine resolve Parquet columns by name: a file that lacks a column then reads NULL for it. Giving the field a default in the schema, as note has above, keeps the column in every file.

Query it

Verify this SQL against the current Athena and Trino documentation.

CREATE EXTERNAL TABLE orders (
  id      bigint,
  country string,
  amount  double,
  note    string
)
PARTITIONED BY (year string, month string, day string)
STORED AS PARQUET
LOCATION 's3://my-bucket/orders/';

MSCK REPAIR TABLE orders;   -- registers new year=/month=/day= folders

SELECT country, sum(amount)
FROM orders
WHERE year = '2026' AND month = '09'
GROUP BY country;

MSCK REPAIR TABLE has to run again when a new day’s folder appears; partition projection avoids that.

In Trino, the same files are a Hive-connector table:

CREATE TABLE hive.lake.orders (
  id bigint, country varchar, amount double, note varchar,
  year varchar, month varchar, day varchar
)
WITH (
  external_location = 's3://my-bucket/orders/',
  format = 'PARQUET',
  partitioned_by = ARRAY['year', 'month', 'day']
);

Keep the Hive connector reading Parquet columns by name (hive.parquet.use-column-names).

ClickHouse: endpoint or s3()

ClickHouse is the one analytics database here with a native endpoint. You can insert rows directly, or write Parquet to object storage and let ClickHouse read it with the s3() table function.

Insert directly

orders_to_clickhouse:
  batch_size: 10000
  input:
    kafka: { topic: "orders", url: "localhost:9092", group_id: "clickhouse-export" }
  output:
    middlewares:
      - transform:
          schema:
            type: object
            required: [id]
            properties:
              id: { type: integer }
              country: { type: string }
              amount: { type: number }
    clickhouse: { url: "http://localhost:8123", database: "analytics", table: "orders" }

The transform schema rejects rows that would not fit the table before they reach ClickHouse. Put a dlq after it to keep them.

Write Parquet, load with s3()

orders_to_lake:
  batch_size: 10000
  input:
    kafka: { topic: "orders", url: "localhost:9092", group_id: "lake-export" }
  output:
    middlewares:
      - transform:
          schema:
            type: object
            required: [id]
            properties:
              id: { type: integer }
              country: { type: string }
              amount: { type: number }
              note: { type: string, default: "" }
    object_store:
      url: "s3://my-bucket/orders"
      format: parquet
      compression: zstd

format: parquet needs the parquet feature, which the app’s default build includes. The sink writes one Parquet file per flushed batch.

The Parquet schema is inferred per batch. An optional field that no row of a batch carries is absent from that file. Give such a field a default in the schema, as note has above, or let ClickHouse fill missing columns and merge the file schemas by name.

Verify this SQL against the current ClickHouse documentation.

INSERT INTO analytics.orders
SELECT id, country, amount, note
FROM s3('https://my-bucket.s3.amazonaws.com/orders/*.parquet', 'Parquet')
SETTINGS
  input_format_parquet_allow_missing_columns = 1,  -- a column absent from a file reads as default
  schema_inference_mode = 'union';                 -- merge the schemas of all files by name

Running the INSERT again loads the same files again. Select by path (_path) or load each prefix once.

Endpoints (connectors)

An endpoint is a source (consumer) or sink (publisher) for messages. Every endpoint — whatever the underlying protocol — is driven behind the same receive_batch / send_batch shape, so any source can feed any sink. In config, an endpoint is a single-key object naming the connector:

input:  { kafka:   { url: "localhost:9092", topic: "orders" } }
output: { mongodb: { url: "mongodb://localhost:27017", database: "app", collection: "orders" } }

This page catalogs the transport endpoints and their behaviour, for library configs and the CLI alike. Each connector has its own connector page with YAML and URL examples, plus a generated parameter table (name, type, default, description). Structural endpoints (ref, fanout, switch, response, …) are documented separately in Structural endpoints.

Supported transports

Kafka, NATS, AMQP (RabbitMQ), MQTT, MongoDB, Postgres CDC (logical replication), PostgreSQL / MySQL / SQLite / MariaDB (SQLx), ClickHouse, HTTP, WebSocket, gRPC, ZeroMQ, Redis Streams, AWS SQS/SNS, cloud object storage (S3 / GCS / Azure), IBM MQ, files, and in-memory channels (memory, in-process and cross-process IPC).

Plugin endpoints

These live in their own repositories and load as native plugins, so the core library carries none of their dependencies:

The Connect plugin has its own page, with every linked input, output and processor.

Outside mqb (Python, Node.js, your own Rust host), or for Connect, install the plugin with brew install marcomq/tap/<repo> or conda install -c marcomq <repo>; it is then discovered automatically.

Consumer vs. subscriber, and nack support

One table per question (which inputs acknowledge, which commit in order, which outputs reply, where TLS is configured) is in CAPABILITIES.md. It is rendered from the table the engine exposes in code and checked by tests.

Endpoints default to a Consumer pattern (a queue: messages are distributed among workers). To get Subscriber (pub/sub) behaviour, per-backend config is required.

• Request-Reply — Native uses protocol-level correlation (HTTP connection, NATS reply subject). Emulated publishes a new message to a reply destination (the reply_to metadata field) carrying a correlation_id.
• Nack — Yes is explicit negative acknowledgement triggering redelivery. Eventual means redelivery depends on timeout or a connection drop. Simulated is handled in-memory. Consumer-local means the nacked message is retried inside the consumer process and is lost if that process dies — the producer is never notified.

Database sources: change capture vs. polling

Databases have no native pub/sub, so a database source is read one of two ways:

• Change Data Capture (CDC) tails the database’s own change log — inserts and updates/deletes — and resumes from a durable log position.
  • PostgreSQL — the postgres_cdc endpoint streams a logical-replication slot (pgoutput). It emits flat JSON rows tagged with postgres.operation metadata (insert/update/delete/truncate). Acking a batch confirms the LSN back to the server; a nack or interrupted run does not advance the confirmed LSN, so replication resumes from the last acknowledged position — at-least-once. Enable with the postgres-cdc feature; requires wal_level = logical and a publication. See the Postgres CDC → JSONL tutorial.
  • MongoDB — consume: capture_all (the default) reads existing documents first and then keeps capturing (no gap, at-least-once); consume: capture_new watches an existing collection for changes from now on. Both emit insert/update/replace/delete tagged with mongodb.operation and checkpoint under cursor_id. Change streams need a replica set; without one, use consume: snapshot for a one-shot read.
  • MySQL, MariaDB, SQL Server, Oracle, Db2 — no native endpoint. Run Debezium Server beside mq-bridge and read its output; see CDC via Debezium.
• Cursor polling pages an existing table by a monotonic cursor_column (WHERE col > $last ORDER BY col ASC), persisting the last read value under cursor_id. Captures appends only — updates and deletes are not observed. Available on SQLx (PostgreSQL / MySQL / MariaDB / SQLite) and ClickHouse; the poll interval backs off exponentially between polling_interval_ms and max_polling_interval_ms. SQLite and ClickHouse are polling-only.

SQLx read mode is chosen by config, not by driver. With no cursor_column, an SQLx source is a competing-consumers work queue: it atomically claims rows via a locked_until lease and deletes them on ack. This requires the table to carry the queue schema — an id, a payload, and a locked_until column — on every driver; it is not a plain full-table read. To read an arbitrary table non-destructively (the ETL path), set cursor_column to switch to cursor polling. A source table missing locked_until fails fast with a permanent error rather than reconnecting forever.

MongoDB consume modes

These are not interchangeable. delete buys exclusivity via four round trips per batch and deletes what it reads, so it is for work queues, not for reading a collection; where you need a single-reader bulk read or ETL pass, the default capture_all is roughly 5x faster (500k docs: delete → null 23,667 rows/s vs. capture_all → jsonl 120,308 rows/s). Note concurrency does not speed up a MongoDB source; batches are fetched serially.

Both capture_* modes read the oplog and therefore require a replica set — a single-node one is enough — and refuse to start without one. On a standalone mongod, use snapshot for a one-shot non-destructive read. snapshot delivers what exists when the run starts and is not a tail: it pages by _id, which is assigned client-side and so does not follow commit order, so it rejects cursor_id rather than resuming above a stored _id and silently skipping whatever a concurrent writer commits below that mark. Incremental reads need commit order — i.e. the oplog, i.e. a replica set.

Removed in 0.4.0: consume: subscriber. It polled seq > last_seq and advanced its watermark to the highest seq it had seen, so a batch whose seq block was reserved first but committed second was skipped permanently — silent loss, with no error and no gap in the delivery count. Use capture_new for ephemeral fan-out (it now reads arbitrary collections, not only documents written by the bridge’s own publisher). The deprecated change_stream: true boolean resolves to capture_new.

Object storage (local / cloud)

The object_store endpoint reads and writes local directories, Amazon S3, Google Cloud Storage, Azure Blob, Cloudflare R2, and anything else the object_store crate speaks. Enable it with the object-store feature. Use local-store:///absolute/path in CLI URLs (it becomes file:///absolute/path internally), or use url: file:///absolute/path in structured YAML. Credentials and cloud backend options come from the environment (AWS_ACCESS_KEY_ID, AWS_REGION, GOOGLE_SERVICE_ACCOUNT, AZURE_STORAGE_ACCOUNT_NAME, …); the URL scheme picks the backend (file://, s3://, gs://, az://). The variables and URL form of each cloud are on the object storage connector page.

• As a sink, each flushed batch is encoded with the file formats (normal JSONL, json, text, raw) and written as one immutable object at <prefix>/[YYYY/MM/DD/]<uuidv7>.<ext>. Objects are write-once.
• As a source, objects under the prefix are listed in key order, fetched whole, split on the delimiter, and emitted. Progress is a durable cursor holding the last fully-acked object key: set cursor_id and an external checkpoint_store (file://, s3://, postgres://, mongodb://) so a restart resumes. Objects are never deleted or rewritten.

archive_to_s3:
  input:  { memory: { topic: "events" } }
  output: { object_store: { url: "s3://my-bucket/events", format: normal } }

replay_from_s3:
  input:
    object_store:
      url: "s3://my-bucket/events"
      cursor_id: "replayer-1"
      checkpoint_store: "file:///var/lib/mqb/s3-cursor.json"
  output: { nats: { subject: "events.replay", url: "nats://localhost:4222" } }

Point checkpoint_store at a different bucket/prefix than the source reads; a cursor object under the source prefix would be listed and re-read as data.

File and object formats

Both file and object_store share the encodings: normal/json/text write the {message_id, payload, metadata} wrapper (the message id survives the round trip); raw writes payloads verbatim (bare documents, no wrapper). csv is supported as a source (and on file sinks). The file endpoints also carry their own compression and encryption fields — see the Compression and Encryption at rest recipes.

A source on normal/json/text expects that same wrapper. A line that is valid JSON but not the wrapper becomes the whole payload (its own metadata is discarded, with a warning); use format: raw for plain JSON lines.

In the wrapper, a UTF-8 payload is a plain JSON string under payload; a binary one (compressed, encrypted, Protobuf, …) is base64-encoded under payload_base64. The two are mutually exclusive, as in the CloudEvents JSON format:

{"message_id":"019f9b12-d786-7ebe-a7ec-a1aa71bc47ae","payload":"{\"order_id\":7}"}
{"message_id":"019f9b12-d78a-7c01-b0f4-2f0f4d6a1c33","payload_base64":"KLUv/SBOAQAA"}

Sources still read the older byte-array form ("payload":[123,34,…]). A binary record written by this version is not readable by an older mq-bridge; text records are compatible in both directions.

Memory endpoint and IPC

The memory endpoint is both an in-process channel and a cross-process IPC transport over Unix domain sockets / Windows named pipes. Its topic field (alias url) doubles as a transport URL. See the Cross-process IPC bridge tutorial and Learn the architecture for the URL schemes, roles (consumer is server, publisher is client), wire format and backpressure.

IBM MQ

IBM MQ is included in the full feature set via the ibm-mq feature, which loads the IBM MQ client library at runtime via dlopen — no IBM SDK is needed to build. The redistributable client only has to be present at runtime, and only if you actually use an IBM MQ endpoint (it is loaded lazily on first connect). The loader finds the client via the platform default name, MQ_INSTALLATION_PATH (e.g. /opt/mqm), or an explicit MQB_IBM_MQ_LIB path. To link statically at build time, use ibm-mq-static (requires the IBM MQ SDK). IBM MQ has its own TLS shape (IbmTlsConfig): tls.cert_file (alias key_repository) is a CMS key repository path, not a PEM file. See the IBM MQ setup guide.

Connection sharing

Publishers that target the same server reuse one underlying transport client by default, consolidating TCP connections, background threads, and batching. Sharing applies to Kafka, NATS, MongoDB, SQLx, HTTP, and gRPC; the client is keyed by connection-level settings (URL, auth, TLS, client options), never by topic/subject/collection. Set shared: false on a publisher to give it a dedicated connection — see Configuration grammar and Performance tuning.

Idempotent writes

For ETL, at-least-once delivery plus an idempotent write gives effective exactly-once. The most robust place to enforce this is the sink’s own unique constraint. See the Upserts & insert-if-absent and Deduplication recipes for the per-sink mechanisms (id_field on MongoDB, ON CONFLICT/ON DUPLICATE KEY on SQL, ReplacingMergeTree on ClickHouse, deterministic message_id + postgres.key on Postgres CDC).

Connectors

One page per connector: purpose, practical examples, and the same settings in both forms — as a mqb copy URL and as a YAML endpoint for config files and the library. The nested Parameters page under each connector lists every field (type, default, description), generated from the same schema the engine uses.

From URL to config

A URL’s query parameters are config fields of the same name. The scheme picks the config key, and the part before the query string becomes url (or path):

So kafka://localhost:9092?topic=orders&group_id=sync becomes:

kafka: { url: "localhost:9092", topic: "orders", group_id: "sync" }

Transport behaviour shared by all connectors — consumer vs. subscriber, nack support, CDC vs. polling, file formats, plugins — is on Overview & capabilities.

All connectors

• PostgreSQL / MySQL / MariaDB / SQLite (including PostgreSQL CDC)
• ClickHouse
• MQTT
• Kafka
• RabbitMQ (AMQP)
• NATS
• Redis Streams
• HTTP
• WebSocket
• gRPC
• MongoDB
• AWS SQS / SNS
• ZeroMQ
• IBM MQ
• File (CSV / JSON / JSONL)
• Directory spool
• Object storage: S3, MinIO / R2, Google Cloud Storage, Azure Blob, local directory
• Apache Pulsar
• Meilisearch
• HTTP bulk (search engines)
• Typesense
• Elasticsearch
• PostgREST and Supabase
• Qdrant
• OpenAI-compatible APIs (LLMs)
• Connect plugin (Redpanda Connect components)

PostgreSQL / MySQL / MariaDB / SQLite

The sqlx connector reads and writes rows in a relational database table. The same connector backs all four schemes — the URL scheme just selects the driver. Postgres additionally has a dedicated CDC connector for streaming change data instead of table reads.

URL format

postgres://[user:pass@]host[:port]/database?table=<name>

The scheme (postgres/postgresql, mysql, mariadb, sqlite) and everything up to the query string is passed straight through as the driver connection string; table (and any other recognised field below) is pulled out of the query string into config. For SQLite, host/database are replaced by a file path, e.g. sqlite:///var/data/app.db?table=orders.

Config (YAML / library)

The same settings as a route endpoint in a config file, or in Route.from_config / fromConfig. Every URL query parameter is a field of the same name under sqlx:.

input:
  sqlx: { url: "postgres://user:pass@localhost/app", table: "orders", cursor_column: "id" }

The config key is sqlx for PostgreSQL, MySQL/MariaDB, and SQLite; the connection string is the url.

Examples

Full-table read (source), one-shot:

mqb copy --drain \
  --from 'postgres://user:pass@localhost/app?table=orders&cursor_column=id' \
  --to null:

cursor_column names a unique, increasing column to read in order. Without it the source treats the table as an mq-bridge work queue, which needs id and locked_until columns.

Write with auto-created table (destination):

mqb copy --drain \
  --from file:///data/orders.csv?format=csv \
  --to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'

Resumable incremental read, keyed by a monotonic column, continuous:

mqb copy \
  --from 'postgres://user:pass@localhost/app?table=orders&cursor_column=id&cursor_id=orders_export' \
  --to kafka://kafka.local:9092?topic=orders

Each restart resumes from the last id seen (persisted via cursor_id) instead of re-copying from the start.

Table to table, or file to table, by column name:

mqb copy --drain \
  --from 'postgres://user:pass@src/app?table=orders&cursor_column=id' \
  --to 'postgres://user:pass@dst/app?table=orders&columns=auto&key=id'

mqb copy --drain \
  --from 'file://orders.csv?format=csv' \
  --to 'sqlite://app.db?table=orders&columns=auto&key=id'

columns=auto writes each top-level field of a record into the column of the same name; the table must exist. key=id makes a rerun update the rows instead of failing on the duplicate key. See Writing by column name.

Custom multi-column insert, MySQL:

mqb copy --drain \
  --from postgres://user:pass@src/app?table=orders \
  --to 'mysql://user:pass@dst/app?table=orders&insert_query=INSERT+INTO+orders+%28id%2C+sku%2C+qty%29+VALUES+%28%24%7Bpayload%3Aid%7D%2C+%24%7Bpayload%3Asku%7D%2C+%24%7Bpayload%3Aqty%7D%29'

(insert_query is shown URL-encoded above — the SQL is INSERT INTO orders (id, sku, qty) VALUES (${payload:id}, ${payload:sku}, ${payload:qty}).)

Key options

Any other query parameter (e.g. sslmode=disable) is left on the connection URL untouched and passed to the driver as-is.

Full field list, types, and defaults: reference/postgres.md.

Reading a table

With cursor_column the source pages through the table and leaves it unchanged:

SELECT … FROM orders WHERE id > <last value> ORDER BY id LIMIT <batch size>

Each row becomes one message whose payload is a JSON object with one field per column:

{"id": 42, "sku": "A-1", "qty": 3, "price": "19.90", "created_at": "2026-10-04 09:15:23.923277+00"}

The cursor column must be unique and only grow, because the next page starts after the last value read:

• A row whose value equals the last one read is skipped, so a column with repeated values (a plain updated_at) loses rows.
• A row that appears later with a smaller value is never read. This happens with a sequence when a long transaction commits after a shorter one that took a higher number. On PostgreSQL the reader stops each page before the first row written by a transaction that is not older than every open one, so a sequence column is safe; while a write transaction stays open anywhere on the server, newer rows wait for it. A column filled from now() (the transaction’s start time) or by the application can still go wrong, and so can a view, which has no xmin (the reader logs a warning). On MySQL, MariaDB and SQLite there is no such guard. Where writers run concurrently, prefer PostgreSQL CDC.
• Updates and deletes of rows already read are not seen. Use PostgreSQL CDC for those.
• The column can be an integer or a text type. On PostgreSQL it can also be a timestamp, timestamptz, date, numeric or uuid.

Column types. Integers, floats, booleans and text keep their JSON type. The other types arrive as text:

Binary columns (bytea, blob) are written as a hex string. To get numbers or timestamps in another form, add a transform, or set timestamps: rfc3339 on PostgreSQL.

Progress. cursor_id stores the last value, by default in a table mqb_cursors_<table> of the source database. For a read-only user, point checkpoint_store somewhere else; see Checkpoints & resumable copies. Without cursor_id the position is kept for the current run only.

Polling. With --drain (exit_on_empty) the run ends at the last row. Otherwise the source asks for new rows every polling_interval_ms (default 100); set max_polling_interval_ms to let the interval grow while the table is idle.

Queue tables

Without cursor_column the table is used as a queue between mq-bridge routes: one route writes messages into it, another takes them out. This is not a way to read an existing table.

A sink without columns and insert_query writes the whole payload into a payload column. auto_create_table: true creates the table; on PostgreSQL it is:

CREATE TABLE jobs (
  id BIGSERIAL PRIMARY KEY,
  payload BYTEA NOT NULL,
  locked_until TIMESTAMPTZ,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

enqueue:
  input:  { http: { url: "0.0.0.0:8080" } }
  output: { sqlx: { url: "postgres://localhost/app", table: "jobs", auto_create_table: true } }

work:
  input:  { sqlx: { url: "postgres://localhost/app", table: "jobs", delete_after_read: true } }
  output: { kafka: { url: "localhost:9092", topic: "jobs" } }

• The source claims a batch of rows for 60 seconds (locked_until) and skips rows another consumer has claimed, so several consumers can share one table.
• With delete_after_read: true an acknowledged row is deleted. A row that failed stays and is delivered again when its claim ends.
• Set delete_after_read: true for a queue. Without it a row is only hidden for the 60 seconds of its claim: nothing marks it as done, so it is delivered again and again, once a minute, for as long as the route runs. The default is off so that a route deletes nothing unless told to; it is only useful if something else removes the processed rows.
• To copy a table instead of emptying it, use cursor_column (Reading a table). The mqb copy command warns when a source has neither setting; a route in a config file starts without a warning.
• Only the payload is stored. Message metadata does not pass through the table.

Writing

A row the database rejects for a reason that will not change (a type error, a violated constraint) fails as non-retryable and goes to a dlq if one is configured. Connection errors and deadlocks are reported as retryable, for a retry middleware to repeat.

Connection, TLS and pool

• Credentials can be in the URL or in username / password, which take precedence. Keep them out of the config file with secrets.
• TLS. Either put the driver’s option into the URL (?sslmode=verify-full on PostgreSQL, ?ssl-mode=VERIFY_IDENTITY on MySQL/MariaDB), or set tls.required: true, which selects exactly these modes. tls.ca_file names a private CA, tls.cert_file and tls.key_file a client certificate; the key file must be unencrypted. The drivers’ default mode falls back to plain text when the server offers no TLS.
• Pool. Endpoints with the same URL share one pool of at most max_connections (10) connections. Raise it when a route’s concurrency is higher, and set shared: false to give an endpoint a pool of its own.

Database notes

• PostgreSQL. A table in another schema is table: "sales.orders". bulk_copy and the type casts of columns: auto exist only here.
• MySQL / MariaDB. Both schemes use the same driver. A key needs a UNIQUE or PRIMARY KEY index and becomes ON DUPLICATE KEY UPDATE.
• SQLite. sqlite:///var/data/app.db is an absolute path, sqlite://app.db a relative one. A sink creates the file when it does not exist.

Recipes

• Upserts & insert-if-absent: ON CONFLICT and ON DUPLICATE KEY.
• Enrichment & lookups: lookup_query reads a row for each message.
• Checkpoints & resumable copies: where the cursor is stored and how to reset it.
• Deduplication and Running aggregates can keep their state in a SQL table.
• Postgres CDC → JSONL, and MySQL, SQL Server & Oracle CDC via Debezium.

PostgreSQL CDC

A separate connector for streaming logical-replication changes (insert/ update/delete) instead of reading a table snapshot. Uses postgres-cdc:// (alias pgcdc://) to select the endpoint kind; the connection URL underneath it is a plain Postgres URL.

postgres-cdc://[user:pass@]host[:port]/database?publication=<name>&slot_name=<name>

In config, the key is postgres_cdc:

input:
  postgres_cdc: { url: "postgres://user:pass@localhost:5432/app", publication: "orders_pub", slot_name: "mqb_orders" }

Stream changes from a publication into Kafka, continuous:

mqb copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to kafka://kafka.local:9092?topic=app-changes

Replicate a table into another PostgreSQL instance, continuous:

mqb copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to 'postgres://user:pass@otherhost/replica?table=orders&auto_create_table=true'

publication must already exist on the source (CREATE PUBLICATION mqb_pub FOR TABLE orders;); slot_name is created automatically if missing.

The server must be PostgreSQL 14 or newer and run with wal_level = logical.

TLS follows the URL (?sslmode=verify-full&sslrootcert=/ca.pem, with sslcert and sslkey for a client certificate) or the tls block; both connections the source opens use the same settings. sslmode=prefer, the default, tries TLS first and falls back to plain text, so it does not guarantee TLS.

A nacked change makes the source reconnect and deliver again from the last confirmed position, so the changes after it come twice. Rows of one transaction are confirmed together, when its last row is acked.

What a change event contains

The payload is the row as a JSON object; the metadata names the change: postgres.operation (insert, update, delete, truncate), postgres.schema, postgres.table, postgres.lsn and postgres.key (the primary key values as a JSON array).

• An update that changes the primary key also carries postgres.old_key. A sink that upserts by key has to delete that row itself, or the old row stays. With split_key_change: true the source does it: it sends a delete for the old key first, then the update. That is two events for one change, and the events after it in the same transaction get other message ids than without the option, so do not switch it on a route whose deduplication state must stay valid across the change.
• A delete carries the key columns only, the other columns are null. Set REPLICA IDENTITY FULL on the table to get the whole old row.
• A large value that an update did not change (an unchanged TOAST column) is not in the log; its field is left out of the payload, which is different from null.
• numeric is a string, to keep every digit. bigint is a JSON number: a reader that holds numbers as doubles loses precision above 2^53. Numbers inside json / jsonb are read as doubles too.
• bytea is base64. Dates, times, uuid and interval are the text Postgres prints. NaN and Infinity are strings. One-dimensional arrays of the common types are JSON arrays; other arrays, ranges, enums and composite types are their Postgres text.

Full field list: reference/postgres-cdc.md.

Writing by column name

With columns: auto the sink reads the table’s columns at start and builds the INSERT from each record. It works on PostgreSQL, MySQL/MariaDB and SQLite.

• Matching. A field goes into the column with the same name; if there is none, a name that differs only in case matches (ID → id). When a record has both spellings, the exact one is written and the other counts as a field without a column. Fields without a column are not written, and the first one is logged as a warning.
• extra_column. Names a column that takes those fields instead, as one JSON object: with extra_column=extra, {"id":1,"color":"red"} writes id = 1 and extra = {"color":"red"}. A record with no such field leaves the column out. When a record fills the column itself with an object (or with JSON text holding one, as a SQL source delivers it), the collected fields are added to it and its own keys win. An upsert replaces the stored object; it does not merge into it.
• Missing fields. A column a record does not name is left out of the statement: on insert it gets its default, on update it keeps its value. An explicit null writes NULL.
• Types. On PostgreSQL every value is cast to the column’s type, so a numeric, timestamptz, uuid, jsonb or enum column accepts the text a SQL source or a CSV file delivers, a JSON array goes into an array column, and 0/1 go into a boolean. A nested object or array is written as JSON text. A value the column cannot take fails the batch with the database’s message.
• key. Generates ON CONFLICT (key) DO UPDATE (PostgreSQL, SQLite) or ON DUPLICATE KEY UPDATE (MySQL/MariaDB). When one batch has several records with the same key, the last one wins.
• bulk_copy (PostgreSQL) works with columns: auto, but not together with key. A batch whose records name different columns is written in one transaction.
• A record that cannot be mapped — not a JSON object, or no field is a column — is rejected alone; the rest of the batch is written.

columns cannot be combined with insert_query or auto_create_table.

PostgreSQL / MySQL / MariaDB / SQLite

Schemes: postgres://, postgresql://, mysql://, mariadb://, sqlite://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

PostgreSQL CDC (logical replication)

Schemes: postgres-cdc://, pgcdc://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

publication_tables

[]

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

ClickHouse

Reads from or bulk-inserts into a ClickHouse table over ClickHouse’s HTTP interface (port 8123 by default; 8443 for HTTPS).

URL format

clickhouse://host[:port]?table=<name>[&database=<name>]

clickhouse:// is rewritten to http:// (and clickhouses:// to https://) before being handed to the ClickHouse client — the scheme only selects the endpoint kind on the CLI.

Config (YAML / library)

The same settings as a route endpoint in a config file, or in Route.from_config / fromConfig. Every URL query parameter is a field of the same name under clickhouse:.

output:
  clickhouse: { url: "http://localhost:8123", database: "analytics", table: "orders" }

The url is the HTTP interface (http:// or https://); clickhouse:// is CLI-only.

Examples

Bulk insert from a full-table Postgres read, one-shot:

mqb copy --drain \
  --from postgres://user:pass@localhost/app?table=orders \
  --to 'clickhouse://localhost:8123?table=orders&database=analytics'

Async insert for high-throughput streaming writes, continuous:

mqb copy \
  --from kafka://kafka.local:9092?topic=events \
  --to 'clickhouse://user:pass@ch.local:8123?table=events&database=analytics&async_insert=true'

Resumable, non-destructive read of an existing table into Kafka:

mqb copy \
  --from 'clickhouse://localhost:8123?table=events&database=analytics&cursor_column=id&cursor_id=events_export' \
  --to kafka://kafka.local:9092?topic=events

Each restart resumes from the last id seen instead of re-reading from the start. (Per-column mapping via columns is a map field, so it can’t be set from a query param — use a YAML route config for that.)

Key options