Kafka

lumnik Pro. Consume a Kafka-compatible topic — Apache Kafka, Redpanda, Azure Event Hubs — into a hub table. Each run reads from the stored position to the head the topic had when the run started, then completes. The hub keeps the offsets, and nothing is ever written back to the broker.


Type	`kafka`
Protocol	Kafka wire, JSON messages
Target	`connector.t_<name>`
Position	kept by the hub, per partition — nothing to set up on the broker
Delivery	a message read twice lands once
Schedulable	cron
Edition	Pro — the type is absent from the open edition, where a `kafka` manifest is refused at validate

Quickstart

Every screen on this page reads the demo broker that ships with the repo

docker compose -f docs/connectors/kafka/demo/docker-compose.yml up -d
python3 docs/connectors/kafka/demo/acme-topics.py        # needs: pip install confluent-kafka

The manifests point at host.docker.internal:29092: the hub runs in a container, where localhost is the hub itself. Topics and gestures: The demo broker.

# 10-quickstart.yaml — one topic, one partition, no auth.
#
# A broker address, a topic and a table name. Nothing to set up on the broker: the hub keeps
# each partition's position itself. A run reads up to the head the topic had when it started,
# then completes.
#
# Expect: 2 400 rows in connector.t_acme_partners; a second run reads 0.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-partners
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_partners
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.partners
  format: json

$ lm validate -f docs/connectors/kafka/demo/10-quickstart.yaml
✓ 10-quickstart.yaml — valid
$ lm apply -f docs/connectors/kafka/demo/10-quickstart.yaml
Applied connector "acme-kafka-partners" — 1 endpoint
$ lm connector run acme-kafka-partners
run scheduled: c5fc42d2-87f9-4328-acd2-b29213562a54

"Scheduled" is not "done": the run shows in lm run list a second or two later. Run it a second time, then look:

$ lm connector run acme-kafka-partners
run scheduled: 39ae9f8f-65f8-405a-a6cf-42afb1b4e7e1
$ lm run list --connector acme-kafka-partners
ID        CONNECTOR            ENDPOINT  STATUS     IN    OUT   SKIPPED  DEAD-LETTERS  STARTED           ENDED
39ae9f8f  acme-kafka-partners  default   Completed  0     0     0        0             2026-09-13 02:21  2026-09-13 02:21
c5fc42d2  acme-kafka-partners  default   Completed  2400  2400  0        0             2026-09-13 02:21  2026-09-13 02:21

The first run read the whole topic. The second found nothing new. STARTED / ENDED print in your terminal's time zone (these screens: Paris, UTC+2). schedule is evaluated in UTC.

What landed in connector.t_acme_partners — each top-level JSON field became a column:

 partner_id |           name            |   city    | credit_limit |      updated_at      
------------+---------------------------+-----------+--------------+----------------------
       1000 | Hydraulique Lyon 1000     | Lyon      |          500 | 2026-01-01T00:00:00Z
       1001 | Roulements Bruxelles 1001 | Bruxelles |          637 | 2026-01-01T00:01:00Z
       1002 | Fixations Milano 1002     | Milano    |          774 | 2026-01-01T00:02:00Z
       1003 | Pneumatique Porto 1003    | Porto     |          911 | 2026-01-01T00:03:00Z
(4 rows)

No schema to declare. Each column takes its type from the first messages that carry the field:

JSON value Column type

whole number bigint

number with a decimal point numeric

string (timestamps included) text

boolean boolean

object, array jsonb

A field added later: Schema evolution. - Nothing to provision on the broker. No consumer group; the position is stored by the hub — Partitions and the cursor. - A message read twice is one row — Storage layout.

The manifest

spec:
  name: erp_events                  # required — the table: connector.t_erp_events
  brokers: broker-1:9092,broker-2:9092   # bootstrap servers, literal…
  # brokers_env: KAFKA_BROKERS      # …or by name, resolved at run time (lm secret, then the hub's environment)
  topic: erp.events                 # required — one topic per connector
  format: json                      # the only value in v1: one JSON object per message
  poll_timeout_ms: 2000             # optional, [100, 60000]; what one poll may wait for the broker
  max_poll_records: 500             # optional; leave it at its default
  auth:                             # optional; omitted = no SASL handshake
    kind: sasl-plain                # none | sasl-plain
    username_env: KAFKA_USER        # names, never values
    password_env: KAFKA_PW
    security_protocol: SASL_SSL     # or SASL_PLAINTEXT for a SASL broker without TLS
  schedule: "*/5 * * * *"           # optional cron, evaluated in UTC by the hub scheduler
  window: "06:00-22:00"             # optional fence for the scheduled slots, UTC, end exclusive
  promote:                          # optional; a nested value as a typed column
    - { name: temperature_c, path: $.object.temperature, type: decimal }

There is no target and no mapping block: the table is connector.t_<name>, and messages land as they are. promote only adds columns: see A value nested in the message.

Three more keys belong to the Event Hubs preset: preset, namespace, connection_string_env.

Configuration reference

