Backfill and Re-Snapshot Safely

The question

You need to put history into a sink that is already receiving live changes. How do you reload it without overwriting newer rows, bringing back deleted ones, or leaving rows behind that no longer exist at the source?

A backfill is not a special kind of data. It is more at-least-once delivery, with one difference: it deliberately sends old content, and it sends it at the same time as the live stream. CDC delivery is at-least-once, and end-to-end exactly-once across independent systems is not achievable in general. What you can reach is exactly-once processing (effectively-once): at-least-once transport into an idempotent sink keyed on the primary key and ordered by the source log position (LSN, SCN or binlog coordinate), never by a timestamp, with deletes kept as delete markers. A backfill is safe exactly when the sink already has those three properties. Duplicates during the overlap are expected and harmless.

This page is the task view. How an initial snapshot hands off to the stream, and how incremental snapshots use watermarks, are explained on Snapshotting (watermarks, idempotent sink patterns) and are not repeated here. The guard itself is derived on Making Sinks Bulletproof, and the delete marker on Deletes That Stay Deleted.

The Debezium, Kafka and Kafka Connect behaviour below was checked by reading the documentation on 2026-10-09 (links at the end; the Debezium pages are the 3.7 documentation). Nothing was run against a live Debezium or Kafka. The SQL was run on SQLite in this site’s unit tests. Where a behaviour is not stated in the pages read, the text says “check your version” instead of guessing.

When and why you backfill

A sixth reason is that the Kafka topics were deleted or lost retention. Debezium lists “Kafka topics are deleted and must be rebuilt” as a reason for an ad hoc snapshot. If the only thing wrong is the sink and the topic still holds the history, a replay from Kafka is cheaper than touching the source.

The five ways to reload

In every case the sink sees rows it may already have. That is the design, not a fault.

Choice 1: snapshot.mode

The property decides whether a connector snapshots when it starts. The values below are the ones the Debezium 3.7 PostgreSQL and MySQL pages list. Names have changed across releases: the MySQL page marks schema_only as deprecated in favour of no_data and schema_only_recovery as deprecated in favour of recovery. Check the page for your version and your connector before you set one.

What position do the snapshot rows carry? The MySQL page says all read events of an initial snapshot contain the same binlog position, the one read before the scan. The PostgreSQL page says streaming continues from the position read at the start of the snapshot, and its example event shows source.lsn and a sequence field (the last committed LSN, then the current LSN), but that example is a create event and the page does not say which source.lsn a read event carries: look at one in your own topic. Rows of an incremental snapshot have op = 'r' and source.snapshot = "incremental" in the Debezium example, which is how a sink can tell them apart for monitoring (not for ordering).

Choices 2 and 3: the execute-snapshot signal

Both are requested by writing a signal to the source database’s signal table (the source channel) or to a Kafka signalling topic. The signal table is named by signal.data.collection, and signal.enabled.channels defaults to source. The table has the columns id, type and data. The statement below follows the Debezium PostgreSQL page; it is not run here.

-- From the Debezium documentation pattern. Not run in this site's tests.
INSERT INTO myschema.debezium_signal (id, type, data)
VALUES ('backfill-2026-10-09-1',
        'execute-snapshot',
        '{"data-collections": ["public.customers"],
          "type": "incremental",
          "additional-conditions": [
            {"data-collection": "public.customers", "filter": "id >= 100000 AND id < 200000"}
          ]}');

