Debezium to Pulsar to Delta Lake: making silent failures loud

At Happy-Place I work on a multi-tenant analytics platform. Its job sounds simple: every change in a customer’s operational database should end up in a lakehouse that people can query, without anyone writing pipeline code and without one tenant’s data ever touching another’s. Most sources are MongoDB; PostgreSQL, MySQL, Oracle and SQL Server work too.

The machinery behind that is a change data capture (CDC) pipeline. Debezium Server reads each database’s change log, Apache Pulsar carries the events, and Airflow-scheduled PySpark jobs land them in Delta Lake tables on MinIO.

Almost every serious bug we hit while building it had the same shape: something broke, and every component kept reporting success. This post is about those failures, and the rule they taught us: make failure loud.

The pipeline at a glance

1. SourceMongoDB change streams, or logical replication for relational databases.
2. CaptureDebezium Server 3.0, one pod per connector. Offsets, including MongoDB's resume token, live on a persistent volume.
3. TransportApache Pulsar. Each organisation, project and environment gets its own isolated space with its own credentials, and one topic per collection.
4. BronzeAirflow runs a Spark Structured Streaming job with availableNow: drain what's waiting, append the raw events to Delta, exit.
5. SilverRead bronze's change feed, fold each document's events in order, and MERGE the result into Delta.
6. GoldPer-organisation aggregates, queried through Trino.

Nobody configures a connector by hand. A control-plane service backed by PostgreSQL renders the Kubernetes manifests, commits them to a GitOps repository, and Argo CD applies them. That turns out to matter: several fixes below are “the control plane now does X before anything else runs”.

Why Pulsar

We chose Pulsar over Kafka for three reasons:

• Multi-tenancy is built in. Pulsar has tenants and namespaces, each with its own permissions and policies such as retention. Isolating customers is a configuration problem, not a separate cluster per customer.
• Storage and compute are separate. Brokers are stateless; Apache BookKeeper stores the data. Each side scales on its own.
• Retention and replay. Durable subscriptions and retained messages mean a consumer can stop for hours and pick up exactly where it left off.

Keep that last point in mind, because it’s exactly what went wrong.

1. The topics nobody read

The first time we wired Debezium to Pulsar, events didn’t arrive. Debezium’s routing rules rewrote each destination into a full persistent:// topic URI. The Debezium Server Pulsar sink then prefixed its own default tenant and namespace, public/default/, onto that URI, and Pulsar rejected the result as an invalid topic name. Bronze, meanwhile, was subscribed to the topics it expected and saw nothing at all, which looks a lot like a quiet database.

The fix was to stop fighting the sink. We removed the routing rewrite and set the sink’s tenant and namespace properties so it builds the full topic name itself. We also moved responsibility to the control plane: it now creates the namespace, every topic, the retention policy, the subscription policy and the access grants before Debezium starts. The producer never discovers its destination; the destination already exists.

That day also turned up a token-file key with the wrong case, an empty-body admin request that returned 412, and an unexpected 307 redirect. Each fix revealed the next. Three days later, after allowing Spark’s executor-to-driver traffic through a default-deny network policy, the full chain from connector to bronze to silver went green for the first time.

2. Retention zero: acknowledged meant deleted

The day after that first green run, silver started failing. Silver was fine; the real problem was one step earlier.

Pulsar’s broker retention defaulted to zero, so messages were deleted as soon as bronze acknowledged them. That’s harmless while everything runs on schedule. But after a 14-hour gap between runs, bronze’s Spark checkpoint pointed at positions whose data no longer existed, and the Pulsar connector correctly refused to continue with a “potential data loss” error. Every tenant was affected.

We fixed it in three parts:

• Seven days of retention, with no size cap, set on every namespace when it’s provisioned.
• No message TTL, deliberately. Retention keeps data for replay; a TTL would expire unread messages.
• failOnDataLoss stays on. Turning it off would have made the error go away by silently reading zero rows. A gap in the stream should stop the pipeline, not hide inside it.

The result is a property we now design around: an outage is a delay, not a loss. If the pipeline is down for a day, the next run catches up from where it stopped.

While we were in the broker config, we found a second timer. Our brokers were set to expire inactive subscriptions after 120 minutes, which would have deleted our pre-created notification subscription and then garbage-collected its topic. We pinned expiry to zero per namespace before it ever fired.

3. Don’t start on half a snapshot

Debezium’s default snapshot mode reads the existing data first, then streams changes. If silver runs halfway through that snapshot, downstream tables show half a customer’s data, and nothing about the result looks wrong.

So every connector’s medallion DAG is born paused. Debezium’s own snapshot notifications are switched on and sent through the same Pulsar sink to a notifications topic in the tenant’s space:

debezium.source.notification.enabled.channels=sink,log
debezium.source.notification.sink.topic.name=dbz-notifications

The control plane consumes that topic. When the “initial snapshot completed” event arrives, it unpauses the DAG and triggers it exactly once. Exactly once takes some care when notifications can be redelivered:

• A row lock on the first transition. The connector row is updated only where snapshot_completed_at IS NULL, under SELECT … FOR UPDATE, so two deliveries can’t both win.
• A deterministic run ID, snapshot_<notification_id>. If Airflow answers 409 Conflict, the run already exists, and that counts as success.
• A durable trigger. The completion event can arrive before the DAG has even been deployed, so the trigger is stored and retried with backoff from 30 seconds up to 30 minutes.

