7. Offsets and delivery

Chapter 6 ended on a periodic offset flush. This chapter is what that means for correctness.

What an offset is

A position in the source database’s log, not in Kafka.

   POSTGRESQL   LSN + txId
                {"lsn": 439108255,
                 "txId": 8841207,
                 "lsn_commit": 439108255}

   MySQL        file + position, or GTID set
                {"file": "mysql-bin.000042",
                 "pos": 1049382,
                 "gtids": "3E11FA47-...:1-8841207"}

   MONGODB      a resume token
                {"resume_token": "826512A9..."}

   SQL SERVER   change LSN + commit LSN

This is the answer to “where do I resume?” and it is entirely about the source, which is why chapter 3 called it production data.

The replay window

offset.flush.interval.ms=60000     # a common default

   t=0s    offset flushed. LSN 439108255.
   t=0-60s 41,000 events produced to Kafka
   t=58s   THE CONNECTOR CRASHES
   t=70s   it restarts, reads the last
           flushed offset: 439108255
   t=70s   it resumes from there

   → THOSE 41,000 EVENTS ARE PRODUCED AGAIN

Every one of them is a duplicate. Same rows, same values, same keys — new Kafka offsets.

   offset.flush.interval.ms

   LOWER   (5s)   fewer duplicates on
                  restart, more writes to
                  the offset store
   HIGHER  (60s)  fewer writes, a larger
                  replay window

   → this dial trades duplicate volume
     against overhead. it does NOT trade
     away duplicates.

Lowering it does not make delivery exactly-once. There is always a window, because the offset is written after the records, and no ordering of those two makes them atomic.

At-least-once, and why

   Debezium produces the record
   Kafka acknowledges it
   → THEN the offset is flushed

   crash between those two
     → the record is in Kafka
     → the offset says it was not
     → it is produced again
     → DUPLICATE. never loss.

   the other order — flush the offset first —
     → crash
     → the offset says done
     → the record was never produced
     → LOSS.

Duplicates are the deliberate choice, because a duplicate is recoverable by an idempotent consumer and a loss is not recoverable by anything.

Duplicates are not rare and not an error condition. They happen on every restart, every rebalance, every deploy, every failover, and during every snapshot (chapter 5). A consumer that cannot tolerate them is broken, not unlucky.

Making consumers idempotent

The one obligation CDC places on everyone downstream.

The good news: change events are unusually easy to make idempotent, because every event carries the full row and its primary key.

   ✓ UPSERT BY PRIMARY KEY
     the same event applied twice produces
     the same row. this covers most sinks.

     Elasticsearch: PUT /orders/_doc/4471
     Postgres:      INSERT ... ON CONFLICT
                    (id) DO UPDATE
     ClickHouse:    ReplacingMergeTree
     Iceberg:       MERGE INTO

   ✓ VERSION-AWARE UPSERT
     keep source.lsn (or ts_ms) alongside the
     row; ignore an event whose version is
     not newer.
     → also protects against OUT-OF-ORDER
       reprocessing, which upsert alone
       does not

   ✓ A DEDUPLICATION KEY
     source.lsn + table is unique per change.
     store recent ones; skip repeats.
     → for sinks that cannot upsert

   ✗ BLIND APPEND
     "INSERT a row into an audit table per
      event" → duplicates accumulate silently

   ✗ SIDE EFFECTS
     sending an email, charging a card,
     calling an external API
     → the dangerous case. see below.

ClickHouse’s ReplacingMergeTree exists for exactly this, and Iceberg’s row-level operations describe the same pattern with equality deletes. Both guides arrive at upsert-by-key independently, because it is the answer.

Side effects are the real problem

   "send a shipping email when status
    becomes 'shipped'"

   the connector restarts
   → the same change event arrives twice
   → TWO EMAILS

No upsert saves you here, because the effect is outside your database.

   THE FIXES

   1  DEDUPLICATE ON source.lsn
      store "we have acted on lsn 439108255
      for order 4471". check before acting.
      ← the general answer

   2  MAKE THE EFFECT IDEMPOTENT UPSTREAM
      an idempotency key on the email API,
      on the payment API. most good APIs
      support this.

   3  ACT ON STATE, NOT ON EVENTS
      "every 5 minutes, email everyone whose
       status is 'shipped' and email_sent is
       false" — driven by a table CDC keeps
       up to date.
      ← turns an at-least-once event stream
        into an idempotent reconciliation

Option 3 is underrated. It converts the hard problem into the easy one, and it survives more than duplicates — it survives the consumer having been down.

Ordering