• data-collections has no default. It is an array of table names or regular expressions, matched case-sensitively against schema.table. An empty array means no action.
• type is incremental (the default when omitted) or blocking.
• additional-conditions appends a WHERE filter per table. This is how you backfill a key range or one tenant rather than the whole table. The sweep below must repeat the same filter. For a blocking snapshot, the filter is a full SELECT, the same form as snapshot.select.statement.overrides.
• Chunk size. incremental.snapshot.chunk.size defaults to 1024 rows. A bigger chunk runs fewer queries and needs more memory to buffer.
• Watermarks. incremental.snapshot.watermarking.strategy is insert_insert (open and close entries per chunk in the signal table) or insert_delete (one entry, removed afterwards, which stops the signal table growing quickly). Setting read.only to true lets a PostgreSQL 13 or later connector with a read-only connection run incremental snapshots without writing to the signal table. The mechanism itself is on Snapshotting.
• Control. pause-snapshot and resume-snapshot signals pause and resume an incremental snapshot; stop-snapshot stops it, optionally for chosen tables. An interrupted incremental snapshot resumes where it stopped.
• Where the rows go. When you run an ad hoc snapshot on an existing table, the connector appends the content to the topic that already exists. The backfill is therefore interleaved with live events in the same topic.

Incremental versus blocking. An incremental snapshot keeps streaming; Debezium buffers each chunk and drops a buffered read when a streamed event for the same key arrives during the chunk’s window, because the streamed event supersedes it. A blocking snapshot stops streaming, snapshots, then resumes, and the documentation warns that a delay between the signal and the actual start can make the connector emit some records that duplicate records the snapshot captured. Use blocking when you want the big table loaded faster than incremental chunks allow and can accept the pause.

Choice 4: replay from Kafka

When the topic still holds the history, you can rebuild the sink without touching the source. For a plain consumer group, Kafka’s kafka-consumer-groups.sh --reset-offsets takes one group and a topic scope, offers --to-earliest, --to-datetime, --to-offset, --shift-by and --from-file, shows what it would do unless you add --execute, and requires the group’s consumers to be inactive first. For a Kafka Connect sink connector, the REST API has offset management endpoints (KIP-875): GET, DELETE and PATCH on /connectors/{name}/offsets, which require the connector to exist and be in the stopped state. The same procedure is on Offsets & Replays and Reconciliation and Offset Surgery.

Three limits decide whether the replay is enough:

• Retention. With the default delete cleanup policy, old segments are discarded when the retention time or size limit is reached (retention.ms defaults to 7 days). --to-earliest means the earliest offset that still exists, which may not be the start of history.
• Compaction. A compacted topic keeps at least the latest value for each key, so it can rebuild current state, and Kafka’s design page describes exactly that use. It drops the intermediate versions.
• Deletes on compacted topics. A record with a key and a null value is a delete marker that is itself cleaned out after delete.retention.ms (default 1 day). Kafka documents that setting as a bound on how long a consumer reading from offset 0 has to complete the read to get a valid final state. A sink that still holds old rows and replays a compacted topic can miss a delete whose marker has already been removed, and keep a row the source no longer has. Treat a compacted replay as a reload and run the sweep afterwards.

Choice 5: reload the sink from the source by SQL

Sometimes the sink and the source are reachable from one place (a replica, a foreign table, a batch job) and a query is simpler than a signal. The cost is that no connector chooses the boundary for you, and the whole correctness argument rests on it. The rule: stamp the loaded rows with a position at or below the first change of every transaction that is still open when you start the read.

The reason is where a streamed position comes from. On PostgreSQL the source.lsn of a streamed event is the position of the change’s WAL record, not of the commit (the Debezium 3.7 PostgreSQL page says it is “the position in the write-ahead log where the transaction changes were written”; in PostgreSQL’s logical.c the change callback reports the change’s own LSN, while commit-time messages use the transaction’s end LSN or an abort LSN). A transaction that began before the stream’s start position and commits after your read therefore arrives with positions below that start. If you stamped the loaded row with the start position, the loaded row would beat that change, which would then be skipped permanently.

• Take a lower bound. For example the restart_lsn of the slot, read before the read starts (PostgreSQL documents it as the address of the oldest WAL the slot’s consumer might still need; verify that it is below the first change of every open transaction on your version).
  • Use the connector’s own slot, on the primary. Check that restart_lsn is not NULL and that wal_status is not lost; a lost slot has no usable bound.
  • A lower bound that is far behind only costs you coverage: the reload repairs rows stored below the stamp and leaves rows stored above it.