lm describe connector-type kafka prints every key with its meaning. The rules:

Key	Default	Rule
`name`	—	a plain identifier: letters, digits, underscore, not starting with a digit — it becomes `connector.t_<name>`
`brokers` / `brokers_env`	—	at least one is required; when both are set, `brokers` is used. `brokers_env` is a name the hub resolves when the run starts
`topic`	—	must match `[a-zA-Z0-9._-]+`
`format`	`json`	the only accepted value; Avro and Protobuf are not yet supported
`poll_timeout_ms`	`2000`	integer in `[100, 60000]`
`max_poll_records`	`500`	integer ≥ 1; leave it at its default
`auth.kind`	`none`	`none` or `sasl-plain`; `sasl-plain` requires `username_env` and `password_env`, and refuses a blank one
`auth.security_protocol`	`SASL_SSL`	`SASL_SSL` or `SASL_PLAINTEXT`. A wrong value is not caught by `lm validate`: the run fails when it connects
`schedule`	—	five-field cron, validated at apply, evaluated in UTC
`window`	—	`HH:MM-HH:MM`, UTC, end exclusive, may cross midnight; a manual run ignores it
`promote`	—	a list of `{ name, path, type, transform? }`, rest's grammar: `path` a JSONPath, `type` one of `text`, `bigint`, `integer`, `decimal`, `boolean`, `date`, `timestamptz` (not `jsonb`); a name used once, never a reserved one
`metadata.scopes`	—	the business scope(s) the table is askable in; the validator warns when none is declared

lm validate and lm apply enforce these rules, the value of security_protocol aside. Five mistakes in one file, all named at once:

# 90-refused.yaml — five mistakes in one manifest, every one named at validate.
#
# `lm validate` refuses what `lm apply` would refuse, field by field, before anything is
# written. This file is not to be fixed: a clean validate on it is the defect.
#
# Expect: exit 1, five errors and one warning — the warning is `poll_timeout`, a key the
#         connector reads nowhere (the real one is poll_timeout_ms).

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-refused
  connector_type: kafka
  scopes: [billing]

spec:
  name: 9acme                        # not a plain identifier: cannot become connector.t_9acme
  brokers: host.docker.internal:29092
  topic: acme/partners               # a slash is not in [a-zA-Z0-9._-]
  format: avro                       # v1 reads json only
  poll_timeout: 50                   # unknown key — the real one is poll_timeout_ms
  max_poll_records: 0                # must be a positive integer
  auth:
    kind: oauth                      # none | sasl-plain

$ lm validate -f docs/connectors/kafka/demo/90-refused.yaml
✗ 90-refused.yaml — 5 error(s), 1 warning(s)

  ERROR    spec.name
          must be a plain identifier (letters, digits, underscore; not starting with a digit) — it becomes the hub table connector.t_<name>

  ERROR    spec.topic
          topic must match [a-zA-Z0-9._-]+

  ERROR    spec.format
          format v1 supports only 'json'
          → Use: format: json

  ERROR    spec.max_poll_records
          max_poll_records must be a positive integer (1..2147483647)

  ERROR    spec.auth.kind
          auth.kind must be 'none' or 'sasl-plain'

  WARN     spec.poll_timeout
          unknown key — ignored by the runtime
          → Remove it, or check the spelling

Fix errors above before running 'lm apply -f docs/connectors/kafka/demo/90-refused.yaml'.
Error: validation failed: 5 error(s)

Read the warning: the key list is closed, inside auth too.

A misspelled key (poll_timeout), or a target copied from another manifest, is named and ignored. The manifest still applies.
A mapping block is refused: a check declared there would never run.
A misspelled security_protocol key is named and ignored too. The default SASL_SSL then stays, and a plaintext SASL broker fails in the TLS handshake.

How a run reads a topic

Partitions and the cursor

A topic has one or more partitions, each with its own offsets. The hub stores one position per partition. acme.orders is keyed by partner over six partitions:

# 20-six-partitions.yaml — a topic over six partitions.
#
# acme.orders is keyed by partner over six partitions. Each partition keeps its own position:
# the next run starts where each one stopped.
#
# Expect: 24 000 rows in connector.t_acme_orders.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-orders
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_orders
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.orders
  format: json

$ lm run list --connector acme-kafka-orders
ID        CONNECTOR          ENDPOINT  STATUS     IN     OUT    SKIPPED  DEAD-LETTERS  STARTED           ENDED
3463bf93  acme-kafka-orders  default   Completed  24000  24000  0        0             2026-09-13 02:21  2026-09-13 02:21

Each partition keeps its own position. The next run starts where each one stopped.
The first run reads everything the topic still retains. A partition added later is read from its beginning.
Nothing is set up on the broker: no consumer group, and each connector keeps its own position.

The head is the target

When a run starts, it reads the head of every partition (the high watermark), reads up to it, then completes. A message produced while the run reads belongs to the next run:

