Live streams

Collector: Kafka

Consume JSON, raw numbers or Sparkplug B from Kafka topics, with offsets committed only after the data is saved.

Kafka

The kafka source is a consumer for customers whose data already flows through Kafka (or Redpanda, Confluent, Event Hubs' Kafka endpoint). It reads topics with a consumer group and turns each message into values. It only reads: nothing is ever produced, and the only thing it writes back is its own committed offsets. Install it with pip install "kymoscope-collector[kafka]" (Sparkplug B included).

collector.yaml

sources:
  # JSON messages, one record per sample, over TLS with SCRAM
  - name: Customer bus
    type: kafka
    bootstrap_servers: [kafka1.customer.local:9093, kafka2.customer.local:9093]
    topics: [customer.telemetry]
    security_protocol: SASL_SSL
    sasl_mechanism: SCRAM-SHA-512
    username: kymoscope
    password: env:KAFKA_PASSWORD
    ca: customer-ca.pem
    items_path: samples[*]
    tag_path: tag
    value_path: value
    time_path: ts
    time_format: epoch_ms

  # Wide JSON: one message per machine, one column per tag, topics chosen by a pattern
  - name: Lines
    type: kafka
    bootstrap_servers: kafka.customer.local:9093
    topic_pattern: "^line\\..*"
    security_protocol: SSL
    values: {Speed: speed_rpm, Temp: oven.temp_c}
    start: "2026-10-24T06:00:00Z"

  # Sparkplug B bytes carried on a topic, the edge node id in the message key
  - name: UNS bridge
    type: kafka
    bootstrap_servers: kafka.customer.local:9093
    topics: [sparkplug]
    format: sparkplug
    security_protocol: SSL

Messages

Name	Type	Description
`format`	`json \| raw \| sparkplug` Default: `json`	json: mapped exactly like the REST source (items_path, time_path, time_format, tag_path and value_path, or values for a wide record; see REST and webhooks). raw: the message is one number as text, named by tag. sparkplug: a Sparkplug B protobuf payload; metrics are named "<message key>/<metric name>" (just the name when there is no key). Only named metrics are read, and aliases cannot be resolved because Kafka carries no birth state.
`tag`	`string` Default: `{topic}`	raw only: the channel name; {topic} and {key} (the message key as text) are replaced.
`tags, max_tags`	`list, number` Default: `, 5000`	Collect only these channels, or admit at most max_tags new ones (the rest are dropped and counted in tags_dropped).

Without a time_path (and for raw and Sparkplug without a metric time) the value time is the timestamp Kafka stored with the message, which is the producer's time unless the topic uses LogAppendTime. A message that cannot be decoded is skipped and counted in decode_errors. Avro and Schema Registry payloads are not supported: have the producer write JSON, or add a JSON topic with a stream processor.

Connection and security

Name	Type	Description
`bootstrap_servers`required	`host:port or list`	Any reachable brokers; the rest are discovered.
`topics, topic_pattern`required	`list, regex`	Exactly one of them. A pattern also picks up topics created later.
`group_id`	`string` Default: `kymoscope-<source name>`	The consumer group. Collectors sharing a group share the partitions; a new group starts at "start".
`security_protocol`	`SSL \| SASL_SSL \| PLAINTEXT \| SASL_PLAINTEXT` Default: `SSL`	The two plaintext protocols send data (and with SASL the password) unencrypted and need allow_insecure: true.
`sasl_mechanism`	`PLAIN \| SCRAM-SHA-256 \| SCRAM-SHA-512`	With SASL_SSL. username and password accept env:NAME and file:path. Kerberos and OAuth are not supported.
`ca, cert, key`	`paths`	A PEM CA for a private customer CA, and a client certificate with its key (cert and key go together).
`insecure`	`bool` Default: `false`	Do not verify the broker certificate (a warning is logged). Test systems only.
`start`	`latest \| earliest \| ISO time` Default: `latest`	Where a group with no committed offset begins; an ISO time starts at the first message at or after it. Once the group has committed offsets they win.
`max_batch, discover_s, lag_every_s, reconnect_max_s`	`numbers` Default: `500, 5, 10, 60`	Messages per read, how long discover listens, how often lag is measured, the longest wait between reconnects.

Ask the Kafka administrator for a read-only user: Read on the topics and on the consumer group, and Describe on the topics. The collector only needs to reach the brokers' advertised addresses on the listener you name, so check the advertised listeners when connecting from outside the cluster network.

No data is skipped

Offsets are not committed automatically. The collector commits a batch's offsets only after the values in it are on the wire or in the stream's disk buffer, so a crash or restart re-reads what was not yet saved (at-least-once) and never skips it. After a restart a few values can therefore arrive twice. When the collector stops, messages read since the last commit are read again on the next start. When several collectors share a group and a partition moves to another one, the offsets already saved are committed before it moves, so only values not yet saved (a second or so) are read again by the new owner.

discover, backfill and health

discover samples the last 50 messages of each partition (up to discover_s seconds) and lists the channels found. backfill reads the requested range by message timestamp with its own consumer that never commits, so it does not disturb the live group. Health reports connected, messages_per_s, values_per_s, decode_errors, lag (messages still to read, summed over the partitions, measured every lag_every_s), commits, commit_errors, rebalances, reconnects and tags_dropped.