• Or wait until no older transaction is open before you read. In order (the order matters):
  1. Take P = pg_current_wal_lsn().
  2. Then take X = pg_snapshot_xmax(pg_current_snapshot()).
  3. Poll pg_snapshot_xmin(pg_current_snapshot()) until it is at or above X. The PostgreSQL documentation defines xmin as the lowest transaction id that is still active.
  4. Only then read the data, and stamp the loaded rows with P.
  Take P before X. Also check pg_prepared_xacts: a prepared transaction stays active until it is committed or rolled back. On PostgreSQL older than 13 use the txid_* equivalents (txid_current_snapshot, txid_snapshot_xmax, txid_snapshot_xmin). pg_stat_activity.xact_start and backend_xmin are the wrong tools: neither says which transaction ids are still active. The functions and the definition of xmin are documented; that this makes P a safe stamp is my reasoning (every transaction that wrote before X has finished, and every later one wrote after P).
• A new, empty target can take the lowest possible position, so loaded rows only fill keys the stream has not touched.
• Per-key convergence only. The load gives no transactional consistency across rows while it runs: two rows changed by one source transaction can be visible at different versions until the stream catches up.

This is my reasoning from those two sources, not a vendor procedure. The unit tests reproduce the straddling transaction with the unsafe stamp and with a lower bound; prove it in staging with a long transaction that starts before the read and commits after it (the runbook has the step). Debezium’s own handoff is described as streaming from the position read at the start of the snapshot; that page describes no sink-side comparison.

Do not stamp the loaded rows with the position after the read, or with a timestamp. A position taken after the read can be above a change the read did not see, and then the older loaded row beats the newer streamed one.

The correctness rule

A backfill produces events that must not overwrite newer streamed changes. The sink enforces that itself; it cannot rely on the order in which the reload and the stream happen to arrive. The statement is the same one used for the stream, applied to backfill rows too:

CREATE TABLE customers (
  id         BIGINT  PRIMARY KEY,
  email      TEXT,                          -- NULL on a delete
  is_deleted BOOLEAN NOT NULL DEFAULT FALSE,
  source_lsn BIGINT  NOT NULL               -- position of the change that wrote this row
);

-- Apply ONE row, from the stream or from a backfill. For a backfill row,
-- source_lsn is the snapshot's boundary position (see the sections above).
INSERT INTO customers (id, email, is_deleted, source_lsn)
VALUES (7, 'a@example.com', FALSE, 150)
ON CONFLICT (id) DO UPDATE
SET email      = EXCLUDED.email,
    is_deleted = EXCLUDED.is_deleted,
    source_lsn = EXCLUDED.source_lsn
WHERE customers.source_lsn < EXCLUDED.source_lsn;

The strict < does three jobs. A backfill row with a lower position than the stored one is skipped, so it cannot overwrite a newer streamed change. A delete marker that is stored with a position blocks a backfill row from the time before the delete, so the key cannot come back. And applying the same row twice does nothing, because the second copy has an equal, not a higher, position. The unit tests for this page apply the same seven rows in every one of the 5,040 possible orders, and then replay them again in reverse; every run ends in the same table.

Three consequences are easy to miss:

• The backfill rows must carry positions, in the same unit the stream uses. A reload that stamps the load time, or that writes with a plain upsert, overwrites newer rows. A plain upsert keyed on the primary key, such as the Debezium JDBC sink’s insert.mode=upsert, is idempotent for replays of the same event; the pages I read describe it as insert-or-update on the key and mention no comparison of positions, so an older event arriving later still replaces a newer row. Compare Which Row Wins in Your Target for the setting in each warehouse.
• The guard skips corrections that keep the same position. If you reload to fix a transform bug, the corrected rows carry the positions the bad ones had, so the strict guard skips them. For a content fix, load into a new table with the corrected transform, let it catch up from the stream, check it, and switch readers over. Do not loosen the guard on the live table to <=: a row with an equal position would then replace the stored one, so a late backfill row stamped at the shared boundary would beat a sweep marker or a streamed change at that position.
• Ties need an ordinal. If two changes to one key can share a position, compare the pair (position, ordinal). See Making Sinks Bulletproof.

