Kafka (Debezium)

The Kafka source consumes Debezium change events from a topic. Anything Debezium captures (Oracle, Db2, Cassandra, Informix, Vitess, SQL Server, PostgreSQL, MySQL…) can feed Meilisearch this way.

Prerequisites

A Debezium connector writing one table's changes to one topic, typically <topic.prefix>.<schema>.<table>.
The JSON converter for keys and values, with or without schemas. Avro and Protobuf are not supported.
Brokers reachable from the service in plaintext: the connector takes a list of host:port brokers and has no SASL or TLS settings.

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "topic.prefix": "dbserver1",
    "table.include.list": "inventory.customers",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "true",
    "value.converter.schemas.enable": "true"
  }
}

(Database connection settings of the Debezium connector are omitted; see the Debezium documentation for your database.)

Connection settings

Field	Meaning
`brokers`	One or more `host:port` seed brokers, for example `kafka-1:9092, kafka-2:9092`.

Task fields

Field	Default	Meaning
`table`	required	The Debezium topic, for example `dbserver1.inventory.customers`. A missing topic fails with a hint to check `topic.prefix` and the table include list.
`groupId`	`cdc-<task id>`	The consumer group whose committed offsets track progress. Never share it with other consumers.

How changes are applied

Envelope. Messages may be wrapped as {"schema": …, "payload": …} or be the bare envelope. Debezium's op is mapped: c, u and r (snapshot read) upsert the after image; d deletes using the before image, or the message key when there is none; t is a truncate. Tombstones (empty values after a delete), heartbeats, schema changes and transaction metadata are skipped.
Types. With schemas enabled, Debezium logical types are converted: io.debezium.time.Timestamp, MicroTimestamp, NanoTimestamp, ZonedTimestamp, Date and Kafka Connect Timestamp/Date become times; org.apache.kafka.connect.data.Decimal becomes a number.
Schema and key. Columns are inferred from up to 50 messages read in at most 5 seconds from the start of the topic. The primary key comes from the message key's fields (id when it cannot be determined); set mapping.primaryKey to override it.
Snapshot. Replays the topic from its start up to the current end offsets and keeps the latest image per message key in memory, then writes the rows still alive. Memory grows with the number of live rows.

Very large topics

Use the stream sync mode instead of snapshot_stream: the topic is consumed from the group's committed offsets, or from the earliest offset when the group has none, as ordinary change events, without holding the table in memory. Debezium's own snapshot (op: r) then fills the index.

Positions and resume

A position is a JSON object mapping topic/partition to the next offset to read. Confirmed positions are also committed to the consumer group, so tools like kafka-consumer-groups.sh --describe --group cdc-<task> show the task's lag. If an offset is no longer available (topic retention removed it), the position is lost and the task resyncs automatically.

Troubleshooting

Error	Fix
`kafka: topic "…" does not exist; check the Debezium connector topic.prefix and table include list`	Use the exact topic name Debezium writes.
`kafka: delete without before image or key`	Keep the message key (do not drop it in an SMT), or enable before images on the source.
Decoding errors on every message	The topic uses Avro or Protobuf; switch the connector to the JSON converter.