Apache Kafka

Install

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

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

Starter config

source:
  type: kafka
  config:
    brokers: ""    # REQUIRED — Comma-separated bootstrap server list, e.g.
    topics: []    # REQUIRED — One or more topics to subscribe to.
    group_id: ""    # REQUIRED — Consumer group ID; required by librdkafka and used for partition assignment.
    # 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
    # auto_offset_reset: ""    # one of: earliest, latest
    # max_messages: 0    # Stop after this many messages have been consumed.
    # idle_timeout: 0    # Stop after this many seconds of no new messages.
    # poll_timeout: 1    # Single-poll timeout (advisory).
    # session_timeout: 30    # Kafka `session.timeout.ms`.
    # on_decode_error: null    # What the source does when a single message fails to decode.
    # extra_client_config: {}    # Raw librdkafka client properties to pass through.
    # batch_size: 1000    # Messages per emitted `StreamPage`.

Capabilities

Config fields