We considered forwarding notifications with a Pulsar Function and decided against it. Keeping the logic in the control plane meant one less moving part on the broker.

4. The sink that lost events quietly

This one was the most uncomfortable. Large MongoDB documents, up to 16 MB each, pushed Debezium’s JVM, which had roughly a 256 MB heap, out of memory. The Debezium Server 3.0 Pulsar sink handled the failed send by logging it and marking the batch as finished anyway. The offset advanced past events that never reached Pulsar.

Our first fix made it worse in a new way. Raising the heap to 70% of the container meant Netty’s direct memory, which defaults to the heap size, pushed the pod over its memory limit, and Kubernetes killed it mid-snapshot.

The final configuration treats memory as a budget and failure as something to crash on:

# JAVA_OPTS
-XX:MaxRAMPercentage=50.0 -XX:MaxDirectMemorySize=256m -XX:+ExitOnOutOfMemoryError

debezium.source.max.queue.size.in.bytes=134217728   # 128 MiB, bound the queue by bytes
debezium.sink.pulsar.producer.blockIfQueueFull=true  # apply backpressure instead of failing
debezium.sink.pulsar.producer.sendTimeoutMs=0        # never time out a send

The important line is ExitOnOutOfMemoryError. Instead of limping on and losing data, the pod dies, Kubernetes restarts it, and Debezium resumes from its last flushed offset. The out-of-memory case stops losing data and becomes at-least-once delivery instead, and duplicates are something the rest of the pipeline already handles. Each connector runs as a single replica with a Recreate rollout strategy, so a restart never overlaps with a second copy of the same connector.

A related detail: Debezium reads its configuration only at startup. The pod template carries a checksum of the config, so any config change rolls the pod instead of leaving it running on stale settings.

5. Bronze is dumb on purpose

Bronze does as little as possible. It reads each topic with the Spark Pulsar connector as a Structured Streaming job, with trigger(availableNow=True) so Airflow can run it like a batch:

reader = (spark.readStream.format("pulsar")
    .option("topic", table.topic)
    .option("subscriptionName", subscription_name)        # durable, stable name
    .option("predefinedSubscription", subscription_name)
    .option("maxRecordsPerTrigger", "100000")
    .option("startingOffsets", "earliest")                # first run only
    .option("failOnDataLoss", "true"))                    # a gap fails loudly
query = (reader.load().writeStream
    .foreachBatch(append_to_bronze_delta)
    .option("checkpointLocation", table.bronze_checkpoint_path)
    .trigger(availableNow=True)                           # drain, then exit
    .start())

It appends the raw Debezium envelope to Delta with Change Data Feed enabled, and does no deduplication, so delivery is at-least-once. Payloads it can’t parse go to a dead-letter table instead of being dropped. The per-trigger cap started at 1,000 records; after a 50,000-row snapshot produced 50 separate Delta commits, we raised it to 100,000.

6. Ordering without a guard in the MERGE

The silver MERGE is a plain upsert:

cdf_df = (cdf_df.repartition("_id")
                .sortWithinPartitions(F.col("_id"), F.col("debezium_ts").asc()))
# mapInPandas folds each document's events in order -> one final state per _id
(DeltaTable.forPath(spark, silver_path).alias("t")
    .merge(updates_df.alias("s"), "t._id = s._id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

There’s no “only update if newer” condition. Ordering is enforced before the MERGE, by a chain of guarantees:

1. Per-key order in transport. Debezium keys each message by the document’s _id, so all changes to one document land on the same partition, in order.
2. One reader. Bronze reads through an exclusive subscription.
3. One writer. Each bronze table has exactly one writer, and each connector’s DAG runs with max_active_runs=1.
4. One fold per document. Silver sorts each document’s events and folds them one document at a time into a single final state.
5. A cursor that moves only after commit. Silver advances its position only after the MERGE succeeds. A failure replays the same range, which is safe because an upsert is idempotent.

The fold replaced an earlier version that grouped events per Arrow batch. When one document’s events straddled two batches, it produced two partial states for the same document, and the MERGE applied both. Folding one document at a time fixed it.

Deletes are soft: the row gets _is_deleted = true and keeps its last known contents. If that previous state can’t be read back, the events go to a dead-letter table rather than being applied to an empty document.

What I’d tell someone starting out

1. Set Pulsar retention explicitly. With a retention of zero, acknowledged means deleted, and a stalled consumer can't recover.
2. Check subscription expiry before you rely on pre-created subscriptions.
3. Create topics and permissions before the producer starts, and let the sink build full topic names itself.
4. Keep failOnDataLoss on. An error you can see beats an empty read you can't.
5. Make producer failures crash the process, and budget heap and direct memory together.
6. Gate downstream jobs on snapshot completion, with an idempotent, durable trigger.
7. Decide where ordering is enforced, enforce it there, and write the invariants down.
8. Measure end-to-end latency from day one. It's the number everyone asks about, and much easier to add early.

The next problem was further downstream: a silver table holding 0.5 GB of live data took up 47 GB on MinIO and pushed a test node over the edge. Deletion vectors, liquid clustering and a shorter VACUUM window fixed it. That’s the next post.