Kafka
Produces to or consumes from a Kafka topic via librdkafka.
URL format
kafka://broker[:port]?topic=<topic>
kafka:// is stripped before being handed to librdkafka’s
bootstrap.servers, which expects a bare host:port list, not a URI — the
scheme only selects the endpoint kind on the CLI. For a multi-broker
cluster, use the ?url= escape hatch (see
Quick Start):
kafka://_/?url=broker1:9092,broker2:9092&topic=orders.
Examples
Forward an MQTT stream into a Kafka topic, continuous:
mq-bridge-app copy \
--from mqtt://broker.local:1883?topic=sensors/+/temperature \
--to kafka://kafka.local:9092?topic=sensor-readings
Consume with a durable consumer group, continuous:
mq-bridge-app copy \
--from 'kafka://kafka.local:9092?topic=orders&group_id=mqb-orders-sync' \
--to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'
Without group_id, the consumer runs in ephemeral subscriber mode
(unique group ID, starts from the latest offset).
SASL-authenticated broker:
mq-bridge-app copy \
--from postgres://user:pass@localhost/app?table=orders \
--to 'kafka://kafka.local:9093?topic=orders&username=svc&password=secret'
TLS (an object-typed field, not a scalar) can’t be set via a query param;
use the ?url= escape hatch to pass a librdkafka connection string with TLS
options embedded.
Key options
| Option | Purpose |
|---|---|
topic | Topic to produce to or consume from. |
group_id | Consumer group ID; omit for ephemeral subscriber mode. |
username / password | SASL authentication. |
tls | TLS configuration (object; not settable via a scalar query param — use ?url= to pass driver-level TLS options). |
delayed_ack | Publisher-only: don’t wait for broker acknowledgement. |
Full field list: reference/kafka.md.