CDC Without Kafka: Server, Engine, Search Sync

Taking Kafka away changes less than you think

Kafka gives a CDC pipeline a replayable log, partitions that keep a key in order, a place to store connector offsets and, with an opt-in, a documented exactly-once hop. Run Debezium without it and you lose those, but the delivery guarantee does not get better: it is at-least-once, duplicates follow every crash, and the destination still has to decide which version of a row wins. End-to-end exactly-once across independent systems is not achievable in general. What you can reach is effectively-once processing: an idempotent write keyed on the primary key and ordered by the source log position, with deletes kept as markers.

This page covers four non-Kafka paths: Debezium Server, the embedded engine, managed CDC services, and a consumer you write yourself that keeps a search index or cache in sync. For the Kafka-side claims see Is CDC exactly-once?; for the per-target ordering rules see Which row wins. Transport semantics (Pulsar, Kinesis, Pub/Sub) are in Transports other than Kafka; the tool landscape is in Tooling and Compare.

As of 2026-10-09. Read against the Debezium 3.7 documentation. The Elasticsearch and Redis behaviour in the search section was run on Elasticsearch 9.5.5 and Redis 8.10.2, single local nodes. Nothing else on this page was run against a live product, and where a fact comes from source code it says so and says “check your version”.

Which path for which need

Debezium Server: one connector, one sink, offsets in a store you pick

Debezium’s documentation describes Debezium Server as a ready-to-use application that streams change events from a source database directly to messaging and storage systems without requiring Apache Kafka. Each instance runs exactly one connector. You choose the destination with debezium.sink.type, prefix connector settings with debezium.source., and set the offset store yourself, because the Kafka Connect runtime that normally does it is not there. The excerpt below is the documentation’s own Kinesis example (not run by this site):

debezium.sink.type=kinesis
debezium.sink.kinesis.region=eu-central-1
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=localhost
debezium.source.topic.prefix=tutorial

Where the offset lives, and what a crash does

Debezium’s state-storage page says that if you run a connector in Debezium Server or the embedded engine you must configure the offset store explicitly. The options it documents:

• File (FileOffsetBackingStore, the Server default): a local file; the page calls it ideal for single-node applications or testing.
• JDBC (JdbcOffsetBackingStore): a table, debezium_offset_storage by default, in the source database or another one.
• Redis (RedisOffsetBackingStore): a hash key, metadata:debezium:offsets by default on the Server page.
• Kafka (KafkaOffsetBackingStore): a compacted topic. It is documented on the storage page; the Server page lists only file, memory, JDBC and Redis, so check that your Server version accepts it.
• Memory: volatile, lost on a crash, for tests only.

In every case the store is written by the engine on its own schedule. Nothing in these pages puts the offset in the same transaction as your destination write. The embedded engine documentation states the consequence for a crash: the application “will likely receive some source records after restart that it had already processed”, up to n duplicates with a flush interval of 0 (n being the batch size) and up to n × m with a longer interval. Set offset.flush.interval.ms explicitly: the engine page says its default is one minute while the storage page lists 6000 ms, so check your version.

Per-sink facts that matter for duplicates and order

Debezium’s Kafka Connect exactly-once opt-in is documented for connectors deployed in Kafka Connect. These pages do not document an equivalent for Debezium Server. There is no Elasticsearch or OpenSearch sink in the Server sink list I read, so the search path below is a consumer you write.

The embedded engine: you own the commit

The engine is a library (debezium-api plus debezium-embedded) that runs a connector inside your application. From Debezium 3.2 the default implementation is the asynchronous engine, which Debezium Server also uses. When you need batching or an async API, the documentation points to DebeziumEngine.ChangeConsumer, whose single method is handleBatch(List<R> records, RecordCommitter<R> committer). Per its Javadoc you call markProcessed for each record and markBatchFinished when the batch is done.

The commit-after-process contract

1. Write the record’s effect to your destination.
2. Only then call markProcessed(record). It registers that record’s source offset to be flushed; in the engine source on main it does that and markBatchFinished asks the commit policy whether to flush now.
3. Call markBatchFinished() at the end of the batch.
4. If a write fails, do not mark that record or any later one, and stop the engine. My reading, untested: the connector’s in-memory position has already moved on, so only a restart re-reads from the last stored offset.
5. Stop the engine with close(), not by shutting down the executor. The documentation says an interrupt can leave it unclean and the restarted application may see records it had already processed.

Marking a record before its write is durable turns a crash into data loss, because the stored offset has moved past a change the destination never got. Marking after turns a crash into a repeat, which an idempotent write absorbs. The simple notifying(record -> ...) form has a documented hazard: the page says the handler should not throw, and that an exception is logged and the engine moves on to the next record, so your application might become inconsistent with the database. The default wrapper in the source on main appears to propagate other exceptions instead, so docs and code may disagree; I did not run either. Catch inside the handler and decide.