Rows the snapshot never mentions

A snapshot emits the rows that exist when it reads. A row deleted at the source while the sink was missing events is not in the snapshot, and its delete event died with the log, so the old row stays live in the sink. Nothing about the guard can fix that: it only compares rows it receives.

The remedy is a sweep: for the keys the backfill covered, mark every live sink row that is not in the source as deleted, with a position at the boundary. The marker keeps the key deleted if an old update turns up later.

-- backfill_keys: the keys the backfill read from the source, loaded into the sink's database.
-- :boundary is the position the backfill rows carry; :lo and :hi bound the keys the backfill covered.
UPDATE customers
SET email = NULL, is_deleted = TRUE, source_lsn = :boundary
WHERE NOT is_deleted
  AND source_lsn < :boundary
  AND id BETWEEN :lo AND :hi
  -- Repeat the backfill's additional-conditions filter here, if it had one:
  -- AND tenant_id = 5
  AND id NOT IN (SELECT id FROM backfill_keys);

• Scope it exactly like the backfill. Without the id BETWEEN bound, a filtered backfill would mark every row it did not read as deleted. A key range is not enough when the backfill also had an additional-conditions filter: a tenant-5 backfill swept with only lo = 10 and hi = 900 would mark live tenant-6 rows in that range as deleted. Repeat the filter in the sweep (the commented line), which means the sink needs that column.
• The key list must be complete for the scope. Under the runbook the backfill is applied first, so backfilled rows already sit at the boundary and the sweep leaves them alone. The hazard is a key that is live at the source but missing from both the list and the load (the list was built with a different filter or in a different pass): the sweep marks it deleted at the boundary, and a later row for it at that same position is skipped by the strict guard, so it stays deleted until the stream next changes it. The unit tests demonstrate it, and also that a sweep which runs before the load, with a list that lacks a key, has the same effect. Read the list from the source in one pass.
• Use the same lower bound for :boundary. With the stamp rule from choice 5, a higher boundary would sweep a row that a transaction straddling the read inserted and the stream already applied.
• Why a key list and not the position alone. Rows from an incremental snapshot may not all carry one boundary (check a sample), and rows the connector drops because the stream changed them during a chunk window never reach the sink as snapshot rows. The position test still matters: it protects rows the stream wrote after the list was read.
• Check the count first. Run the same WHERE as a SELECT COUNT(*) and read the number before you update. A large number after a short outage means the scope or the boundary is wrong.

The delete marker itself, and when a marker may finally be purged, are covered on Deletes That Stay Deleted. A purge bound must account for a backfill: the lowest position a replay source can still deliver now includes the snapshot you just ran.

Re-running a backfill, and the overlap

A backfill that fails halfway has to be restartable, and an operator will run it twice by mistake. With the guard from the rule, a second run of the same rows does nothing, and the sweep is idempotent as well: it only touches rows that are still live and below the boundary. The unit tests run both twice and compare the tables.

Expect duplicates in these places and design for them rather than hunting for them:

• An incremental snapshot chunk that is retried after a failure.
• The window around a blocking snapshot’s start, which the documentation says can duplicate rows.
• A change that is in both the snapshot rows and the stream. The guard keeps one version.
• A replay that starts earlier than strictly necessary, on purpose, to be safe.

None of these needs a deduplication step in front of the sink, because the sink already compares positions. Deduplication by event id, which an outbox relay uses, is a different tool for a different problem.

Cost and throttling

