Kafka

Reads from or writes to a Kafka topic, through librdkafka. It works with anything that speaks the Kafka protocol: Apache Kafka, Redpanda, Confluent Platform.

URL format

kafka://broker[:port]?topic=<topic>[&group_id=<group>]

Rules that the table follows:

• One broker is enough. The client asks it for the rest of the cluster. A list of brokers is not a valid URL, so it goes in url=, with _ as a placeholder host.
• tls, producer_options and consumer_options take JSON. Put the whole URL in single quotes so the shell leaves the brackets and quotes alone.
• An unknown parameter is an error. kafka://host?topic=orders&foo=bar does not start; librdkafka options go in producer_options or consumer_options.

Config (YAML / library)

Every URL parameter is a field of the same name under kafka:. In a config file url is the bare host:port list, without kafka://.

Read a topic with a consumer group and write it to PostgreSQL:

orders_to_postgres:
  input:
    kafka:
      url: "broker1:9092,broker2:9092"
      topic: "orders"
      group_id: "orders-sync"
  output:
    sqlx:
      url: "postgres://app:secret@localhost/shop"
      table: "orders"
      columns: auto
      key: "id"

Write to a topic, keyed by customer, with eight partitions if the topic is new:

webhooks_to_kafka:
  input:
    http: { url: "0.0.0.0:8080", path: "/orders", method: "POST" }
  output:
    kafka:
      url: "localhost:9092"
      topic: "orders"
      partition_key: "customer_id"   # a metadata field; see Writing
      partitions: 8

Copy one topic to another cluster, with retries and a dead-letter topic for what fails:

mirror_orders:
  input:
    kafka: { url: "old-kafka:9092", topic: "orders", group_id: "mirror" }
  output:
    middlewares:
      - retry: { max_attempts: 5 }
      - dlq:
          endpoint:
            kafka: { url: "new-kafka:9092", topic: "orders.failed" }
    kafka: { url: "new-kafka:9092", topic: "orders" }

Reading

Whether a restart continues where it stopped depends on group_id:

• Offsets are committed after the output accepted the batch, never before. A crash in between delivers that batch again (at least once). Make the output idempotent or add deduplication where a duplicate matters.
• A record’s headers become the message’s metadata, and the record value is the payload, byte for byte. Kafka knows no format; a JSON value stays JSON.
• --drain ends at the end of the topic. With mqb copy --drain and a group_id, the copy stops when every partition has been read to its end. A topic that does not exist counts as empty.
• source_metadata: true adds the topic, partition and offset of each record as mqb.src.kafka_* metadata.
• consumer_options override the defaults. mq-bridge sets auto.offset.reset (earliest with a group, latest without) and enable.auto.commit=false; leave the second one alone.

Writing

• The topic is created if it is missing, with partitions partitions (default 6) and the cluster’s default replication factor. Set create_topic: false where topics are managed elsewhere or the user may not create them. An existing topic is never changed.
• The record key decides the partition. By default the key is the message id, which spreads records evenly. With partition_key: customer_id the key is the value of the metadata field customer_id, so all records of one customer keep their order. To key by a payload field, copy it into metadata first with a transform.
• Metadata is written as record headers, together with the message id in mq_bridge.message_id.
• Sent means acknowledged by all in-sync replicas. The producer is idempotent (acks=all), compresses with lz4 and waits 1 ms to fill a batch. delayed_ack: true stops waiting for the broker: faster, and a crash can lose what was not yet confirmed.
• producer_options override all of that. With acks set to 0 or 1, idempotence is switched off too, as librdkafka requires.

output:
  kafka:
    url: "localhost:9092"
    topic: "clicks"
    producer_options:
      - ["compression.type", "gzip"]
      - ["linger.ms", "20"]
      - ["message.max.bytes", "5000000"]

Authentication and TLS

input:
  kafka:
    url: "broker.example.com:9093"
    topic: "orders"
    group_id: "orders-sync"
    username: "svc-orders"
    password: "${KAFKA_PASSWORD}"
    tls:
      required: true
      ca_file: "/etc/ssl/kafka-ca.pem"       # leave out for a public CA
      cert_file: "/etc/ssl/client.pem"       # client certificate (mutual TLS)
      key_file: "/etc/ssl/client.key"

This is what Confluent Cloud expects, with the API key as username and the API secret as password. Another mechanism or protocol is set through the options, which are applied last:

consumer_options:
  - ["sasl.mechanism", "SCRAM-SHA-512"]

Without ca_file the broker certificate is checked against the CA certificates of the system.

Building it yourself? TLS and SASL are in the prebuilt mqb, the Python wheels and the Node packages. In your own Rust build they need the Cargo feature kafka-ssl (part of full) in addition to kafka. Without it such a connection fails at start with OpenSSL not available at build time.

Troubleshooting

Key options

Full field list: reference/kafka.md.