What is guaranteed, precisely, because people assume more than is true.

   ✓ GUARANTEED
     all changes to ONE ROW, in commit order
     — because the key partitions them to
       one partition (chapter 6)

   ✓ GUARANTEED
     within one table, per key

   ✗ NOT GUARANTEED
     across different rows
   ✗ NOT GUARANTEED
     across different tables
   ✗ NOT GUARANTEED
     across topics

   INSERT INTO orders ...;        → topic A
   INSERT INTO order_items ...;   → topic B
   COMMIT;

   → a consumer may see the item BEFORE the
     order. routinely does.

Per-row ordering is usually all you need, and it is genuinely guaranteed. When you need more — a consumer that must not see a child before its parent — the options are: put both tables in one topic keyed by the parent, use transaction metadata (chapter 4), tolerate the reordering with retries, or join downstream in Flink.

Exactly-once

Kafka Connect supports exactly-once for source connectors, and Debezium can use it.

exactly.once.source.support=enabled          # on the Connect worker
exactly.once.support=required                # on the connector

   HOW: the offset write and the record
   production happen in ONE KAFKA
   TRANSACTION. either both or neither.

   → no replay window
   → no duplicates from a restart

Read the qualifications carefully.

   ✓ exactly-once INTO KAFKA
   ✗ exactly-once END TO END
     a consumer with side effects still
     needs idempotence

   ✗ it does not eliminate SNAPSHOT
     overlap duplicates (chapter 5) —
     those are by design

   ~ throughput cost: transactions add
     overhead

   ~ CONSUMERS MUST READ COMMITTED
     isolation.level=read_committed, or
     they see aborted records

   ~ connector support varies. verify for
     yours and your version.

It is worth enabling where supported, and it does not remove the obligation to be idempotent. Design consumers as if duplicates will happen, then enable exactly-once as defence in depth.

The schema history topic

For MySQL, SQL Server and Oracle — a second piece of state, and losing it is worse than losing offsets.

schema.history.internal.kafka.topic=schema-history.orders-cdc

It stores every DDL statement the connector has seen. On restart, Debezium replays it to reconstruct what each table looked like at each point in the log — because a binlog event from three days ago must be interpreted against the schema as it was three days ago, not as it is now.

   ⚠ SINGLE PARTITION
   ⚠ INFINITE RETENTION (never compacted,
     never deleted)
   ⚠ replication factor 3
   ⚠ NEVER let a cleanup policy touch it

   losing it → the connector cannot start
             → recovery, or a full
               re-snapshot

This topic is the most commonly damaged piece of Debezium state, usually by a well-meaning retention policy applied cluster-wide. Chapter 14 has it as a failure mode.

Losing offsets

   → Debezium has no idea where it was
   → snapshot.mode decides what happens

     initial       → FULL RE-SNAPSHOT
     never         → start from NOW.
                     everything between the
                     loss and the restart is
                     GONE.
     when_needed   → full re-snapshot,
                     unattended

Neither outcome is good. initial means hours of load and a snapshot flood; never means silent data loss. Back up the offset store, and for Debezium Server put it on durable storage (chapter 3).

Resetting offsets deliberately

Sometimes you want to. There is no clean button.

   THE OPTIONS

   1  INCREMENTAL SNAPSHOT  ← almost always
      the right answer
      signal a re-snapshot of the affected
      tables. streaming continues. no offset
      surgery.

   2  A NEW CONNECTOR NAME
      offsets are keyed by connector name.
      a new name = a clean slate, with a new
      slot. delete the old slot afterwards,
      or chapter 2 happens.

   3  EDIT connect-offsets DIRECTLY
      produce a tombstone for the connector's
      offset key. fiddly, error-prone, and
      the connector must be stopped.

Reach for the incremental snapshot. It is the feature that made offset surgery mostly unnecessary, and the other two options carry real risk.

What to take from this chapter

• An offset is a position in the source database’s log, not in Kafka.
• Offsets are flushed periodically, so a restart replays everything since the last flush. Lowering offset.flush.interval.ms shrinks the window; nothing closes it.
• At-least-once is deliberate: records are produced before offsets are written, because a duplicate is recoverable and a loss is not.
• Duplicates happen on every restart, deploy, rebalance and snapshot. A consumer that cannot tolerate them is broken.
• Upsert by primary key covers most sinks, and change events carry the full row and key, which makes this easy. Add a version check against source.lsn to survive reordering too.
• Side effects are the hard case — deduplicate on source.lsn, use an upstream idempotency key, or drive the effect from state rather than from events.
• Ordering is guaranteed per row and nothing more. A child row can arrive before its parent.
• Connect’s exactly-once gives exactly-once into Kafka and does not remove the need for idempotent consumers.
• The schema history topic must be single-partition, never compacted, never expired, and losing it is worse than losing offsets.
• Losing offsets means a full re-snapshot or silent data loss, depending on snapshot.mode.
• To re-emit data, signal an incremental snapshot rather than editing offsets.

Next: topics and routing — naming, partitioning, and how to decide what goes where.