One of the biggest pain points in Debezium since its inception was the sup-optimal support for changes to the captured tables list. As a user, you create a new connector with a list of tables to be captured (table.include.list and related options); at a later point in time, it may become necessary to adjust this configuration, so to capture further tables which where not part to CDC initially. If it suffices to only stream changes from these tables, then the problem is pretty simple to solve. But what if you also need to capture the existing contents of the tables?

Capturing existing data in tables is traditionally done by Debezium in the snapshot phase. This phase is executed once upon the first connector start-up, and its objective is capturing consistent data at a point of time (transforming data at rest into data in motion). This can be a fairly long operation, and by definition, it must be executed completely or not at all - a bit like transaction semantics. This means that if the snapshot is not completed due to a connector restart for instance, it must be re-executed from scratch, and everything already done is thrown away. Also, while the snapshot is taken, any data modifications that are executed in parallel in the database are not streamed until the snapshot has been completed. This could lead to problems with database resources for very large snapshots, as transaction logs must be kept available until the streaming is started.

We are thus ended up with three issues to be solved:

The problem was well known, and over time we developed workarounds and also ideated possible improvements and new solutions. As a workaround, the general recommendation was to use a multiple connector approach. The user was asked to:

This method somewhat did the trick, but is very clumsy, and all the questions around snapshot consistency mentioned above still apply.

The next step was a community contribution into the Debezium connector for MySQL via DBZ-175. It was based on the notion of having multiple binary log readers in place. One reader would capture the originally configured tables, while the other one will snapshot the new tables and then capture changes from the new tables. The latter reader would catch up with the original one, and then they would be reconciled and merged into a single one.

The code was working well, but it never left the incubating stage, as the process itself was quite complex and liable to errors in corner cases. Last but not least, it was an ingenious approach, but unfortunately not portable to other connectors.

In late 2019, the Netflix engineering team announced that they had developed an in-house change data capture framework. They also came up with an innovative solution of executing concurrent snapshots using watermarking, described in the paper DBLog: A Watermark Based Change-Data-Capture Framework by Andreas Andreakis and Ioannis Papapanagiotou.

The main idea behind this approach is that change data streaming is executed continuously together with snapshotting. The framework inserts low and high watermarks into the transaction log (by writing to the source database) and between those two points, a part of the snapshotted table is read. The framework keeps a record of database changes in between the watermarks and reconciles them with the snapshotted values, if the same records are snapshotted and modified during the window.

This means that the data is snapshotted in chunks - no lengthy process at the connector start, and also in case of crashes or a controlled termination of the connector, the snapshotting can be resumed since the last completed chunk.

As per Netflix, the implementation is provided for MySQL and PostgreSQL databases.

Before moving to Debezium’s implementation of the watermark-based snapshotting approach, a small detour is needed.

Sometimes it can be useful to control Debezium from the outside, so to force it to execute some requested action. Let’s suppose it is necessary to re-snapshot an already snapshotted table - a so-called ad-hoc snapshot. The user would need to send a command to Debezium to pause the current operation and do the snapshot. For that purpose, Debezium defines the concept signals, issued via a signalling table. This is a special table, designated for communication between the user and Debezium. Debezium captures the table and when the user requires a certain operation to be executed, they simply write a record to the signalling table (sending a signal). Debezium will receive the captured change and then execute the required action.

When we became aware of DBLog’s snapshotting approach, we decided that the method is a universal one and that we could try to adopt it in Debezium, too. Also as we share a lot of codebase among the different connectors (using the Debezium connector framework) our objective was to implement it in the Debezium core component, so that all connectors would benefit from the feature at once. The design and implementation were driven by the DDD-3 Debezium design document.

Incremental snapshotting in Debezium is available in form of ad-hoc snapshots. The user does not configure the connector to execute the snapshot, but instead they use the signalling mechanism to send a snapshot signal and thus trigger a snapshot of a set of tables. The signal in question is called execute-snapshot and the signal message follows the format of:

When a table snapshot is requested, then Debezium will do the following:

When a chunk is queried, a dynamic SQL statement is built, selecting the next incremental.snapshot.chunk.size records, whose primary keys are larger than the last one from the previous chunk (or the first primary key for the first chunk) and which are smaller or equal to the recorded maximum primary key.

The reading of a chunk is a slightly complicated procedure:

Why is this needed? Why it is not enough to just query the database? The answers lie in the following picture:

Debezium is not the only process accessing the database. We can expect a multitude of processes accessing the database concurrently, potentially accessing the same records which currently are snapshotted. As shown in the picture, any changes to data are written to the transaction log based on the commit order. As it is not possible to precisely time the chunk read transaction to identify potential conflicts, the open and close window events are added to demarcate the time in which the conflicts can happen. Debezium’s task is the deduplication of those conflicts.

For that purpose, Debezium records all events generated by the chunk into a buffer. When the snapshot-window-open signal is received, then all events coming from the transaction log are checked whether they belong to the snapshotted table(s). If yes, then the buffer is checked whether it contains the primary key. If yes, then the snapshot event is dropped from the buffer, as this is a potential conflict. And as it is not possible to correctly order the snapshot and transaction log events, only the transaction log event is kept. When the snapshot-window-close signal is received, the remaining snapshot events in the buffer are sent downstream.

The following image shows an example of how such a buffer works and how are the transaction log events are filtered before being sent downstream:

Records K2, K3, and K4 exist already in the database. Before the snapshot window opens, records K1 gets inserted, K2 updated, and K3 deleted. These events are sent downstream as they are read from the log. The snapshot windows opens, and its query selects K1, K2, and K4 into the buffer. While the window is open, the deletion of K4 is retrieved from the transaction log; the snapshot event for K4 is dropped from the buffer and the deletion event is sent downstream. K5 and K6 are inserted, which is retrieved from the log, corresponding events will be emitted. Depending on the specific timing, there may be read events for them in the buffer too (in the image that’s the case for K5), which would be dropped. When the snapshot window closes, the remaining snapshot events for K1 and K2 will be emitted from the buffer.

Having discussed the general concepts, let’s explore things a bit more in an example. We will use our standard tutorial deployment to demonstrate ad-hoc incremental snapshotting. We are using PostgreSQL as the source database. For this demo, you will need multiple terminal windows.

In the beginning we will start the deployment, create the signalling table, and start the connector:

From the log we see that as per the table.include.list setting only one table is snapshotted, customers:

In the next step we will simulate continuous activity in the database:

The topic dbserver1.inventory.customers receives a continuous stream of messages. Now the connector will be reconfigured to also capture the orders table:

As expected, there are no messages for the orders table:

Now let’s start an incremental ad-hoc snapshot by sending a signal. The snapshot messages for the orders table are delivered to the dbserver1.inventory.orders topic. Messages for the customers table are delivered without interruption.

If you were to modify any record in the orders table while the snapshot is running, this would be either emitted as a read event or as an update event, depending on the exact timing and sequence of things.

As the last step, let’s terminate the deployed systems and close all terminals:

In this blog post, we have discussed the motivation for the notion of incremental snapshotting, as introduced by the DBLog paper. We have reviewed the methods used in the past to achieve the described functionality. Then we dived into the deep waters of the implementation of this novel snapshotting approach in Debezium, and in the end we tried to use it live.

We hope you will find incremental snapshotting useful and we look forward to your feedback, experiences, and use cases. In a future blog post, we’ll talk about the support for incremental snaphots of read-only databases (supported by the Debezium MySQL connector as of version 1.7) and how to trigger ad-hoc snapshots using a Kafka topic as the means of signalling instead of a database table.