# 75-live.yaml — a topic that is being written to while the run reads it.
#
# Start the feeder, then the run:
#   python3 …/acme-topics.py feed --rate 200 --seconds 30 &
#   sleep 10; lm connector run acme-kafka-live
#
# A run is a catch-up, not a subscription: it reads to the head each partition had when the
# run STARTED, then completes. What arrived while it was reading belongs to the next run.
#
# Expect: Completed, well before the feeder stops, with `in` below what the feeder produced;
#         the next run reads the rest.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-live
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_live
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.live
  format: json

The feeder wrote 200 messages a second for 30 seconds; the run was started 10 seconds in:

$ python3 docs/connectors/kafka/demo/acme-topics.py feed --rate 200 --seconds 30 &
$ sleep 10; lm connector run acme-kafka-live
run scheduled: 644240c1-6741-407a-ae57-43aecdb19f1f
  ▪ fed 3986 messages at 200.0/s into acme.live
$ lm connector run acme-kafka-live
run scheduled: f5046f38-66ed-4598-8a18-2106c09d7072
$ lm run list --connector acme-kafka-live
ID        CONNECTOR        ENDPOINT  STATUS     IN    OUT   SKIPPED  DEAD-LETTERS  STARTED           ENDED
f5046f38  acme-kafka-live  default   Completed  2673  2673  0        0             2026-09-13 02:23  2026-09-13 02:23
644240c1  acme-kafka-live  default   Completed  1313  1313  0        0             2026-09-13 02:23  2026-09-13 02:23

The first run read the 1 313 messages present at its start, and completed while the feeder was still writing. The second read the 2 673 that arrived since. Together 3 986, the feeder's count.

A run is reported Partial in two cases only:

the broker had already deleted offsets the stored position pointed to (A log that moved under the cursor);
a long run of unreadable messages stopped the run early. The next run continues after them (Poison messages).

A run that simply ends below the head stays Completed: An empty poll.

Partial is the status to alert on — see Observability.

A run on a topic that has not moved completes at once with IN 0.

Continuous consumption

To keep the table current, declare a schedule. Each slot is a fresh catch-up from the stored cursor to the head of that moment:

# 30-cadence.yaml — a topic kept current on a cron, inside a quiet window.
#
# `schedule` is evaluated in UTC; each slot catches up from the last position. `window` fences
# the slots: one due outside it waits, and the run starts when the window opens. A manual
# `lm connector run` ignores the fence. A slot that fires while a run is still going is skipped: a message never lands twice.
#
# Expect: the same 2 400 rows as the quickstart, and SCHED filled in `lm connector list`.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-cadence
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_partners_cadence
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.partners
  format: json
  schedule: "*/5 * * * *"      # every five minutes, UTC
  window: "06:00-22:00"        # UTC, end exclusive — a slot due at night is skipped

$ lm apply -f docs/connectors/kafka/demo/30-cadence.yaml
Applied connector "acme-kafka-cadence" — 1 endpoint
$ lm connector list
NAME                TYPE   TENANT  SCHED        WINDOW       STATUS  LAST RUN
acme-kafka-cadence  kafka  1       */5 * * * *  06:00-22:00

Evaluated in UTC by the hub's scheduler. A re-apply without schedule unschedules the connector.
No overlap. A slot that fires while a run is active is skipped and recorded as Cancelled with the reason.
window fences the slots. A slot due outside "06:00-22:00" waits, and the run starts when the window opens. The end is exclusive, and the range may cross midnight ("22:00-06:00"). It gates the start only. A manual lm connector run ignores it.
An external trigger works too: your own cron plus lm connector run <name>.

One connector reads one topic, with no key or header filtering. To land the same messages twice, apply a second connector on the same topic with a different name.

Storage layout

Nothing to set up: the first run with a row to write creates connector.t_<name>.

Each top-level JSON field is a column, in the message's order.
A message read twice is one row: a producer retry, or a re-read after lm endpoint reset-cursor. Processes and webhooks fire for new rows only.
Each tenant sees only its own rows.

OUT counts every message read, even one the table already held. Re-read the whole topic, then compare with the table's row count:

$ lm endpoint reset-cursor acme-kafka-partners/default
$ lm connector run acme-kafka-partners
$ lm run list --connector acme-kafka-partners
ID        CONNECTOR            ENDPOINT  STATUS     IN    OUT   SKIPPED  DEAD-LETTERS  STARTED           ENDED
39b5d5bd  acme-kafka-partners  default   Completed  2400  2400  0        0             2026-10-24 22:36  2026-10-24 22:36
b80f7cda  acme-kafka-partners  default   Completed  2400  2400  0        0             2026-10-24 22:35  2026-10-24 22:35
$ lm data tables
MAPPING                               SCHEMA     TABLE                      ROWS
kafka:acme_partners                   connector  t_acme_partners            2400

OUT 2400 twice, and the table still holds 2 400 rows.

Deleting a kafka connector deletes its data

lm connector delete removes the connector's rows from connector.t_<name>, its cursor and its run history. The table goes with the last connector writing into it.
The topic is untouched. Re-applying the manifest reads it again from the beginning.
It is the only way to drop the rows — lm endpoint purge refuses a kafka endpoint:

$ lm endpoint purge acme-kafka-partners/default
Error: API error 400: This endpoint writes to a connector table whose rows are not tracked per endpoint, so it cannot be purged on its own. Deleting the connector removes its data, along with its run history.

Schema evolution

The producer changed its mind at message 1 201 and added two fields. acme.partners-drift:

# 50-drift.yaml — a schema that changes halfway through the topic.
#
# The first 1 200 messages carry six fields; from message 1 201 the producer added
# `vat_number` and `segment`. Nothing in this manifest says so: a field that appears later
# becomes a column when it does, empty for the rows written before.
#
# Expect: 2 400 rows, eight columns, `vat_number` filled 1 200 times.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-drift
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_partners_drift
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.partners-drift
  format: json

$ lm run list --connector acme-kafka-drift
ID        CONNECTOR         ENDPOINT  STATUS     IN    OUT   SKIPPED  DEAD-LETTERS  STARTED           ENDED
18c2f825  acme-kafka-drift  default   Completed  2400  2400  0        0             2026-09-13 02:21  2026-09-13 02:21

 column_name  | data_type
--------------+-----------
 partner_id   | bigint
 name         | text
 city         | text
 country      | text
 credit_limit | bigint
 updated_at   | text
 vat_number   | text
 segment      | text

 rows | with_vat_number | with_segment
------+-----------------+--------------
 2400 |            1200 |         1200

A new field becomes a new column during the run. Rows written before it hold NULL.
A field that disappears leaves its column NULL for later rows.
The new column takes its type from the first messages that carry it. If they all hold null, it lands as text.
A column keeps its type, with one exception: a whole number that receives a decimal. To change it otherwise once the producer has settled: delete the connector, apply it again, run. The table is created again.
The chat knows the new column after the run (Pro).

A whole-number column that receives a decimal

A column whose first messages held whole numbers is bigint. The first message with a decimal value widens it to numeric:

Earlier values do not change: 1000 stays 1000.
The column then reads NUMERIC in the schema cards, as a decimal column from the start does.
Under an entity, nothing to do. The entity's view is rebuilt with the widened column, in the same step.
Under a view you made (a report, a BI view), the run fails and names it:
```
connector.t_orders_x: amount cannot widen — view connector.v_sales_report reads it and the hub did not make it. Drop it, then run the connector again.
```
Drop the view, run the connector again, then create the view again. The next run reads again the messages that did not land: no row is lost.

Names the producer chose stay — and stay askable

A kafka manifest has no mapping block: the field names on the topic are the column names. acme.partners-raw carries typical ERP names:

# 55-raw.yaml — what the producer sends is what lands.
#
# acme.partners-raw carries field names a database column is not supposed to have
# ("Partner Name", "credit.limit", "MONTANT_€"), a nested object and an array. A kafka
# manifest has no mapping block, so nothing renames or flattens them: each top-level key
# becomes a column under that exact name (quoted), and a nested value lands as jsonb.
#
# Expect: 500 rows; columns "Partner Name", "credit.limit", "MONTANT_€", address (jsonb),
#         tags (jsonb) — and every one of them askable by name in chat.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-raw
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_partners_raw
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.partners-raw
  format: json

 column_name  | data_type
--------------+-----------
 partner_id   | bigint
 Partner Name | text
 credit.limit | bigint
 MONTANT_€    | numeric
 address      | jsonb
 tags         | jsonb

 partner_id |       Partner Name        | credit.limit | MONTANT_€ |                    address                     |             tags
------------+---------------------------+--------------+-----------+------------------------------------------------+-------------------------------
       1000 | Hydraulique Lyon 1000     |          500 |    0.0000 | {"city": "Lyon", "street": "0 rue de l'Usine"} | ["erp", "partner", "batch-0"]
       1001 | Roulements Bruxelles 1001 |          637 |    1.5000 | {"city": "Bruxelles", "street": "1 rue de l'Usine"} | ["erp", "partner", "batch-1"]

A space, a dot, a currency sign or a capital is kept. The hub and the chat quote the name in SQL:

$ lm ask --scope billing "How many rows are in connector.t_acme_partners_raw and what is the largest MONTANT_€?"
There are 500 rows in connector.t_acme_partners_raw. The largest MONTANT_€ is 748.5000.
  data as of: 2026-09-13T00:23:26.656851Z (last completed run)

…
  1 row returned

  SQL: SELECT COUNT(*), MAX("MONTANT_€") AS max_montant_eur FROM "connector"."t_acme_partners_raw"

address and tags landed as jsonb, values intact. The order of an object's keys is not kept: {"street", "city"} comes back {"city", "street"}.

A value nested in the message: `promote`

A LoRaWAN network server nests what a sensor measured. ChirpStack's Kafka integration publishes each uplink with the decoded reading under object, and every event type on the same topic. acme.probes carries 500 such events from a store's cold-room probes:

