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 per-connector query parameters used by copy (name, type, default, required, description), see the generated URL parameter reference and the hand-written connector pages. 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).

Consumer vs. subscriber, and nack support

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 — set consume: capture_new to watch an existing collection for changes from now on, or consume: capture_all to read existing documents first and then keep capturing (no gap, at-least-once). It emits insert/update/replace/delete tagged with mongodb.operation and checkpoints under cursor_id. Change streams need a replica set.
• 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. The default consumer buys exclusivity via four round trips per batch; where you only need a single-reader bulk read or ETL pass, use capture_all — it is roughly 5x faster (500k docs, standalone MongoDB: consumer → 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.

Cloud object storage (S3 / GCS / Azure)

The object_store endpoint (alias s3) reads and writes Amazon S3, Google Cloud Storage, Azure Blob, Cloudflare R2, and anything else the object_store crate speaks. Enable with the object-store feature. Credentials and backend options come from the environment (AWS_ACCESS_KEY_ID, AWS_REGION, GOOGLE_SERVICE_ACCOUNT, AZURE_STORAGE_ACCOUNT, …); the URL scheme picks the backend (s3://, gs://, az://).

• 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.

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).