• Read load on the source. Chunk queries compete with application traffic. Smaller chunks mean shorter queries and more of them; incremental.snapshot.chunk.size (1024) is the lever. For the initial snapshot, snapshot.fetch.size (10240 rows per batch by default) and snapshot.max.threads (1 by default; above 1 the connector splits tables by primary-key range) set the pressure. A table without a primary key is processed as a single chunk on one thread.
• Pausing. You can pause and resume an incremental snapshot with signals instead of killing it, for example at the start of a peak.
• Streaming lag. A blocking snapshot stops streaming for its duration, so lag on every captured table grows until it ends. On PostgreSQL my reading is that a connector that is not consuming the slot also lets WAL accumulate, so watch the retained WAL as described on Postgres Replication Slots & WAL Growth.
• Topic size. An ad hoc snapshot appends the table to its existing topic. Check that retention and disk can hold the live events and a full copy of the table.
• Sink write rate. A full reload is a burst of upserts. Size batches and warehouse capacity for it, and run it off peak. The guard makes the work harmless, not free: a skipped row still costs a lookup.

This page gives no cost figures. Any number would depend on your source, sink and prices.

Verify the result

Do not stop at “the job finished”. The checks below fail on the three ways a backfill goes wrong: missing rows, stale rows and ghost rows. More patterns are on Testing a CDC Pipeline, Reconciliation and Offset Surgery and the Merge cookbook.

1. Counts by key range. Count live sink rows per bucket and compare with the same count at the source. Exclude marker rows (is_deleted) on the sink side.
2. Checksums by key range. Hash a payload per bucket on both sides. The hash function is database-specific; the example below is PostgreSQL syntax.
3. Suspect rows. After a complete backfill and sweep, no live row in scope should sit below the boundary. A non-zero count means ghosts, an incomplete load or the wrong boundary.
4. Max position per key. The highest stored position per bucket should be at or above the boundary and no higher than the newest position the stream has delivered.
5. Spot-check recent keys. Change a few rows at the source during the backfill on purpose, and confirm the sink holds the newest value, not the snapshot’s.

-- Counts and the highest position per 1000-key bucket (runs on PostgreSQL and SQLite).
SELECT id / 1000 AS bucket, COUNT(*) AS live_rows, MAX(source_lsn) AS max_lsn
FROM customers
WHERE NOT is_deleted
GROUP BY id / 1000
ORDER BY bucket;

-- Suspect rows after the sweep: should be 0 (same predicate as the sweep).
SELECT COUNT(*) AS suspects
FROM customers
WHERE NOT is_deleted
  AND source_lsn < :boundary
  AND id BETWEEN :lo AND :hi;

-- Payload checksum per bucket, PostgreSQL syntax. Run the equivalent at the source and compare.
SELECT id / 1000 AS bucket, COUNT(*) AS live_rows,
       md5(string_agg(id::text || ':' || COALESCE(email, ''), ',' ORDER BY id)) AS checksum
FROM customers
WHERE NOT is_deleted
GROUP BY id / 1000
ORDER BY bucket;

The first two queries are executed by the unit tests on SQLite. The checksum query uses md5 and string_agg, which SQLite does not have; the unit tests run it on a PostgreSQL-compatible engine (PGlite). Checksums also have blind spots: columns you do not hash, and rows the source changed since you read. See the lies section of Testing a CDC Pipeline.

Runbook: reload history without breaking the sink