{"deduplicationId": "00000000-0000-4000-8000-000000000248", "time": "2026-01-01T04:08:00Z",
 "deviceInfo": {"deviceName": "PROBE-04", "devEui": "a84041000181c004", …},
 "fCnt": 49, "fPort": 2, "data": "AH0=", "object": {"temperature": -19.4}, …}

Without promote, object lands as one jsonb column. An entity or a Process when: compares columns, so the temperature is out of their reach. promote reads it into a typed column:

# 56-promote.yaml — a value nested in the message, promoted to a typed column.
#
# acme.probes is what ChirpStack's Kafka integration publishes for a store's cold-room probes: every
# event type on one topic, and the decoded reading under `object`. Each top-level key still lands as
# a column (object as jsonb); promote adds one typed column per entry, read by JSONPath.
#
# Expect: 500 rows; temperature_c numeric for the 449 readable uplinks, NULL for the 50 status
#         events and for uplink 250 (its codec sent "sensor-fault") — 1 cell error on the run.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-probes
  connector_type: kafka
  scopes: [cold-chain]

spec:
  name: acme_probes
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.probes
  format: json
  promote:
    - { name: temperature_c, path: $.object.temperature, type: decimal }
    - { name: device, path: $.deviceInfo.deviceName, type: text }
    - { name: measured_at, path: $.time, type: timestamptz }

$ lm apply -f docs/connectors/kafka/demo/56-promote.yaml
Applied connector "acme-kafka-probes" — 1 endpoint
$ lm connector run acme-kafka-probes
run scheduled: 8cf5834e-7601-4a0e-8468-d5c6f6118564
$ lm run get 8cf5834e
ID        CONNECTOR          ENDPOINT  STATUS     IN   OUT  SKIPPED  DEAD-LETTERS  STARTED           ENDED             EVALUATED
8cf5834e  acme-kafka-probes  default   Completed  500  500  0        0             2026-11-03 18:41  2026-11-03 18:41  yes
⚠ 1 cell(s) not delivered as declared (a hook, or a promoted value of the wrong type) — the rows were still written

The three promoted columns sit beside the top-level fields, with the declared types:

     column_name     |        data_type
---------------------+--------------------------
 deduplicationId     | text
 time                | text
 deviceInfo          | jsonb
 …
 object              | jsonb
 rxInfo              | jsonb
 txInfo              | jsonb
 temperature_c       | numeric
 device              | text
 measured_at         | timestamp with time zone
 margin              | bigint
 externalPowerSource | boolean
 batteryLevel        | numeric

  device  |      measured_at       | temperature_c |             object
----------+------------------------+---------------+---------------------------------
 PROBE-04 | 2026-01-01 04:08:00+00 |      -19.4000 | {"temperature": -19.4}
 PROBE-05 | 2026-01-01 04:09:00+00 |               |
 PROBE-01 | 2026-01-01 04:10:00+00 |               | {"temperature": "sensor-fault"}
 PROBE-02 | 2026-01-01 04:11:00+00 |      -20.3000 | {"temperature": -20.3}

A message without the value gets NULL, and that is not an error. PROBE-05 at 04:09 is a status event, which carries a battery level and no object.
A value that does not read as its type is NULL and one cell error. The run's warning above is uplink 250, whose codec sent "sensor-fault" where a number should be. The row is written and object keeps what was sent.
The column is typed from the first run, even when its first messages have no value: the declared type is used.
A promoted column named like a top-level field replaces it. { name: time, path: $.time, type: timestamptz } turns the text time into a timestamp. Do it before the first run: a column keeps its type.
transform: works as on rest, e.g. from_epoch for a number of seconds.

The two mistakes lm validate names:

# 57-promote-refused.yaml — two promote entries refused at validate.
#
# promote reads one scalar into one typed column. A nested object already lands as jsonb under its
# top-level field, and `id` is one of the hub's seven reserved columns.
#
# Expect: exit 1, two errors, each naming its entry.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-probes-refused
  connector_type: kafka
  scopes: [cold-chain]

spec:
  name: acme_probes_refused
  brokers: host.docker.internal:29092
  topic: acme.probes
  format: json
  promote:
    - { name: reading, path: $.object, type: jsonb }          # the whole object: it is already a column
    - { name: id, path: $.deduplicationId, type: text }       # id is a hub column: reserved

$ lm validate -f docs/connectors/kafka/demo/57-promote-refused.yaml
✗ 57-promote-refused.yaml — 2 error(s), 0 warning(s)

  ERROR    spec.promote[0].type
          kafka promotes a scalar — a nested value already lands as jsonb under its top-level field
          → Promote the value inside it, e.g. path: $.object.temperature, type: decimal

  ERROR    spec.promote[1].name
          'id' is a hub system column (_ingested_at, _mapping_version, _row_hash, _run_id, _source, id, tenant_id)
          → Name the promoted column otherwise

Fix errors above before running 'lm apply -f docs/connectors/kafka/demo/57-promote-refused.yaml'.
Error: validation failed: 2 error(s)

Seven column names are reserved

