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 owns the offsets, and nothing is ever written back to the broker.
| Type | kafka |
| Protocol | Kafka wire, JSON messages |
| Target | connector.t_<name> |
| Offsets | hub-owned, per partition — no consumer group on the broker |
| Delivery | at-least-once into a table deduplicated by content hash |
| Schedulable | cron |
| Edition | Pro — the type is absent from the open edition, where a kafka manifest is refused at validate |
A topic is the third kind of legacy source after files and databases: the ERP already publishes its events, and nobody downstream has a table of them. This connector turns a topic into one — the same tenant-isolated, askable table the CSV and JDBC connectors produce, with the same dead-letter queue and the same run ledger — without a consumer group to provision, an offset to commit, or a schema to declare.
Features
- Hub-owned offsets, no consumer group. The connector assigns partitions and seeks to positions it stores itself; the broker never sees a group, a commit or a rebalance.
- A run is a catch-up, not a subscription. It measures the head of
every partition once, drains to it and completes — on a topic that is being written to while it
reads. A run that stopped short of its target is reported
Partial, neverCompleted. - Auto-schema, and it evolves. Column types are inferred from the first
chunk; a field the producer adds later becomes a
textcolumn mid-run, and nested values land asjsonb. - Names the producer chose are kept — and stay askable.
Partner NameorMONTANT_€becomes a column under that exact name, and the chat quotes it. - Per-message dead-letter queue. A message that is not a JSON object is set aside with its coordinates and raw value, its offset advances, and the run goes on. Tombstones are set aside under a reason that names the deleted key.
- What the broker deleted is named. A stored position
that retention has overtaken yields a WARN with the exact lost offset range and a
Partialrun — never a green run over a silent gap. - A read never writes to the source. The consumer is built with automatic topic creation off, so asking about a topic cannot create it; an unknown topic is refused by name.
- Credentials by name. Brokers, usernames and passwords are named, never
written: the
lm secretregistry first, the hub's environment second, and a refusal that names both places it looked. - Vendor presets.
preset: event-hubsexpands a namespace and a connection string into a full SASL/SSL kafka config at apply time. - Schedule and window. A cron keeps the table current; a UTC window fences the slots; a per-endpoint lock makes overlap impossible.
Quickstart
Every screen on this page reads a demo broker that ships with the repo
One Redpanda in a container and one Python file that fills it with eleven topics over the same 2 400 fictional ACME partners the REST demo serves. Start it, fill it, and the broker is ready to be read:
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, because the hub runs in a container and
localhost inside it is that container. Which topic carries which shape, and the three
gestures that change the broker: The demo broker.
# 10-quickstart.yaml — one topic, one partition, no auth.
#
# The whole manifest is a broker address, a topic and a table name. There is no consumer
# group to provision on the broker: the hub keeps each partition's read position itself, in
# its own cursor table, the way it keeps a JDBC watermark. A run reads from that position to
# the head the topic had when the run 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
✓ docs/connectors/kafka/demo/10-quickstart.yaml — valid
$ lm apply -f docs/connectors/kafka/demo/10-quickstart.yaml
{"name":"acme-kafka-partners","connectorId":"4cc236fc-e37f-4df7-8bc5-0173021923e7","endpointsApplied":1}
$ lm connector run acme-kafka-partners
run scheduled: c5fc42d2-87f9-4328-acd2-b29213562a54
"Scheduled" is not "done" — the run ledger is where the result lands, a second or two later. Run it a second time before looking:
$ 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 the stored position already at the head
and read nothing. STARTED / ENDED print in your terminal's time zone (these screens:
Paris, UTC+2); the hub stores UTC, and 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)
- Nothing was declared about the schema. Types are inferred from the first chunk: JSON
whole numbers land as
bigintand numbers carrying a decimal point asnumeric(18,4), strings astext(timestamps included —updated_atis the string the producer sent), booleans asboolean, objects and arrays asjsonb. The columns follow the message's own field order. Schema evolution is what happens when the producer changes its mind. - Nothing was provisioned on the broker. No consumer group exists for this connector; the position is a row in the hub's cursor table — Partitions and the cursor.
- The table deduplicates by content. 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, ≥ 1; clamped to the 500-row chunk
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
There is no target block and no mapping block: the table is connector.t_<name> and the
messages land as they are. A manifest started from a jdbc or csv one gets those two keys named
at validate rather than silently stored — see the warning below.
Three more keys exist and belong to the Event Hubs preset:
preset, namespace, connection_string_env.
Configuration reference
lm describe connector-type kafka prints the schema the validator enforces — type, importance
and the one-line meaning of every key, nested blocks included:
$ lm describe connector-type kafka
Connector type: kafka
name REQUIRED [STRING, HIGH]
Logical connector/table name (target table connector.t_<name>)
brokers optional [STRING, HIGH]
Comma-separated bootstrap servers
brokers_env optional [STRING, HIGH]
Env var holding the bootstrap servers
topic REQUIRED [STRING, HIGH]
Kafka topic to consume
format optional [STRING, MEDIUM]
Record value format (default json)
poll_timeout_ms optional [LONG, LOW]
Consumer poll timeout in ms [100, 60000]
max_poll_records optional [LONG, LOW]
Max records per poll (positive integer; clamped to the engine chunk size)
auth optional [NESTED, MEDIUM]
Broker authentication (kind: none | sasl-plain)
kind optional [STRING, HIGH]
Authentication method (none | sasl-plain)
username_env optional [STRING, HIGH]
SASL username source: lm secret name, else env var
password_env optional [STRING, HIGH]
SASL password source: lm secret name, else env var
security_protocol optional [STRING, MEDIUM]
Wire protocol for sasl-plain (SASL_SSL | SASL_PLAINTEXT)
schedule optional [STRING, MEDIUM]
Cron expression for recurring catch-up runs — the hub scheduler fires it, evaluated in UTC
window optional [STRING, MEDIUM]
Fence scheduled runs to a UTC time window, HH:MM-HH:MM (a manual run bypasses it)
preset optional [STRING, LOW]
Vendor source preset expanded at apply time (e.g. event-hubs)
namespace optional [STRING, LOW]
Input of the event-hubs preset: the Azure Event Hubs namespace
connection_string_env optional [STRING, LOW]
Input of the event-hubs preset: env var holding the namespace connection string
| Key | Default | Rule |
|---|---|---|
name |
— | a plain identifier: letters, digits, underscore, not starting with a digit — it becomes connector.t_<name> |
brokers / brokers_env |
— | exactly one of the two must be present; 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; a value above the chunk size is clamped to it — see Chunks |
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 |
lands verbatim in the consumer's security.protocol; it is not validated at apply |
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 |
metadata.scopes |
— | the métier scope(s) the table is askable in; the validator warns when none is declared |
Every rule above is enforced at lm validate and again at lm apply, so a manifest the
command blesses is one the hub accepts. 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
✗ docs/connectors/kafka/demo/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)
The warning is the one worth reading twice. The key list is closed: a key the connector reads
nowhere — poll_timeout for poll_timeout_ms, target or mapping copied from another
manifest — is named rather than accepted and stored. It stays a warning, the manifest still
applies, but poll_timeout: 50 would otherwise read back from lm connector get as a 50 ms
poll on a connector that polls for 2 000. The sweep reaches inside auth, where a misspelled
security_protocol leaves the default SASL_SSL in place and a manifest meant for a plaintext
SASL broker dies in a TLS handshake with the typo as the cause.
How a run reads a topic
Partitions and the cursor
A Kafka topic is one or more partitions, each an ordered log where every message has an
offset. A consumer's position is therefore not one number but one per partition — and that is
exactly what the hub stores. acme.orders is keyed by partner over six partitions:
# 20-six-partitions.yaml — the per-partition cursor.
#
# acme.orders is keyed by partner over six partitions, so one partner's orders sit together
# and the topic is read six offsets at a time. The stored cursor is a map {partition -> next
# offset}: a single-partition topic could never show that the map is really per partition.
#
# Expect: 24 000 rows in connector.t_acme_orders and a cursor carrying SIX keys whose
# offsets sum to 24 000.
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
The position after that run, read out of connector.connector_cursor — the same table that
holds a JDBC watermark or a CSV file position:
{"type": "KafkaOffset", "offsets": {"0": 3880, "1": 3770, "2": 3930, "3": 4120, "4": 4190, "5": 4110}}
Six keys, one per partition, each the next offset to read; they sum to 24 000. The next run seeks every partition to its stored offset and continues from there. A partition with no stored offset — the first run, or a partition the topic gained since — is read from its beginning, so run #1 reads everything the topic still retains.
What the broker does not see: no group.id, no offset commit, no rebalance. The consumer is
built with enable.auto.commit=false, assigns its partitions and seeks explicitly, and is closed
after every chunk. Nothing has to be provisioned or cleaned up on the Kafka side, which is also
why one connector cannot trample another's position — each has its own cursor row.
The head is the target
When a run starts, the hub asks the broker for the head of every partition (the high watermark) and writes it to the log:
INFO run 3463bf93-… target: partition offsets [0: 3880, 1: 3770, 2: 3930, 3: 4120, 4: 4190, 5: 4110]
That is the run's whole definition of done: read from the stored position to those offsets, then complete. A message produced while the run is reading belongs to the next run. On a topic that is written to continuously — which is every real topic — this is what makes a run end at all:
# 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 found 1 313 messages at its start, read exactly those in 270 ms and completed while the feeder was still writing. The second read the 2 673 that had arrived since. Together, 3 986 — the feeder's own count — and no row twice.
A run that ends below the target it gave itself is reported Partial, not Completed.
It has two causes, both certain: the run began by skipping offsets its stored position was
entitled to because the broker had already deleted them (below),
or the reader stopped on its own chunk budget while the target said there was more (a topic
feeding only unparsable messages, below). Being below the
head is not itself partial: the head is an upper bound on readable records — a transactional
producer leaves a commit marker there, compaction leaves gaps — so a run that read everything
readable can end an offset short of it, correctly. Partial is the status worth alerting on:
the run wrote everything it read, kept its position, and simply did not read everything there
was; see Observability.
Chunks and the two poll knobs
Every connector run writes in 500-row chunks, each in its own transaction with the cursor persisted beside it. On kafka a chunk is filled by consumer polls, and the two knobs shape them:
max_poll_recordsis how many records one poll may return. The default is the chunk size, and a larger value is clamped to it —5000still gives 6 chunks for 2 400 messages, exactly like the default. A smaller one gives more chunks and more transactions:100gives 25. Measured onacme.partners: 6 chunks in 865 ms against 25 in 1 042 ms.poll_timeout_msis how long one poll waits for the broker. A fresh consumer is built for every chunk, so the first poll of a run pays metadata, connection and fetch out of that budget. The default of 2 000 ms is generous for a broker on the same network; the floor of 100 ms is legal and, on a slow link, is how a run reads nothing and ends.
A second run on a topic that has not moved does not poll at all — the stored position already
sits at the target, so it completes in a few tens of milliseconds 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 by the hub's own scheduler; each slot is a fresh catch-up
# from the stored cursor. `window` fences the slots — one due outside it is skipped, never
# queued — and a manual `lm connector run` ignores the fence. A slot that fires while a run
# is still active on the endpoint is skipped too (per-endpoint advisory lock), so an
# aggressive cadence degrades to back-to-back catch-ups, never to double ingestion.
#
# 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
{"name":"acme-kafka-cadence","connectorId":"973926e1-8936-44ab-918c-3fb06f05b64d","endpointsApplied":1}
$ lm connector list
NAME TYPE TENANT SCHED WINDOW STATUS LAST RUN
acme-kafka-cadence kafka 1 */5 * * * *
acme-kafka-partners kafka 1 Completed 2026-09-13 02:21
- Evaluated in UTC by the hub's own scheduler — no OS cron. A re-apply without
scheduleunschedules the connector. - Overlap is impossible. A slot that fires while a run is still active on the endpoint is
skipped (per-endpoint advisory lock) and recorded as
Cancelledwith the reason. Because a run drains to the head it found and completes, a busy topic does not hold the lock for ever: an aggressive cadence degrades to back-to-back catch-ups, never to double ingestion. windowfences the slots."06:00-22:00"skips — never queues — a slot due at night; the end is exclusive, and the range may cross midnight ("22:00-06:00"). The fence gates a run's start: a run that began inside the window drains to the head it found and completes. A manuallm connector runalways ignores it — an operator asking explicitly is not the scheduler.- An external trigger works too: your own cron plus
lm connector run <name>.
One connector reads one topic. There is no message-key or header filtering: whatever the
topic carries lands whole in that connector's table. A topic that multiplexes several event types
needs one connector per topic on the producer side — or a second connector on the same topic
with a different name, if the same messages must land twice.
Storage layout
Nothing to set up: the table is created by the first run that has a row to write. This section is reference, for when you need it.
Table "connector.t_acme_partners"
Column | Type | Collation | Nullable | Default
------------------+--------------------------+-----------+----------+-------------------------------------------------------
id | bigint | | not null | nextval('connector.t_acme_partners_id_seq'::regclass)
_row_hash | text | | not null |
_ingested_at | timestamp with time zone | | not null | now()
_source | text | | |
_run_id | uuid | | |
_mapping_version | integer | | |
tenant_id | bigint | | not null |
partner_id | bigint | | |
name | text | | |
city | text | | |
country | text | | |
credit_limit | bigint | | |
updated_at | text | | |
Indexes:
"t_acme_partners_pkey" PRIMARY KEY, btree (id)
"t_acme_partners__row_hash_key" UNIQUE CONSTRAINT, btree (_row_hash)
"t_acme_partners_idx_tenant_id" btree (tenant_id)
Policies:
POLICY "tenant_isolation"
USING ((tenant_id = (NULLIF(current_setting('app.current_tenant'::text, true), ''::text))::bigint))
- The seven columns above the line are the hub's:
id,_row_hash,_ingested_at,_source,_run_id,_mapping_version,tenant_id— the same seven everyconnector.t_*table carries. The rest came from the messages, in the order the message lists its fields. _row_hashis the dedup. It is a hash of the row's content,UNIQUE, and a kafka write isON CONFLICT DO NOTHING: a message read twice — a producer retry, a cursor reset — is one row.data.row.changedevents, where a Process, a webhook or a notification rule subscribes to the table, fire for genuinely new rows only.- The row-level security policy is what keeps one tenant's rows invisible to another.
-
records_outcounts writes, including the ones the table refused. Reset the cursor and run again:$ 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 dc41ba74 acme-kafka-partners default Completed 2400 2400 0 0 2026-09-13 02:25 2026-09-13 02:25OUT 2400, and the table still holds exactly 2 400 rows. The re-read is a no-op the ledger reports as a full write, the same way every connector family reports a re-run; the witness that nothing duplicated is the table's row count, not theSKIPPEDcolumn.
Deleting a kafka connector deletes its data
lm connector delete drops connector.t_<name> and the cursor — the same erasure gesture as
the other connector families (a kafka run writes rows and the schema card, not RAG chunks, so
there is nothing else to drop). The topic is untouched: the hub never writes to it, and
re-applying the manifest reads it again from the beginning.
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: the table's columns are
# inferred from the messages, and a field that appears later becomes a column when it does
# (ALTER TABLE ADD COLUMN, mid-run), NULL for the rows written before it existed.
#
# 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
The table was created from the first chunk's six fields; when a message carrying vat_number
arrived in a later chunk, the column was added (ALTER TABLE ADD COLUMN) and the run went on.
Rows written before it existed hold NULL there — nothing was re-read and nothing was lost. A
field that disappears from later messages simply leaves its column NULL for those rows.
The new column is typed like any other: from the chunk that introduces it, by the same
inference the first chunk uses. A whole number arriving on day two lands bigint, one with a
decimal point numeric, a boolean boolean, a nested value jsonb. The two fields above are
strings, so the screen cannot show the difference — but a credit_limit appearing later would.
The added columns come after the ones created first, in the order their chunk presents them.
The bound is the chunk, exactly as it is at creation: inference reads the messages in hand, so a
field that is present but null in all of them still lands text. Each run that adds columns
says so in the hub log, naming them and the type each one got, because a decision taken from a
partial sample is one you may want to revisit — the gesture for that is lm endpoint reset-cursor
followed by a re-read, once the producer has settled.
Each run that changed the table's shape refreshes the scope's schema card, so the chat knows the new column before the next question is asked.
Names the producer chose stay — and stay askable
A kafka manifest has no mapping block — no hooks run at ingestion, nothing renames,
flattens or filters — so the field names on the topic are the column names in the table.
acme.partners-raw carries the shapes an ERP export is fond of:
# 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, a capital — each is a quoted identifier in PostgreSQL, and the hub quotes it everywhere it generates SQL. The schema card the chat reads names every column quoted too, so generated SQL quotes it as well:
$ 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)
SQL: SELECT COUNT(*), MAX("MONTANT_€") AS max_montant_eur FROM "connector"."t_acme_partners_raw"
The nested address and the tags array landed as jsonb, values intact. One thing does not
survive: object key order — {"street", "city"} was produced and {"city", "street"} came
back, because jsonb stores an object as a set of keys. Order that matters travels in an array.
A kind:Workflow lifecycle can be declared over
such a column just as it is; see also
Names the source chose.
Seven column names are reserved
id, _row_hash, _ingested_at, _source, _run_id, _mapping_version, tenant_id — the
hub adds these to every connector.t_<name> table. A top-level JSON field sharing one of them is
refused at write time, on a brand-new table and on an existing one alike. id is the one you
will meet: acme.partners-reserved carries the REST demo's partner record verbatim, id included.
# 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 seven
# 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
Because there is no mapping block, there is no in-product rename: the field has to reach the hub under another name, which means changing what the producer puts on the topic.
When the topic misbehaves
A broker does not lie the way an HTTP server can — it answers honestly every time. What misbehaves is the topic: a message that is not what the schema expects, a deletion, a name that does not exist, and time: a log start that moved under the stored position, a poll that came back empty for the wrong reason.
Poison messages — the per-message DLQ
acme.partners-poison holds 1 000 messages of which 143 are not a JSON object — a truncated
object, a JSON array, a JSON scalar:
# 40-poison.yaml — the per-message dead-letter queue.
#
# acme.partners-poison holds 1 000 messages of which exactly 143 are not a JSON object: a
# truncated object, a JSON array, a JSON scalar, at every index where i % 7 == 3. Poison is
# placed deterministically, never at random — a source whose corpus changes between runs
# cannot tell a fix from a coincidence.
#
# Expect: Completed, in=1000 out=857 skipped=143, 143 dead-letters under MalformedSource,
# 857 rows in the table, and the identity 1000 = 857 + 143 by hand.
apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
name: acme-kafka-poison
connector_type: kafka
scopes: [billing]
spec:
name: acme_partners_poison
# The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
brokers: host.docker.internal:29092
topic: acme.partners-poison
format: json
$ lm run get bd4d3df9
ID CONNECTOR ENDPOINT STATUS IN OUT SKIPPED DEAD-LETTERS STARTED ENDED
bd4d3df9 acme-kafka-poison default Completed 1000 857 143 143 2026-09-13 02:21 2026-09-13 02:21
⚠ 143 rows quarantined — see: lm dlq list --run bd4d3df9-a25b-4844-9a46-d25ef97d5dab
$ lm dlq list --run bd4d3df9-a25b-4844-9a46-d25ef97d5dab
ID CONNECTOR RUN REASON ATTEMPTS MESSAGE
4d3fb883 acme-kafka-poison bd4d3df9 MalformedSource 1 acme.partners-poison:0:997:997 — message value is not a JS…
844d1f31 acme-kafka-poison bd4d3df9 MalformedSource 1 acme.partners-poison:0:990:990 — invalid JSON: Unexpected …
c1a2e423 acme-kafka-poison bd4d3df9 MalformedSource 1 acme.partners-poison:0:983:983 — message value is not a JS…
$ lm dlq get 844d1f31
ID: 844d1f31-5a15-401e-8f5b-4564bd89f89f
Connector: acme-kafka-poison (98bea7c4-7052-4186-838a-79110ffa58e2)
Run: bd4d3df9-a25b-4844-9a46-d25ef97d5dab
Reason: MalformedSource
Attempts: 1
Message: acme.partners-poison:0:990:990 — invalid JSON: Unexpected end-of-input within/between Object entries
Payload: {"raw":["{\"broken\": "],"file":"acme.partners-poison:0:990","line":990}
- The conservation identity holds by hand:
IN 1000 = OUT 857 + SKIPPED 143, and the table holds 857 rows. A poison message is read, counted, set aside — never silently dropped. - The coordinate is
topic:partition:offset, and the raw value travels with it (capped at 8 192 characters), so the message can be found on the broker and read. - Offsets still advance. A poison message is dead-lettered once and never re-read: the next
run on this topic reads
IN 0. A poll of only poison keeps polling for a real row, bounded by the chunk budget, so a continuously-poison topic cannot wedge a run — the reader stops on its budget, the run endsPartialwith the dead-lettered offsets persisted, and the next run resumes past them. - Inspect entries with
lm dlq—list,get, thendiscard, orreplayedto record that you re-sent the message yourself (the hub never re-reads it); watch the queue in the TUI under:dlq.
Tombstones — a deletion is not garbage
On a compacted topic, a record with a null value is how a producer says "this key is
deleted"; log compaction keeps it as the deletion marker. acme.partners-compacted holds 1 000
partners, then 300 tombstones:
# 45-compacted.yaml — a compacted topic carrying tombstones.
#
# 1 000 partner records, then the producer deletes the first 300 keys — on a compacted topic
# a deletion is a record with a NULL value, the tombstone log compaction keeps as the marker.
# The hub's ingestion is append-only: a tombstone is dead-lettered under a reason naming the
# key, never applied as a DELETE on the table.
#
# Expect: Completed, in=1300 out=1000 skipped=300, 300 dead-letters each naming its key,
# 1 000 rows in the table.
apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
name: acme-kafka-compacted
connector_type: kafka
scopes: [billing]
spec:
name: acme_partners_compacted
# The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
brokers: host.docker.internal:29092
topic: acme.partners-compacted
format: json
$ lm run get e6d58e37
ID CONNECTOR ENDPOINT STATUS IN OUT SKIPPED DEAD-LETTERS STARTED ENDED
e6d58e37 acme-kafka-compacted default Completed 1300 1000 300 300 2026-09-13 02:21 2026-09-13 02:21
⚠ 300 rows quarantined — see: lm dlq list --run e6d58e37-3c4f-4f3f-aef7-906d79812a2a
$ lm dlq get 42d8ea82
ID: 42d8ea82-9207-4f38-b8dc-cf06706676f3
Connector: acme-kafka-compacted (4448919e-1cef-432b-844a-dd9d78f76571)
Run: e6d58e37-3c4f-4f3f-aef7-906d79812a2a
Reason: MalformedSource
Attempts: 1
Message: acme.partners-compacted:0:1299:1299 — Kafka tombstone (key '1299'): the message value is null — the producer deleted that key, and append-only ingestion dead-letters the marker rather than applying it as a delete
Payload: {"raw":[],"file":"acme.partners-compacted:0:1299","line":1299}
lumnik ingestion is append-only — the hub mirrors what the topic carried, it never removes a
row behind your back — so a tombstone becomes one DLQ entry naming the key it deleted, which
is all a tombstone carries. It counts as a skip (1300 = 1000 + 300), it does not block, and the
1 000 rows around it landed in the same run. If those deletions matter to your table, the
deletion has to reach the hub as data — a field the producer sets — because there is no
mapping hook here to interpret one.
A topic that does not exist
# 60-absent.yaml — a topic the broker does not have.
#
# acme.absent is never created, and the consumer the hub builds is forbidden to create it
# (a read must not write to its source). The refusal names the topic and says which of two
# situations it is: a first read (check the name) or an endpoint that had read this topic
# before (the broker lost it, or the name was edited).
#
# Expect: Failed — `kafka topic 'acme.absent' does not exist on the broker`.
apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
name: acme-kafka-absent
connector_type: kafka
scopes: [billing]
spec:
name: acme_absent
# The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
brokers: host.docker.internal:29092
topic: acme.absent
format: json
$ lm run get c1008671
ID CONNECTOR ENDPOINT STATUS IN OUT SKIPPED DEAD-LETTERS STARTED ENDED
c1008671 acme-kafka-absent default Failed 0 0 0 0 2026-09-13 02:21 2026-09-13 02:21
✖ run failed: kafka topic 'acme.absent' does not exist on the broker: it reports no partition for that name, so there is nothing to read. Check spec.source.topic for a typo, or create the topic on the broker. The hub never creates one (an ingestion run must not write to its source).
Two things happened in those 36 ms. The consumer asked the broker about acme.absent with
automatic topic creation off — the client's default is on, and a broker left at Apache's own
default would have created the topic and handed back one empty partition, turning a typo into a
green run over zero rows and a write into your cluster. And the empty answer was refused by
name, with the cause and the gesture, instead of the client's own
Consumer is not subscribed to any topics or assigned any partitions.
The message says which of two situations it is. A first read is a name to check. An endpoint that has a stored position read this topic before — the topic was deleted or renamed on the broker, or the manifest's topic name was edited — and the message says so instead of sending you to hunt a typo in a manifest that worked for months.
A log that moved under the cursor
Retention, compaction or an administrative trim moves a partition's log start forward. A stored
position below it names messages that no longer exist. The Kafka client repairs such a seek by
itself (auto.offset.reset=earliest) and says nothing — which is how a run that lost 500
messages would look exactly like one that read everything. Replay it on the demo broker:
# 70-trimmed.yaml — the broker deletes what the cursor still points at.
#
# Retention, compaction or an administrative trim moves a partition's log start forward. A
# stored position below it names messages that no longer exist. Replay it in three gestures:
#
# lm connector run acme-kafka-trimmed # run 1: 5 000, cursor at 5 000
# python3 …/acme-topics.py feed --topic acme.partners-trimmed --rate 500 --seconds 2 # ~1 000 more
# python3 …/acme-topics.py trim 5500 # offsets 0..5499 are gone
# lm connector run acme-kafka-trimmed # run 2
#
# Expect: run 2 ends `Partial`, its error names offsets 5000..5499 as unreadable, and the
# hub log carries the same sentence at WARN.
apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
name: acme-kafka-trimmed
connector_type: kafka
scopes: [billing]
spec:
name: acme_partners_trimmed
# The hub runs in Docker: `localhost` there is the hub's own container, not your machine.
brokers: host.docker.internal:29092
topic: acme.partners-trimmed
format: json
$ lm connector run acme-kafka-trimmed # run 1
$ python3 docs/connectors/kafka/demo/acme-topics.py feed --topic acme.partners-trimmed --rate 500 --seconds 2
▪ fed 607 messages at 500.0/s into acme.partners-trimmed
$ python3 docs/connectors/kafka/demo/acme-topics.py trim 5500
✓ acme.partners-trimmed now starts at offset 5500 (high watermark 5607): offsets 0..5499 are gone
$ lm connector run acme-kafka-trimmed # run 2
$ lm run list --connector acme-kafka-trimmed
ID CONNECTOR ENDPOINT STATUS IN OUT SKIPPED DEAD-LETTERS STARTED ENDED
49b90bd6 acme-kafka-trimmed default Partial 107 107 0 0 2026-09-13 02:22 2026-09-13 02:22
c30cf76f acme-kafka-trimmed default Completed 5000 5000 0 0 2026-09-13 02:22 2026-09-13 02:22
$ lm run get 49b90bd6
• run partial: acme.partners-trimmed[0]: the stored position is offset 5000 but the broker's log now starts at 5500 — offsets 5000..5499 are no longer on the broker (retention, compaction or a trim removed them) and can never be read; resuming at 5500
Run 1 read 5 000 and stored position 5 000. The producer wrote 607 more; the broker deleted
everything below 5 500. Run 2 asked the broker where the log starts before seeking, saw its
position overtaken, read the 107 messages that remained — and reported itself Partial, with
the exact lost range in its message and the same sentence at WARN in the hub log. Nothing
recovers those 500 messages; no connector could. What changed is that a run that read less than
the topic ever held no longer looks like a complete one.
A first run against a topic whose start has already moved is different: nothing was lost here,
and the run completes with an INFO line saying which offsets left the broker before this
connector ever read it. lm endpoint reset-cursor deletes the stored position, so a reset
endpoint is honestly a first read again.
The gesture, when it repeats: raise the topic's retention, or shorten the interval between runs.
An empty poll, and what it does not prove
A run ends when a poll returns nothing — that is how "caught up" is detected. A poll that
timed out before its first fetch arrived looks identical, and a fresh consumer per chunk
means the first poll of a run pays metadata, connection and fetch out of poll_timeout_ms. On a
slow link at the 100 ms floor, a run can read nothing from a topic holding thousands of messages
and end.
Because the run knows its target, that silence is named. A poll that returns nothing while the
position is still below the head ends the run on a WARN stating both numbers:
WARN acme.orders: the poll returned no record within 100 ms and the position is below the high watermark on partition(s) [0: at 0, high watermark 3880, …] — the poll may have timed out before the first fetch arrived, or those offsets may hold no readable record (compaction, transaction markers). The run ends here either way. Raise poll_timeout_ms if this repeats on a topic known to have data
It names two causes and chooses neither, because it cannot: compaction gaps and transaction
markers occupy offsets no consumer is ever handed, so a run that read everything can end below
the watermark, correctly. The run stays Completed — this is the one shortfall the hub cannot
prove — and the message goes to the hub's log only. When a run ingested less than you expected
and the ledger is green, read the log for the topic's name.
An honest empty topic looks like this, in 46 ms and with no such line:
$ lm run list --connector acme-kafka-empty
ID CONNECTOR ENDPOINT STATUS IN OUT SKIPPED DEAD-LETTERS STARTED ENDED
2b7a9ee6 acme-kafka-empty default Completed 0 0 0 0 2026-09-13 02:21 2026-09-13 02:21
A broker that does not answer
A bootstrap address that answers nothing fails the run on the Kafka client's own metadata
timeout — one wait of about 60 s, then a Failed row saying the broker did not answer,
naming the address and the topic. The run measures its target first; when that is what times
out, the reader is not given the same 60 s to spend over again, because a broker that did not
answer a metadata request will not answer a fetch either. An address that refuses the connection
fails the same way. Neither is reported as a successful empty sync, and the failure is marked
retryable — a broker that comes back is what the next scheduled slot finds.
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 |
Every *_env key on this connector — brokers_env, username_env, password_env — is a
name. The hub resolves it when the run starts: the lm secret
registry for your tenant first, then its own environment (an OS variable, then a JVM system
property). The value is never in the manifest and never in the stored config.
# 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 it before the name resolves, and 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
The same resolution serves the credentials. sasl-plain refuses a name that resolves to nothing
— or to blanks — before opening a connection, so the failure is about the secret and not
about SASL:
# 65-sasl.yaml — SASL/PLAIN credentials, by name.
#
# Neither the username nor the password is in this file: `username_env` and `password_env`
# name registry entries (`lm secret set ACME_KAFKA_USER`, `lm secret set ACME_KAFKA_PW`).
# `security_protocol` picks the wire: SASL_SSL is the default, SASL_PLAINTEXT is for a
# broker that authenticates without TLS. Both land verbatim in the consumer's config.
#
# The demo broker authenticates nobody, so this manifest cannot complete against it — it is
# here for the two refusals worth seeing: the one an unresolved name gets before any
# connection, and the one a broker that speaks no SASL gives 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 052f113a
✖ 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
The demo broker authenticates nobody, so once both names resolve the broker refuses, in its own words — the message to expect from a listener that does not speak the mechanism you declared:
$ lm run get 9e962b07
✖ run failed: Unexpected handshake request with client mechanism PLAIN, enabled mechanisms are []
security_protocollands verbatim in the consumer'ssecurity.protocol; it is not validated at apply.SASL_SSLis the default; a SASL listener without TLS wantsSASL_PLAINTEXT.- The username and password are embedded in the JAAS login string with proper escaping — a secret containing a backslash or a double quote cannot break the parser or smuggle an extra option.
- SASL/SCRAM, mTLS and OAuth are not yet supported.
Azure Event Hubs — the event-hubs preset
Event Hubs exposes a Kafka-compatible endpoint that authenticates as SASL/PLAIN with the fixed
username $ConnectionString (the literal string, dollar sign included — it is not a secret) and
the namespace connection string as the password. The preset writes that config for you:
connector_type stays kafka, and spec.preset rides as an internal field the apply handler
expands and removes.
# 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
{"name":"event-hubs-telemetry","connectorId":"6ba84373-8816-4ef2-98dd-6b98477c924e","endpointsApplied":1}
$ lm connector get event-hubs-telemetry
name: event-hubs-telemetry
typeId: kafka
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
The stored config is a plain kafka one: preset, namespace and connection_string_env did
their work at apply time and do not survive. The two names it reads resolve like every other
*_env key — registry first, then the hub's environment:
lm secret set EVENTHUBS_CONN_STR # the namespace Primary Connection String, Endpoint=sb://…
lm secret set EVENT_HUBS_SASL_USERNAME # the fixed literal $ConnectionString
An unknown preset name is refused at validate (unknown source preset 'event-hubbs') — a typo
never silently degrades to plain kafka — and a preset missing an input reports each missing key
by name (preset 'event-hubs' requires 'namespace') before anything is expanded. No consumer
group is created on the Event Hubs side either; there is nothing to provision or clean up there.
Limitations
- JSON only. One JSON object per message value. Avro, Protobuf and a Schema Registry are not
yet supported;
formataccepts onlyjson. - 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 flatten, no filter, no lookup. A field must reach the hub under the name and shape it should land with.
- No pre-flight.
lm source jdbc testandlm source rest testhave no kafka sibling yet; the first run is the test. It is a cheap one: a topic that does not exist is refused in tens of milliseconds, an unresolved secret before any connection. - 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-plainonly. 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— theKINDcolumn readskafka; Enter drills into the singledefaultendpoint:runs— one row per run,Partialin its own colour besideCompletedandFailed: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 from the stored cursor to the head of that moment; slots at night are skipped. A slot that meets a running run is skipped too, so the cadence can be as tight as you like. See Continuous consumption.
Read a topic again from the beginning
lm endpoint reset-cursor <connector>/default
lm connector run <connector>
Result: the stored position is deleted and the run reads everything the topic retains. Rows
already in the table are refused by _row_hash and counted in OUT anyway — the table's row
count is the witness. The command prints nothing and exits 0.
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 has its own
cursor and its own table; the broker sees two independent readers.
Result: two connectors over one topic, useful when two métiers 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 |
run completes with IN 0 on a topic that has data, and the hub log says the poll returned no record within N ms |
the first poll timed out before its first fetch — or those offsets hold no readable record | raise poll_timeout_ms; if it repeats on a compacted or transactional topic that is genuinely caught up, it is the watermark, not the run |
a run ends Partial |
it stopped below the head it found at start: the broker deleted offsets the stored position was entitled to, or the topic fed only unparsable messages up to the chunk budget | lm run get names which; for the first, nothing can be recovered — check retention.ms against your cadence |
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: source column 'id' collides with a hub system column |
a top-level field shares one of the seven reserved names | 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 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; reset the cursor to re-read if the early rows carry it on the broker now |
lm validate warns unknown key on target or mapping |
copied from a jdbc or csv manifest; kafka has neither | remove the block |
lm validate refuses the manifest: the connector type kafka is unknown to the hub |
the hub is the open edition | Kafka is a Pro connector |
See also
- The demo broker — the source every screen here reads
- Overview — apply/validate flow shared by all connectors
- Transformers — the hooks other connectors run at ingestion, and that kafka does not
- Scopes — how
metadata.scopesmakes ingested rows discoverable in chat - Processes — react —
data.row.changedon the rows a run writes - Observability —
Partialas the status to alert on - Diagnosing a failed run