1. Name the reason from the table, and the scope: which tables, which key range.
2. Check the sink first. It must key on the primary key, carry a never-NULL position column, compare positions with a strict <, and keep deletes as markers. Fix that before any load. If it does not, do not start.
3. Rehearse in staging with a copy of the table. Include a long transaction that starts before the snapshot and commits after it, and confirm that the changed row ends with the newer value.
4. Check the log is still there from the position you need. On MySQL, binlog retention: a snapshot that runs longer than it loses the handoff. On PostgreSQL the slot keeps WAL for you, so the risk is the opposite: disk growth while the snapshot runs, or the slot being invalidated by max_slot_wal_keep_size (see the slots runbook). In Kafka, topic retention.
5. Pick the choice from the table and write the signal or the command down before you run it, including the filter.
6. For choices 2 and 3 on PostgreSQL, record the slot’s restart_lsn before the run and check a sample. My reasoning, not a vendor statement: the straddling-transaction hazard of choice 5 may apply if an incremental-snapshot read event carries a position near the close of its chunk window. After the run, take a few keys that changed during it and confirm the sink holds the newer value.
7. Inspect a sample of the first backfill rows: op, source.snapshot and the position they carry. If the position is not what the guard expects, stop.
8. Set the pace (chunk size, threads, fetch size), and agree who may pause the snapshot if the source is under load.
9. Run it and watch lag, retained WAL, topic size and the sink’s write latency.
10. Wait until the sink has applied every backfill row. Consumer lag on the topic is zero and the snapshot is reported complete.
11. Sweep rows the source no longer has, using a complete key list for the same scope, and read the count before you commit it.
12. Verify with the counts, checksums, suspect-row query and spot checks above. Record the boundary, the counts and the checks.
13. Put the settings back. A snapshot.mode of always repeats the full load on every restart. Remove temporary signal permissions and filters.
14. Recompute the purge bound for delete markers, since the backfill just added a replay source.
15. Check the positions are comparable. My reasoning, not a vendor statement: positions only compare within one source log lineage. After a source rebuild, restore or migration, a new cluster can restart its LSN or binlog numbering below the source_lsn already stored, and the guard would then skip every new change. Start a fresh target, or re-key the positions. A MySQL position is a (file, position) pair, so encode it into one number that sorts the same way.

What was tested and what was not

• No Debezium, Kafka, Kafka Connect or database service was run for this page. Every statement about them comes from the documentation linked below, read on 2026-10-09.
• The guarded upsert, the interleaving of backfill and stream rows (all 5,040 orders of seven rows, plus a reversed replay), the idempotent re-run, the sweep and its scoping, the incomplete-key-list hazard, and the counts and suspect-row queries are executed in the site’s unit tests on SQLite only. The straddling-transaction case also runs on PGlite, a PostgreSQL-compatible engine, in the same unit tests.
• The payload checksum query uses PostgreSQL-only functions, so the SQLite tests cannot run it. It runs on PGlite in the same unit tests, on a few fixed rows.
• The execute-snapshot example follows the documentation and was not run.
• Check your version: the position that read events carry on PostgreSQL and for incremental snapshots; whether events of a transaction that straddles the boundary carry a higher position; the exact snapshot.mode names on your release; and how your sink treats an equal position.
• Not covered: sizes and prices, connector-specific tuning beyond the properties named above, and backfills of non-relational sources.

Further resources

On this site

• Snapshotting: the handoff, watermarks and idempotent apply.
• Postgres Replication Slots & WAL Growth: how a slot is lost and what follows.
• Deletes That Stay Deleted and Which Row Wins in Your Target.
• Testing a CDC Pipeline, Merge cookbook and Reconciliation and Offset Surgery.
• Glossary: backfill, watermark, upsert, replication slot, at-least-once.

Primary sources (read 2026-10-09)

• Debezium: PostgreSQL connector (snapshots, snapshot.mode, incremental and blocking snapshots, ad hoc snapshots, configuration properties).
• Debezium: MySQL connector (the read-event position, the deprecated and renamed snapshot modes).
• Debezium: sending signals to a connector (pause, resume and stop).
• Debezium: JDBC sink connector (upsert mode).
• Kafka: resetting consumer group offsets, Kafka Connect REST offsets endpoints, topic configs and log compaction.
• DBLog: a watermark based change-data-capture framework, the design Debezium’s incremental snapshots build on.