id, _row_hash, _ingested_at, _source, _run_id, _mapping_version, tenant_id are the hub's. A top-level JSON field with one of these names fails the run. id is the usual one:

# 58-reserved.yaml — a top-level `id` in the message.
#
# This is the REST demo's partner record verbatim, `id` included. `id` is one of the
# columns the hub adds to every connector.t_* table, and a kafka manifest has no mapping hook
# to rename a field — so the run is refused at write time, on a message that names all seven
# and says what each connector family can do about it.
#
# Expect: Failed — `source column 'id' collides with a hub system column`. No table.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-reserved
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_partners_reserved
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.partners-reserved
  format: json

$ lm run get 7ebc090f
ID        CONNECTOR            ENDPOINT  STATUS  IN  OUT  SKIPPED  DEAD-LETTERS  STARTED           ENDED
7ebc090f  acme-kafka-reserved  default   Failed  0   0    0        0             2026-09-13 02:21  2026-09-13 02:21
✖ run failed: source column 'id' collides with a hub system column (_ingested_at, _mapping_version, _row_hash, _run_id, _source, id, tenant_id); it must reach the hub under another name — jdbc: source.tables[].rename; csv: a map-columns row hook under spec.mapping; kafka: no rename exists, the field must be renamed in the topic payload

There is no rename on kafka: the producer must send the field under another name.

When the topic misbehaves

A bad message, a deletion, a missing topic, a log that moved, a broker that does not answer: each case, as lm shows it, in When the topic misbehaves.

Auth and secrets

`auth.kind`	Config	Wire
`none`	nothing — the default when `auth:` is omitted	plaintext, no SASL handshake
`sasl-plain`	`username_env` + `password_env`, optional `security_protocol`	SASL/PLAIN over `SASL_SSL` (default) or `SASL_PLAINTEXT`

brokers_env, username_env and password_env are names, resolved when the run starts: the lm secret registry of your tenant first, then the hub's environment.

# 62-brokers-env.yaml — the broker address by name, resolved at run time.
#
# `brokers_env` names an entry the hub resolves when the run starts: the `lm secret`
# registry for your tenant first, then the hub process's own environment. Run it once with
# nothing set to read the refusal, then:
#   lm secret set ACME_KAFKA_BROKERS       # host.docker.internal:29092
# and run again.
#
# Expect: run 1 Failed naming BOTH places it looked; run 2 Completed, 2 400 rows.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-brokers-env
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_partners_by_name
  brokers_env: ACME_KAFKA_BROKERS   # the value lives in the registry, never in this file
  topic: acme.partners
  format: json

Run before the name resolves, the refusal names both places it looked:

$ lm run get 3e808fcf
ID        CONNECTOR               ENDPOINT  STATUS  IN  OUT  SKIPPED  DEAD-LETTERS  STARTED           ENDED
3e808fcf  acme-kafka-brokers-env  default   Failed  0   0    0        0             2026-09-13 02:21  2026-09-13 02:21
✖ run failed: 'ACME_KAFKA_BROKERS' (brokers_env) resolves to nothing — no 'lm secret' entry for this tenant, and nothing of that name in the hub's own environment
$ lm secret set ACME_KAFKA_BROKERS          # prompts at the keyboard; what you type is not shown
Created secret "ACME_KAFKA_BROKERS"
$ lm connector run acme-kafka-brokers-env
$ lm run list --connector acme-kafka-brokers-env
ID        CONNECTOR               ENDPOINT  STATUS     IN    OUT   SKIPPED  DEAD-LETTERS  STARTED           ENDED
d071091c  acme-kafka-brokers-env  default   Completed  2400  2400  0        0             2026-09-13 02:22  2026-09-13 02:22
3e808fcf  acme-kafka-brokers-env  default   Failed     0     0     0        0             2026-09-13 02:21  2026-09-13 02:21

sasl-plain refuses, before connecting, a credential name that resolves to nothing or to a blank:

# 65-sasl.yaml — SASL/PLAIN credentials, by name.
#
# Neither the username nor the password is in this file: `username_env` and `password_env`
# name secrets (`lm secret set ACME_KAFKA_USER`, `lm secret set ACME_KAFKA_PW`).
# `security_protocol`: SASL_SSL (the default), or SASL_PLAINTEXT for a broker without TLS.
#
# The demo broker authenticates nobody, so this manifest cannot complete against it. It shows
# two refusals: an unresolved secret name, before any connection; a broker that speaks no SASL,
# after.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: acme-kafka-sasl
  connector_type: kafka
  scopes: [billing]

spec:
  name: acme_partners_sasl
  # The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
  brokers: host.docker.internal:29092
  topic: acme.partners
  format: json
  auth:
    kind: sasl-plain
    security_protocol: SASL_PLAINTEXT
    username_env: ACME_KAFKA_USER
    password_env: ACME_KAFKA_PW

$ lm run get 28d58a30
ID        CONNECTOR        ENDPOINT  STATUS  IN  OUT  SKIPPED  DEAD-LETTERS  STARTED           ENDED
28d58a30  acme-kafka-sasl  default   Failed  0   0    0        0             2026-10-19 01:39  2026-10-19 01:39
✖ run failed: 'ACME_KAFKA_USER' (username_env) resolves to nothing — no 'lm secret' entry for this tenant, and nothing of that name in the hub's own environment