// Untested sketch. Types follow the engine documentation.
class IdempotentSink implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {
  @Override
  public void handleBatch(List<ChangeEvent<String, String>> records,
      RecordCommitter<ChangeEvent<String, String>> committer) throws InterruptedException {
    for (ChangeEvent<String, String> r : records) {
      applyIfNewer(r);               // upsert on the key, only if source position is higher
      committer.markProcessed(r);    // after the effect is durable, never before
    }
    committer.markBatchFinished();
  }
}

The documentation also says: “applications will receive each source record exactly once during normal operation (including restart after a graceful shutdown), but do need to be tolerant of receiving duplicate events immediately following a restart after a crash or improper shutdown”. It then says applications needing more rigorous exactly-once behaviour should use “the full Debezium platform”. Read that against the Debezium exactly-once page, which documents only an opt-in with caveats for connectors on Kafka Connect. For the engine the safe design is the first sentence: expect duplicates after a crash and make the write idempotent.

Managed CDC services: only what their pages state

Absence from a page is not a guarantee either way; ask the vendor, in writing, which position orders the destination and what happens on a restart. If you run your own merge on top of a landing table, the rule from Which row wins applies unchanged. The vendor labs are AWS DMS, Fivetran and Snowflake.

Syncing a search index or cache: the log position is the version

Elasticsearch’s own index API documentation makes the point this site makes: with external versioning “there is no need to maintain strict ordering of async indexing operations run as a result of changes to a source database, as long as version numbers from the source database are used”. AWS says the same about Kinesis consumers: with OpenSearch “you can use a combination of versioning and unique IDs to prevent duplicated processing”. The recipe: document ID = primary key, document body = the full after image, version = the source log position, version_type=external.

What was run on Elasticsearch 9.5.5

The setting is documented as dynamic: index.gc_deletes is “the length of time that a deleted document’s version number remains available for further versioned operations. Defaults to 60s”. The cost of a long window (the pages read do not document it) needs your own measurement. OpenSearch documents the same version_type values and a dynamic index.gc_deletes with a 60 s default; I did not run OpenSearch.

Three consequences

• The version has to be one integer. A PostgreSQL lsn in a Debezium event is already an integer (it fits while it stays below 2⁵³ in JavaScript, about 9 PB of WAL). A SQL Server LSN has three parts (for example 0000003a:00002f50:0005) and a MySQL position is a file and an offset: you must pack them into one monotonic integer yourself, which is a design, not a feature, and positions from different log lineages are not comparable.
• A tie is silently dropped. If two changes to one key could share a version, plain external keeps the first and refuses the second. Debezium’s PostgreSQL page does not say whether events in one transaction can share an lsn, so fold an ordinal into the integer or test your source.
• Deletes need a plan. Either delete with the version and set index.gc_deletes above the longest replay you can suffer, or index a marker document (deleted: true) and filter it out of queries, purging it only once no replay can reach back that far. The first is what the page measured at 60 s by default; the second matches the deletes that stay deleted lesson.

A consumer sketch (request shapes run; the Debezium side untested)

// Node 20+. Debezium JSON events with the PostgreSQL source block.
export function toBulkLines(events, { index, idField = "id", softDelete = false }) {
  const lines = [];
  for (const e of events) {
    if (e == null || e.op == null) continue; // a Kafka-style tombstone or a non-change record
    const row = e.op === "d" ? e.before : e.after;
    const version = e.source.lsn;
    if (!Number.isSafeInteger(version) || version < 0) {
      throw new Error(`position ${version} is not a safe non-negative integer`);
    }
    const meta = { _index: index, _id: String(row[idField]), version, version_type: "external" };
    if (e.op === "d" && !softDelete) {
      lines.push(JSON.stringify({ delete: meta }));
    } else {
      lines.push(JSON.stringify({ index: meta }));
      lines.push(JSON.stringify({ ...row, deleted: e.op === "d" }));
    }
  }
  return lines.length ? lines.join("\n") + "\n" : ""; // "" means nothing to send
}

// Returns indexes to retry (429, 5xx) and indexes that were rejected (everything else).
export function classifyBulk(response) {
  const retry = [];
  const rejected = [];
  response.items.forEach((item, i) => {
    const r = Object.values(item)[0];
    if (r.status < 300) return; // applied
    if (r.status === 409 && r.error?.type === "version_conflict_engine_exception") return; // stale or duplicate
    if (r.status === 404 && r.result === "not_found") return; // delete of an absent doc; the version was still recorded
    if (r.status === 429 || r.status >= 500) retry.push(i);
    else rejected.push(i);
  });
  return { retry, rejected };
}

If toBulkLines returns an empty string (every event was skipped), do not call /_bulk: an empty body is an error, so use if (!body) return; and commit the offsets. Otherwise send the lines to POST /_bulk with Content-Type: application/x-ndjson. Commit the source offset only after every item is applied, a 409, or sent to a dead-letter store; a batch with a retry item is re-sent whole, which is safe because the write is idempotent. The two functions above were run against Elasticsearch 9.5.5 with hand-made events: a create and an update returned 201 and 200, a replayed older change and a duplicate returned 409 and 409, a delete returned 200, a late update after it returned 409, and the marker mode kept {"id":2,"deleted":true} at version 300 against a late update. No real Debezium event was used; this repository’s unit test also runs them on hand-made events.

