Change tracking

A change tracking publishes a database’s changes as events. It runs Debezium inside the integration, which follows the database’s own change log and publishes each insert, update and delete into a channel, where any integration can react to it. You write one document and no route. Every Debezium connector works, including PostgreSQL, MySQL, SQL Server, Oracle, MongoDB and Db2.

Write the change tracking#

apiVersion: weik.io/v1
kind: change-tracking
metadata: { name: order-changes, project: shop }
spec:
  type: postgres
  databaseHostname: "{{orders-db-host}}"
  databasePort: 5432
  databaseDbname: orders
  databaseUser: weikio_cdc
  databasePassword: "{{orders-db-password}}"
  offsetTable: weikio_offsets
  channel: order-changed
  parameters:
    tableIncludeList: public.orders
    additionalProperties.offset.storage.jdbc.connection.url: "jdbc:postgresql://{{orders-db-host}}:5432/orders"
    additionalProperties.offset.storage.jdbc.connection.user: weikio_cdc
    additionalProperties.offset.storage.jdbc.connection.password: "{{orders-db-password}}"

The connector keeps its position in a table of a database you name, normally the tracked database itself, so the three additionalProperties.offset.storage.jdbc.connection parameters are required. A connector with a schema history, sqlserver, mysql, oracle and db2, keeps it beside the offsets in the same database. Both tables belong outside what the connector tracks.

weikio init order-changes --kind change-tracking --project shop writes a change tracking that publishes into order-changes-changes, says so, and prints the weikio init line that creates that channel.

Prepare the database#

Debezium follows the database’s change log, so set the database up for it before the first start. For PostgreSQL, with a login that isn’t a superuser:

• The server runs with wal_level=logical.
• The login has LOGIN REPLICATION, SELECT on each tracked table and CREATE on the schema of the offset table.
• A publication of the tracked tables exists with the name dbz_publication, Debezium’s default.
• pg_hba.conf has a host replication line for the login, because the agent connects from another host.
• Each tracked table has REPLICA IDENTITY FULL, so an update’s before has the old row and a delete’s before the whole row.

SQL Server needs change data capture turned on for the database and each table, with SQL Server Agent running, and MySQL needs a binary log in ROW format. Debezium’s documentation of each connector lists the details. MySQL’s JDBC driver is GPL-licensed and Camel doesn’t bundle it, so a MySQL change tracking lists it in dependencies, such as mvn:com.mysql:mysql-connector-j:9.6.0.

Deploy it#

$ weikio apply -f ./shop/events/
$ weikio deploy shop/order-changes --env production

A change tracking runs one copy at a time, so two agents never run two connectors against one offset table. Its leader stands down while its agent can’t reach the hub. It uses the Debezium version of its Camel line: 3.6.1.Final on Camel 4.22.1 and 3.4.1.Final on 4.18.5.

What lands in the channel#

Each change is one event with Debezium’s envelope as its body, op, ts_ms, before, after and source:

{
  "op": "u",
  "ts_ms": 1760000000000,
  "before": { "id": 1042, "status": "open" },
  "after": { "id": 1042, "status": "paid" },
  "source": { "connector": "postgresql", "table": "orders", "lsn": 24023128 }
}

The event’s ID is made from the change’s position in the database, its table and its key, so a change that Debezium sends again soon after a restart is stored once. A replay after a longer outage is stored again, so readers of a change tracking’s channel should tolerate a duplicate, for example with #weikioIdempotent. The connector stores its position about once a minute, so a restart can replay up to that minute of changes.

Starting over#

The connector keeps its position under the change tracking’s name. A database restored from its own backup gives its positions out again, so after such a restore start the tracking afresh under a new name, which starts with a fresh snapshot. A rename starts from a fresh snapshot for the same reason.