Once both names resolve, a listener that does not speak SASL/PLAIN (the demo broker) refuses:

$ lm run get 9e962b07 > /dev/null
✖ run failed: Unexpected handshake request with client mechanism PLAIN, enabled mechanisms are []

SASL_SSL is the default security_protocol. A SASL listener without TLS needs SASL_PLAINTEXT.
The value of security_protocol is not checked at apply: a wrong one fails the run when it connects.
SASL/SCRAM, mTLS and OAuth are not yet supported.

Azure Event Hubs — the `event-hubs` preset

Event Hubs speaks Kafka with SASL/PLAIN: the username is the literal $ConnectionString, the password the namespace connection string. preset: event-hubs writes that config at apply. connector_type stays kafka.

# 20-event-hubs.yaml — Azure Event Hubs via the 'event-hubs' vendor preset.
#
# Apply and run:
#   lm apply -f docs/connectors/kafka/20-event-hubs.yaml
#   lm connector run event-hubs-telemetry
#
# How the event-hubs preset works
# --------------------------------
# connector_type stays 'kafka' — the preset just pre-wires the bootstrap
# address (namespace.servicebus.windows.net:9093) and SASL/SSL settings from
# the connection string, so you don't have to work those out by hand. No
# Kafka consumer group is created or needed — the hub tracks offsets itself,
# same as any other kafka connector.
#
# The 'connection_string_env' field NAMES the entry that holds the Event Hubs
# connection string (Primary Connection String from the Azure portal). Every *_env key
# resolves the same way: the lm secret registry for your tenant first, then the hub
# process's own environment (OS variable, then JVM system property).
#   lm secret set EVENTHUBS_CONN_STR          # or: export EVENTHUBS_CONN_STR="Endpoint=sb://..."
#
# Auth: Event Hubs over Kafka uses SASL/PLAIN with the fixed username "$ConnectionString"
# (the literal string, dollar sign included — it is not a secret). The preset reads it
# from the fixed name EVENT_HUBS_SASL_USERNAME:
#   lm secret set EVENT_HUBS_SASL_USERNAME    # or: export EVENT_HUBS_SASL_USERNAME='$ConnectionString'

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: event-hubs-telemetry         # unique connector name within the workspace
  connector_type: kafka              # real engine type; the vendor alias is spec.preset below
  scopes: [demo, azure]

spec:
  preset: event-hubs                 # vendor preset — expands to a real kafka config at apply time

  name: telemetry_events             # logical/table name → target table connector.t_telemetry_events

  namespace: my-namespace            # Azure Event Hubs namespace (without .servicebus.windows.net)

  connection_string_env: EVENTHUBS_CONN_STR   # env var holding the Primary Connection String

  topic: telemetry                   # Event Hub name (= Kafka topic name)

  format: json                       # v1: one JSON object per message

$ lm apply -f docs/connectors/kafka/20-event-hubs.yaml
Applied connector "event-hubs-telemetry" — 1 endpoint
$ lm connector get event-hubs-telemetry
id: 55016bf3-406b-43ee-86c8-92da12ca26f3
tenantId: 1
name: event-hubs-telemetry
typeId: kafka
enabled: true
config:
    auth:
        kind: sasl-plain
        password_env: EVENTHUBS_CONN_STR
        security_protocol: SASL_SSL
        username_env: EVENT_HUBS_SASL_USERNAME
    brokers: my-namespace.servicebus.windows.net:9093
    format: json
    kind: kafka
    name: telemetry_events
    topic: telemetry
tags:
    - demo
    - azure

The stored config is plain kafka: preset, namespace and connection_string_env are gone. Besides what the preset writes, it keeps name, topic, format, promote, poll_timeout_ms and max_poll_records. The two poll keys are checked as for plain kafka.

Set the two names the config reads:

lm secret set EVENTHUBS_CONN_STR          # the namespace Primary Connection String, Endpoint=sb://…
lm secret set EVENT_HUBS_SASL_USERNAME    # the fixed literal $ConnectionString

lm validate refuses an unknown preset (unknown source preset 'event-hubbs') and names each missing input (preset 'event-hubs' requires 'namespace').

Limitations

JSON only. One JSON object per message value. Avro, Protobuf and a Schema Registry are not yet supported; format accepts only json.
One topic per connector, and no filtering by message key or header: whatever the topic carries lands in the table. Message keys, headers and timestamps are not stored as columns.
No mapping block. No hooks run at ingestion — no rename, no filter, no lookup. A nested value reaches a column through promote; anything else must reach the hub under the name and shape it should land with.
No pre-flight. There is no lm source kafka test; the first run is the test. A missing topic or an unresolved secret fails it at once.
Append-only. A tombstone is set aside, never applied as a delete. A deletion that must reach the table has to travel as data.
sasl-plain only. No SASL/SCRAM, no mTLS, no OAuth.
Object key order inside a nested value does not survive jsonb.

