CDC Pipeline That Silently Loses Rows

errors.tolerance=all makes the sink skip records that fail serialisation or conversion and continue, committing offsets past them. Failures are not retried and not retained. With errors.log.enable=false and no dead-letter queue, the records are gone - the connector is healthy, the offsets advance, and the data is missing. That is precisely the 6% gap.

To confirm and quantify:

• Set errors.log.enable=true and errors.log.include.messages=true on a canary task, or better, configure errors.deadletterqueue.topic.name plus errors.deadletterqueue.context.headers.enable=true, then restart the task and inspect the DLQ topic for the failing payloads and their error headers.
• Compare the sink's committed offsets against the source topic's end offsets for the affected window - the delta is the loss surface.
• Check whether the source topic is compacted: for a compacted topic you can re-read from the earliest offset and replay into the lake, because the failing records are still on disk (subject to retention).

errors.tolerance=all is a data-loss switch when paired with no DLQ. Correct configuration is errors.tolerance=none (task fails loudly) or all with a DLQ and an alert on DLQ depth.

Schema change. Adding a nullable column with a default is backward compatible, so the source's schema registers fine; the break is downstream. Verify the sink's converter tolerates the extra field (schemas.enable and the Avro/JSON converter settings) and pin Schema Registry compatibility to BACKWARD_TRANSITIVE so future source changes are validated before they reach the lake. If the S3 sink writes Parquet, the writer's schema handling is stricter than JSON - a schema-cache or partition-schema mismatch will reject the new shape, which is the likely original failure. Fix by restarting with a refreshed schema rather than by tolerating errors.

Deletes. A delete is a record with the key and a null value (tombstone). Two things must hold:

• The transform must not drop tombstones before the sink sees them - configure delete handling deliberately (drop.tombstones on/off, or rewrite to a __deleted marker).
• The sink must act on it. For a JDBC sink this means delete.enabled=true, which requires pk.mode=record_key with pk.fields set. For a lake sink, you need a delete-aware write mode; otherwise the tombstone is just a skipped null value and the row survives forever.

Also remember the tombstone only lives on the compacted topic for delete.retention.ms (default 24 h) - replay after that cannot recover deletes.

1. Stop the bleeding: restart the sink task with errors.tolerance=none and a DLQ configured, so no further silent skips happen. Confirm the source topic's retention covers the four-day window.
2. Land into a staging prefix, not the production table: replay the source topic from the offset corresponding to the start of the gap into s3://lake/staging/customers_replay/ using a second sink connector or kafka-console-consumer-based extraction. Replay keeps the original keys and timestamps.
3. Reconcile deterministically by key: the lake table must be treated as an upsert keyed on the customer primary key, taking the record with the highest source offset (or __source_ts_ms), and honouring __deleted tombstones. Do not append the replay - merge it. MERGE/INSERT OVERWRITE on a keyed table, or a compaction job over the two prefixes, gives you idempotent recovery.
4. Verify: row count and a per-key checksum against the source for the affected window, and an explicit check that deleted keys are absent.
5. Prevent recurrence: alert on connector task failure and DLQ depth, keep errors.tolerance=none for sinks with business-critical data, and add a freshness/lag monitor on the lake table.