Redis as a cache

Two honest options. Invalidate, do not write: on every change event delete the cache key and let the reader reload from the database. A duplicate or a reordered event can only cause an extra miss. Debezium documents this pattern for Hibernate’s second-level cache (an incubating Quarkus extension that evicts on update and delete). Race: a reader that loaded the old row before the change can still write it back after the eviction, which is why populating needs a guard. Guarded write: Redis has no compare-and-set on a numeric version, so use a script, which Redis documents as atomic. A plain SET has no guard: on Redis 8.10.2, SET b then SET a left a.

-- KEYS[1] cache key; ARGV[1] version (log position, integer below 2^53);
-- ARGV[2] 'd' for delete; ARGV[3] document; ARGV[4] tombstone TTL in seconds.
-- Returns 1 when applied, 0 when stale or duplicate.
local current = redis.call('HGET', KEYS[1], 'v')
if current and tonumber(current) >= tonumber(ARGV[1]) then return 0 end
if ARGV[2] == 'd' then
  redis.call('HSET', KEYS[1], 'v', ARGV[1], 'deleted', '1', 'doc', '')
  redis.call('EXPIRE', KEYS[1], tonumber(ARGV[4]))
else
  redis.call('HSET', KEYS[1], 'v', ARGV[1], 'deleted', '0', 'doc', ARGV[3])
  redis.call('PERSIST', KEYS[1])
end
return 1

Run on Redis 8.10.2 with redis-cli --eval: v100 then v200 applied (1, 1); replay v100 and a duplicate v200 returned 0 and left v200 in place; a delete at v300 with a 3 s tombstone TTL made a late v200 return 0, and after the TTL expired the same late v200 returned 1 and the key came back. Redis lets you choose the tombstone TTL, so the window is yours, but it has the same shape as gc_deletes. Lua has one numeric type; the probe showed 2⁵³+1 refused as “not newer” than 2⁵³, so keep versions below 2⁵³. Readers must treat deleted = 1 as a miss.

Where exactly-once stops, again

None of these paths adds a shared transaction between the source and the destination. Debezium documents at-least-once delivery and no internal deduplication layer; the engine page tells you to expect duplicates after a crash; Datastream documents at-least-once; Zerobus tells you to deduplicate by source LSN. The achievable claim is exactly-once processing: a position-ordered idempotent write makes the duplicates harmless. See Exactly-once semantics and Is CDC exactly-once?.

Try it: reproduce the probe

With a throwaway Elasticsearch on localhost:9200 (security off), this reproduces the central rows of the table. The commands match what was run, on port 19200 instead of 9200.

curl -s -XPUT 'localhost:9200/probe' -H 'content-type: application/json' -d '{"settings":{"number_of_replicas":0}}'
curl -s -XPUT 'localhost:9200/probe/_doc/7?version=100&version_type=external' -H 'content-type: application/json' -d '{"name":"a"}'
curl -s -XPUT 'localhost:9200/probe/_doc/7?version=200&version_type=external' -H 'content-type: application/json' -d '{"name":"b"}'
curl -s -XPUT 'localhost:9200/probe/_doc/7?version=100&version_type=external' -H 'content-type: application/json' -d '{"name":"a"}'   # expect 409
curl -s -XDELETE 'localhost:9200/probe/_doc/7?version=300&version_type=external'
curl -s -XPUT 'localhost:9200/probe/_doc/7?version=200&version_type=external' -H 'content-type: application/json' -d '{"name":"b"}'   # 409 now, accepted after about 60 s

What to do Monday

1. Write down, per path in use, where the offset is stored and what a crash repeats. If you cannot say, that is the finding.
2. Carry the source log position into every destination write and make it the version or the merge guard.
3. In an embedded-engine handler, move markProcessed after the durable write.
4. For a search index, decide the delete plan (marker document, or index.gc_deletes above your longest replay) and test it with the curl commands above.
5. For each managed service, record what its page says about delivery and ordering, and what you could not find.
6. Replay a batch, swap a pair, and kill the consumer mid-batch in staging (test your pipeline).

Further resources

On this site

• Which row wins in your target and Is CDC exactly-once?
• Transports other than Kafka, Offsets & replays, Search indexing use case.
• Tooling, Compare, Non-relational sources.
• Glossary: Debezium Server, embedded engine, external versioning, at-least-once, deduplication, tombstone.

Primary documentation (read 2026-10-09)

• Debezium Server, Debezium Engine, storing connector state, exactly-once delivery, PostgreSQL connector, Hibernate cache invalidation.
• Source code read on the default branch: debezium-server sinks and AsyncEmbeddedEngine.
• Kinesis: handle duplicate records, PutRecords.
• Pub/Sub ordering, exactly-once delivery.
• Pulsar messaging concepts, deduplication cookbook; Redis XADD, Redis scripting, Lua API.
• Elasticsearch index API (versioning), delete API, index settings (index.gc_deletes); OpenSearch index document, index settings.
• Datastream events and streams, DMS Kinesis target, DMS S3 target, Airbyte CDC, Airbyte incremental append + deduped, Fivetran PostgreSQL.