Skip to content

Kafka connector (kind: kafka)

Consume events from a Kafka-compatible broker (Kafka, Redpanda, Azure Event Hubs) into the same dynamic tables and RAG chunks as any other source. The hub owns the position: the connector uses assign() + seek(cursor) — no Kafka consumer group — and stores the partition→offset cursor exactly like a JDBC watermark.

How it behaves

  • A run catches up, chunk by chunk. The engine loops chunks within one run — each chunk is one consumer poll batch, written in its own transaction through the standard mapping pipeline to connector.t_<name> — until an empty poll says the topic is caught up, which completes the run.
  • First run = full-retention ingest: a partition with no stored cursor seeks to the beginning, so run #1 reads everything the topic still retains. Subsequent runs resume at the stored partition→offset cursor.
  • Non-JSON messages go to the per-message DLQ (MalformedSource) and never block ingestion — the same discipline as the CSV per-row DLQ. Their offsets still advance: a poison message is dead-lettered once and never re-read. A poll of only poison keeps polling for real rows, bounded by the chunk budget, so even a continuously-poison topic cannot wedge a run — the advanced cursor is persisted and the next run resumes past the poison.
  • Dedup by content hash (_row_hash): re-ingesting the same message is a silent conflict; data.row.changed fires only for genuinely-new rows.

Full example — plain Kafka, JSON, no auth

# 10-kafka-json.yaml — plaintext Kafka source: one topic, JSON messages, no auth.
#
# Apply and run:
#   lm apply -f docs/connectors/kafka/10-kafka-json.yaml
#   lm connector run kafka-erp-exports
#
# How the Kafka connector works
# -----------------------------
# The connector uses assign() + seek(cursor) — NO Kafka consumer group.
# The partition→offset position belongs to the hub (stored in connector_cursor),
# exactly like a JDBC watermark.  A run catches up chunk by chunk (each chunk =
# one poll batch in its own transaction) until an empty poll says caught-up;
# every message passes through the mapping pipeline to DynamicTableWriter.
#
# Non-JSON messages are isolated to the DLQ (MalformedSource) and do not stop
# the run — the same discipline as the CSV per-row DLQ (PR #9).
#
# Dedup: the write path keys rows by a UNIQUE content hash (_row_hash).
# Re-ingesting the same message produces a conflict that is silently ignored;
# data.row.changed fires only for genuinely-new rows (É2.b semantics).
#
# Auth: none — no SASL credentials are sent; suitable for a local Redpanda or
# an internal broker that trusts the network.

apiVersion: connectors.lumnik.io/v1
kind: Connector
metadata:
  name: kafka-erp-exports            # unique connector name within the workspace
  connector_type: kafka              # selects the Kafka connector implementation
  tags: [demo]

spec:
  name: erp_events                   # logical/table name → target table connector.t_erp_events

  brokers: localhost:9092            # Kafka / Redpanda bootstrap address (literal)
  # brokers_env: KAFKA_BROKERS       # alternative: resolve from env var

  topic: erp-exports                 # source topic; must be identifier-safe (no spaces)

  format: json                       # v1: one JSON object per message (Avro/Protobuf deferred)

  # No auth block → unauthenticated connection (no SASL handshake).

Field reference

Field Required Meaning
name yes logical/table name → connector.t_<name>
brokers or brokers_env one of the two bootstrap servers, literal or via env var
topic yes must match [a-zA-Z0-9._-]+
format no v1 supports only json (Avro/Protobuf deferred) — one JSON object per message
poll_timeout_ms no consumer poll timeout, range [100, 60000], default 2000
max_poll_records no max records per consumer poll; positive integer, default 500; clamped to the engine chunk size (500)
auth no kind: none (default) or kind: sasl-plain with username_env + password_env (+ optional security_protocol)

With sasl-plain, auth.security_protocol overrides the wire protocol — default SASL_SSL; set SASL_PLAINTEXT for a SASL broker without TLS. The value lands verbatim in the consumer's security.protocol (it is not validated at apply time).

Continuous consumption

A run is a catch-up, not a subscription: it drains the backlog to the head and completes. To keep the table current, declare a schedule: cron in the manifest — like jdbc/csv/rest, it lands on the scheduler (absent cron on re-apply = unschedule). Each run resumes at the stored cursor and catches up again. Overlap is safe: a submission while a run is already active on the endpoint is skipped (per-endpoint advisory lock), so an aggressive cadence degrades to back-to-back catch-ups, never double ingestion. An external trigger works too: cron + lm connector run <name>.

Azure Event Hubs — the event-hubs preset

Vendor presets ride as spec.preset while connector_type stays kafka (the engine never learns the vendor name — validator == runtime). The preset pre-wires the bootstrap address (<namespace>.servicebus.windows.net:9093) and SASL/SSL from the connection string.

Prerequisite — two env vars where the hub runs. Event Hubs' Kafka endpoint 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 reads the username from the fixed env var EVENT_HUBS_SASL_USERNAME and the password from the env var you name in connection_string_env; both resolve from the process environment only (OS env, then JVM system property) — the lm secret store is not consulted:

export EVENTHUBS_CONN_STR='Endpoint=sb://…'           # the namespace Primary Connection String
export EVENT_HUBS_SASL_USERNAME='$ConnectionString'   # the fixed literal username

# 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 real engine type); the vendor alias rides as
# the internal 'spec.preset: event-hubs'.  At apply time SourcePresetRegistry
# expands it into a real 'kind: kafka' config, pre-wiring the bootstrap address
# (namespace.servicebus.windows.net:9093) and SASL/SSL settings from the
# connection string.  The engine never learns the vendor name — validator==runtime
# holds.  This mirrors the pagination-preset pattern from PR #11.
#
# The 'connection_string_env' field names an env var that holds the Event Hubs
# connection string (Primary Connection String from the Azure portal). Kafka resolves
# *_env via the process environment only (OS env, then JVM system property) — the DB
# secret store is NOT consulted here. Export it where the hub runs:
#   export EVENTHUBS_CONN_STR="Endpoint=sb://..."
#
# Auth: Event Hubs over Kafka uses SASL/PLAIN with fixed username "$ConnectionString".
# Set it once in the environment (it is not a secret):
#   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
  tags: [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

An unknown preset name is rejected at validate time (a typo never silently degrades to plain kafka), and a preset missing its inputs (e.g. namespace) reports each missing field before expansion.

See also