Apache Kafka

Install

cargo install faucet-cli --features sink-kafka
faucet init --sink kafka

# Cargo.toml — use it as a library
cargo add faucet-sink-kafka

Starter config

sink:
  type: kafka
  config:
    brokers: ""    # REQUIRED — Comma-separated `host:port` bootstrap brokers, passed straight through as librdkafka's `bootstrap.servers` (e.g.
    topic:
      type: fixed    # one of: fixed, from_path
      name: ""    # REQUIRED — Topic name.
      # --- Alternative variants — replace the block above with one of these ---
      # type: from_path
      # path: ""    # REQUIRED
    # auth: { type: none }    # one of: none, sasl_plain, sasl_scram, ssl, sasl_ssl, tls
      # --- Alternative variants — replace the block above with one of these ---
      # type: sasl_plain
      # config:
        # username: ""    # REQUIRED
        # password: ""    # REQUIRED
      #
      # type: sasl_scram
      # config:
        # mechanism: null    # REQUIRED
        # username: ""    # REQUIRED
        # password: ""    # REQUIRED
      #
      # type: ssl
      # config:
        # ca_path: ""    # REQUIRED
        # cert_path: ""    # REQUIRED
        # key_path: ""    # REQUIRED
        # key_password: ""
      #
      # type: sasl_ssl
      # config:
        # sasl: null    # REQUIRED
        # ssl: null    # REQUIRED
      #
      # type: tls
      # config:
        # ca_path: ""
    # value_format: { type: json }    # one of: json, raw_string, bytes, confluent_avro, confluent_protobuf, confluent_json_schema
      # --- Alternative variants — replace the block above with one of these ---
      # type: raw_string
      #
      # type: bytes
      #
      # type: confluent_avro
      # schema_registry:
        # url: ""    # REQUIRED
        # auth: null
        # cache_capacity: 1024
        # request_timeout: 10
      # bytes_encoding: null
      #
      # type: confluent_protobuf
      # schema_registry:
        # url: ""    # REQUIRED
        # auth: null
        # cache_capacity: 1024
        # request_timeout: 10
      #
      # type: confluent_json_schema
      # schema_registry:
        # url: ""    # REQUIRED
        # auth: null
        # cache_capacity: 1024
        # request_timeout: 10
      # validate: false
    # key_format: { type: json }    # one of: json, raw_string, bytes, confluent_avro, confluent_protobuf, confluent_json_schema
      # --- Alternative variants — replace the block above with one of these ---
      # type: raw_string
      #
      # type: bytes
      #
      # type: confluent_avro
      # schema_registry:
        # url: ""    # REQUIRED
        # auth: null
        # cache_capacity: 1024
        # request_timeout: 10
      # bytes_encoding: null
      #
      # type: confluent_protobuf
      # schema_registry:
        # url: ""    # REQUIRED
        # auth: null
        # cache_capacity: 1024
        # request_timeout: 10
      #
      # type: confluent_json_schema
      # schema_registry:
        # url: ""    # REQUIRED
        # auth: null
        # cache_capacity: 1024
        # request_timeout: 10
      # validate: false
    # value_schema: ""    # Schema text (Avro `.avsc` JSON, Protobuf `.proto`, or JSON Schema) for the **value** when `value_format` is a Confluent Schema Registry format (`confluent_avro` / `confluent_protobuf` / `confluent_json_schema`).
    # key_schema: ""    # Schema text for the **key**, used when `key_format` is a Confluent Schema Registry format.
    # key_path: ""    # JSONPath to the field that becomes each message's key (the first match wins).
    # partition_path: ""    # JSONPath to an explicit destination partition for each message.
    # headers_path: ""    # JSONPath to a flat JSON object whose entries become Kafka message headers (values are stringified).
    # on_key_error: null    # What the sink does when key/partition extraction fails for a record.
    # compression: ""    # Producer-side compression for outbound batches. — one of: none, gzip, snappy, lz4, zstd
    # acks: null    # Broker acknowledgement level required before a send is considered delivered — librdkafka's `acks`.
    # idempotent: true    # Enable the librdkafka idempotent producer (`enable.idempotence`), which de-duplicates broker-side retries so a retried send cannot append the message twice.
    # linger: 0    # How long the producer waits for more messages before sending a batch (librdkafka `linger.ms`).
    # batch_size: 1000    # Maximum number of in-flight `FuturesUnordered` send futures during a single `Sink::write_batch` call.
    # message_timeout: 30    # Delivery deadline for a single message (librdkafka `message.timeout.ms`), in **whole seconds**.
    # max_in_flight: 100    # Hard ceiling on concurrent in-flight sends inside one `Sink::write_batch` call.
    # queue_full_backoff: 0    # How long to wait before retrying a send that librdkafka rejected with `QueueFull`.
    # queue_full_max_retries: 3    # How many times to retry a `QueueFull` send before failing the batch.
    # exactly_once: null    # Exactly-once knobs, grouped (#654 M20).
    # transactional_id_prefix: ""    # **Deprecated** — use `exactly_once.transactional_id_prefix`.
    # commit_token_topic: "__faucet_commit_token"    # **Deprecated** — use `exactly_once.commit_token_topic`.
    # commit_token_topic_partitions: 1    # **Deprecated** — use `exactly_once.commit_token_topic_partitions`.
    # commit_token_topic_replication: -1    # **Deprecated** — use `exactly_once.commit_token_topic_replication`.
    # extra_client_config: {}    # Raw librdkafka producer properties, applied **last** so they override everything this config derives (`acks`, `enable.idempotence`, `compression.type`, `linger.ms`, `message.timeout.ms`, `queue.buffering.max.messages`).

Capabilities

Config fields