The first two entries are basic configuration about the type of the connector and its name. slot.name is the aforementioned replication slot from which the Debezium connector subscribes to CDC events. This is a name that you have to configure yourself. The Debezium connector creates the replication slot the first time it is started. You should see something in your logs like:

"e63c" data-selectable-paragraph="">12:58:25,223 INFO Postgres|debezium-nl-fca-dev__ts6|postgres-connector-task Creating replication slot with command CREATE_REPLICATION_SLOT "debezium_ts" LOGICAL decoderbufs [io.debezium.connector.postgresql.connection.PostgresReplicationConnection]

table.include.list is a regex defining which tables in Postgres should be captured. "plugin.name": "decoderbufs" is the one of two currently available plugin variations, decoding the records into protobuf format. Note that this is just the data format in which the output plug-in transforms the data from Postgres’ write-ahead log’s internal representation into a generally understandable format. Our Debezium connector will not use this as an output format.
The transforms section describes a set of single message transformations applied to each record for the purpose of topic routing. The ByLogicalTableRouter router changes events to topics specified by transforms.Reroute.topic.replacement and transforms.Reroute.topic.regex. The former indicates that all records are written to a topic named debezium_ts_topic. The latter describes a regex to include matching tables names in Postgres. So we match all tables that include the name tr_commandercommunication. We use the logical table router here because we want to route multiple physical tables (in different Postgres shards) into the same topic.
A quite important property is snapshot.mode. When Debezium is connecting to Postgres for the first time, it tries to stream a consistent snapshot of the configured tables. After the snapshot is completed, it starts streaming changes from that point onwards. With the setting initial_only it does this only for the first time it connects. Other options for this setting are always, never, and custom. always is not desirable in our setting because in case of a restart of the connector, it would re-stream the full state of Postgres, which in our case can be quite substantial. An important learning here is that the snapshot is triggered based on the connector.name property. Hence, if you want to re-trigger a snapshot with "snapshot.mode": "initial_only" (for example, when you need to consume events from a new slot), then you need to rename the connector.

Lastly, we configure the connectors to use Avro as a data serialization format. Since we are already using Confluent Cloud, we can use the Confluent Schema registry here. The configuration properties of value.converter configure that the messages should be encoded in Avro whilst using the schema registry indicated at value.converter.schema.registry.url. This worked quite well for us, as the schema is automatically communicated to the registry and we did not have to configure anything else here.

Kafka & Confluent Cloud

After having read the CDC events from our Postgres database, the Debezium connector is pushing them into their respective topics in Kafka. From here, the journey to our data warehouse is pretty much the same as in our internal services pipeline. We use the Snowflake and S3 connectors to further process and archive the data. Once the Debezium events are in the data warehouse, the data team and analysts can process the events.

Operations

Monitoring

We use Prometheus for monitoring the connectors. Next to monitoring JMX and generic Kafka connect metrics, as described in earlier blog posts, there are also a number of Debezium specific metrics which are interesting. Debezium differentiates between two types of metrics, the snapshotting and the streaming ones. Most metrics are available for either situation. For our setup, we only monitor streaming ones, because we are not expecting frequent snapshotting.
One particular metric that we found quite useful in practice is MilliSecondsSinceLastEvent. When Debezium does not receive any events for a long period, this metric’s value will be -1, so we monitor for this value with the following PromQL expression:

"cb9e" data-selectable-paragraph="">avg_over_time(debezium_metrics_millisecondssincelastevent{context="streaming", plugin="postgres"}[60m])) < 0.

Learnings and pitfalls

During the development of this pipeline we noticed that a good collaboration with the database team and the business is important. It took a number of iterations until we found the right combination of settings that fit our use case. For example, the snapshot.mode setting is key to make sure you achieve the delivery guarantees needed by the business in case of a database failure.
The development process can be a bit tedious, because you are merely writing JSON configuration files, so you do not have the feedback that you would have with a programming language. In addition, the error messages Debezium produces can also be misleading and tend to be not very informative.
Another point of attention is that the Debezium connector creates a complex, nested schema for the events, which might not be very well suited for downstream consumption. Many sink systems expect a flat structure in which each entry maps to e.g. a database column. If you find yourself in this situation, you can tackle this with a number of single message transformations offered by Kafka Connect. In particular, the ExtractNewRecordState transformation can be useful to flatten the events into a more digestible format.

Wrapping up

In this blog post, we looked at the FCA data pipeline by the Analytics Platform team. We applied change data capture, Kafka and Snowflake to capture analytics events. Debezium is an exciting CDC implementation which enables you to stream database changes to Kafka. At Picnic we use it to work around the inaccessibility of analytical events of third-party components. See you next time!