CLI

lm validate -f <file.yaml>
lm apply -f <file.yaml>
lm describe connector-type kafka                    # the configuration reference, from the validator itself
lm connector run <connector>
lm connector get <connector>                        # the stored config, presets expanded
lm endpoint reset-cursor <connector>/default        # forget the position: the next run reads from the beginning
lm run list --connector <connector>
lm dlq list --run <run-id> · lm dlq get <id>

TUI

:sources — the KIND column reads kafka; Enter opens the connector's :runs
:runs — one row per run, Partial in its own colour beside Completed and Failed
:dlq — the quarantined messages, with coordinates

Full reference: CLI & TUI.

How to…

Keep a topic mirrored on a cadence

spec:
  schedule: "*/5 * * * *"
  window: "06:00-22:00"      # optional

Result: every five minutes, a catch-up to the head. A night slot waits for the window to open. A slot that meets a running run is skipped. See Continuous consumption.

Read a topic again from the beginning

lm endpoint reset-cursor <connector>/default
lm connector run <connector>

Result: reset-cursor prints nothing and exits 0; the run reads everything the topic retains. Rows already in the table are not duplicated but are counted in OUT.

Inspect the messages that did not land

lm run get <run-id>                 # the ⚠ line says how many, and the exact command to list them
lm dlq list --run <run-id>
lm dlq get <id>                     # coordinates, reason, raw value

Result: one entry per message, reason MalformedSource, coordinate topic:partition:offset. A tombstone's entry names the deleted key; a poison message's entry carries the raw value.

Read the same topic into two tables

Apply a second manifest with the same topic and a different name. Each connector keeps its own position and its own table.

Result: two connectors over one topic, useful when two teams need the same events under two scopes.

Move to Event Hubs

Replace brokers and auth with preset: event-hubs, namespace and connection_string_env, and set the two secrets. Result: the same run ledger, the same table; see the preset.

Troubleshooting

Symptom	Cause	Gesture
first run ingests far more than expected	no stored position yet — a full-retention read, by design	let it complete; later runs are incremental
a green run read less than expected	the broker answered slowly	raise `poll_timeout_ms`
a run ends `Partial`	the broker deleted messages before they were read, or a long run of unreadable messages stopped the run	`lm run get` names which. Deleted messages cannot be recovered: raise the topic's retention, or run more often
run fails: `kafka topic '…' does not exist on the broker`	no partition reported under that name, and the hub never creates one	first read: check `spec.topic` or create the topic; stored position: the topic was deleted or renamed on the broker
run fails: `'NAME' (brokers_env) resolves to nothing`	neither the registry nor the hub's environment holds it	`lm secret set NAME`, or export it where the hub runs
run fails: `Unexpected handshake request with client mechanism PLAIN`	the listener does not speak SASL/PLAIN — wrong port, or `auth` on a plaintext broker	check the broker's listener; drop `auth` for a plaintext one
run fails after ~60 s: `the kafka broker did not answer within the client's metadata timeout`	the bootstrap address answers nothing — wrong host, wrong listener, or a broker that is down	from the hub's container, prove the address; on Docker, `host.docker.internal`, not `localhost`
run fails at once: `No resolvable bootstrap urls given in bootstrap.servers`	the hub cannot resolve the broker host name (e.g. `host.docker.internal` on a hub not running in Docker)	use a host name the hub resolves
run fails: `connector.t_… is already written by connector '…'`	another connector of the tenant already writes `connector.t_<name>`	give this endpoint another `name` — see JDBC
run fails: `source column 'id' collides with a hub system column`	a top-level field shares a reserved name (`id`, `tenant_id`…)	the producer must emit it under another name — no rename exists on kafka
messages missing from the table, `DEAD-LETTERS` > 0	not a JSON object — dead-lettered per message, offsets advance	`lm dlq list --run <id>`; fix the producer and re-send, then `lm dlq replayed ID --new-run-id RUN` (the run that read it again) or `discard`
rows missing, `lm dlq get` says `a Kafka tombstone (key '…')`	the producer deleted that key; ingestion is append-only	nothing to repair — if deletions must reach the table, send them as data
a scheduled slot shows `Cancelled: skipped: another run is already active`	the previous catch-up was still running	nothing — the next slot catches up; widen the cadence if it repeats
a column is `NULL` for the first N rows and filled after	the producer added the field mid-topic — schema evolution	expected: the early messages never carried it
`lm validate` warns `unknown key` on `target`	copied from a jdbc or csv manifest; kafka has none	remove the block
`lm validate` refuses `spec.mapping`: `kafka has no mapping block`	copied from a csv manifest; no hook runs on kafka	remove the block; rename the field in the topic's payload
`unknown connector type: kafka`	the hub is the open edition	Kafka is a Pro connector

JSON value	Column type
whole number	`bigint`
number with a decimal point	`numeric`
string (timestamps included)	`text`
boolean	`boolean`
object, array	`jsonb`