debezium/dbz#2279 Add Databricks Zerobus Ingest sink for Debezium Server - #294
debezium/dbz#2279 Add Databricks Zerobus Ingest sink for Debezium Server#294wagnercsantos wants to merge 18 commits into
Conversation
|
Cross-linking an architectural note for reviewers: @rk3rn3r's dbz#2300 ("Split This sink is Kafka-API-free on the ingestion path by construction — it implements The one point of overlap worth flagging: |
|
Follow-up: added sink observability to this PR (new commits on the branch), so the sink ships with the JMX metrics operators expect from a Debezium Server sink rather than logs alone. WhatA JMX MBean per route,
An optional The metadata extractors handle both payload shapes the sink sees: the full Debezium envelope ( Design note: the metrics are deliberately self-contained and do not implement the shared Tests & validation
Docs updated in the companion PR (debezium/debezium#7725) with a "Monitoring" section listing the attributes. |
Naros
left a comment
Naros
left a comment
There was a problem hiding this comment.
Hi @wagnercsantos this looks quite good. I've left some comments.
Naros
left a comment
Naros
left a comment
There was a problem hiding this comment.
Hi @wagnercsantos, I found a few more items.
| metrics.recordSkipped(); | ||
| LOGGER.trace("Skipping record for destination '{}' (table={}): {}", record.destination(), table, json); | ||
| } | ||
| record.commit(); |
There was a problem hiding this comment.
Should a record committal happen after the flush?
There was a problem hiding this comment.
Good catch — yes. Moved the record.commit() calls to after the flush: the batch is ingested, the touched streams are flushed to durability, and only then are the offsets committed. Committing before the flush could advance the source offset while records were still buffered, so a crash in between would drop them and break the at-least-once guarantee. Added a test (commitsOffsetsOnlyAfterTheStreamIsFlushed) asserting the ingest -> flush -> commit ordering.
For the REST route this already held: post() is synchronous, so each record is durable (2xx) before its commit() — no separate flush step to reorder there.
Naros
left a comment
Naros
left a comment
There was a problem hiding this comment.
I've left two final points below, if you could take a look.
| private final HttpClient httpClient = HttpClient.newBuilder() | ||
| .connectTimeout(CONNECT_TIMEOUT) | ||
| .build(); |
There was a problem hiding this comment.
In the REST code path, I believe we're creating 3 unique instances of HttpClient. Each one of these creates a thread pool, if I am not mistaken. Would there be value in injecting a single instance here or must there be 3 unique instances?
There was a problem hiding this comment.
Good catch — you're right that each HttpClient brings its own thread pool, and there's no reason for the REST path to have more than one. On the REST route there were two: the consumer's client for the record POSTs, plus one created inside DefaultTokenHttpClient for the token exchange.
Fixed by making DefaultTokenHttpClient accept an existing HttpClient, so ZerobusRestChangeConsumer now builds a single client and shares it with the token provider — one client, one pool, serving both the POSTs and the token exchange. The no-arg constructor is still there for the Kafka-route login handler, which has no client of its own to share.
| * committed past the record, so nothing is dropped silently; set {@code errors.tolerance=all} on | ||
| * the engine if you prefer to skip such records instead. | ||
| */ | ||
| public class StringifyFields<R extends ConnectRecord<R>> implements Transformation<R> { |
There was a problem hiding this comment.
I wonder if there could be value in placing this in debezium-connect-plugins?
There was a problem hiding this comment.
Agreed — there's nothing Zerobus-specific about the mechanism, only the use case (a Delta VARIANT column), so it belongs in the shared transformations. Moved it to io.debezium.transforms.StringifyFields in debezium-connect-plugins, registered in that module's Transformation service file, and dropped it (plus its test and service registration) from the sink module. The javadoc is now phrased generically, mentioning VARIANT and JSON/JSONB columns as examples rather than framing it around Zerobus.
That part lives in a separate PR against debezium since it touches another repo: debezium/debezium#7782. Worth noting the merge order — this PR removes the SMT from the sink, so it should land together with (or after) that one, otherwise there'd be a window where the transformation doesn't exist anywhere. The sink itself doesn't reference the class (it's user-configured via debezium.transforms.*), and debezium-connect-plugins is already a transitive compile dependency of debezium-server-core, so it stays on the classpath either way.
|
Pushed two commits that finish the observability story for this sink. Liveness and connection metrics. Metrics on the Kafka route. That route writes through the generic Since Tested end to end against a real workspace on the Kafka route, matching the counters and One thing found along the way is unrelated to this sink and went to its own issue, |
|
Hi @wagnercsantos and @Naros — I reviewed the current PR head in detail against the ZeroBus sink we have been validating for NVIDIA. I think this PR is the right upstream foundation for our use case: it already provides the native Debezium Server integration, the GA ZeroBus SDK route, REST and Kafka-compatible options, typed JSON normalization, OAuth, SDK recovery controls, flush-before-commit ordering, JMX metrics, and broad tests. Building on this PR would let us contribute the NVIDIA requirements upstream instead of maintaining a competing sink. The important clarification is that this is already a native Debezium Server sink. Our gap is not the runtime; it is the additional raw CDC data and operational contract we need.
The existing flush-before-commit design and typed JSON path should remain intact. I propose adding a mode boundary along these lines: debezium.sink.zerobus.payload.mode=typed|envelope
debezium.sink.zerobus.record.format=json|protobufFor this PR, Would you be open to us contributing these changes directly to your PR branch? I can base signed commits on the current head and open a focused PR into |
|
Heads-up on a regression I introduced in Adding The root cause was the Verified locally with the same two checks the job performs: the default One consequence worth flagging: the SDK bundles per-platform JNI libraries, so the default distribution grows by ~39 MB (498 MB total). That is consistent with the sink jar already shipping there, and |
|
Hi @wagnercsantos and @Naros, I pushed NVIDIA's agreed envelope contribution as It preserves the existing typed JSON path and adds JSON/Protobuf envelopes, deterministic source identity, explicit tombstones, encoded-size validation, sink-native filters, source-order-preserving table groups, and acknowledgement-gated offset commits. Local verification passed the focused envelope suite ( One follow-up before merge: the PR description and documentation PR #7725 still reflect the pre-envelope implementation and should be refreshed with the new configuration and behavior. |
|
@omkarmehta06 — reviewed Protobuf dependency — moved to the BOM in The distribution already resolves For context, the connectors stay on Protobuf 3.25.5:
return changeConsumerHolder.tombstoneSupport()
.map(isSupported -> (CapturingTombstoneEvents) () -> isSupported)
.orElseGet(() -> () -> false); // default: no tombstonesThe sink did not implement Your mapper already handles the case correctly — One request on And one observation from the run. I went looking at the landed rows at my delta table and I did benchmark hashing before raising it, since it isn't free — the digest is an extra pass on top of building the canonical string, not a replacement. But the CPU cost is negligible next to what it saves on the wire and in the table, and this route is bound by round trips rather than local processing, so the trade looks worth making. What a hash costs is readability: the key can currently be read to see the LSN and the operation. Though What the runs covered, all against a live workspace with Protobuf 4.33.2 in the distribution:
One thing that cost me a run and is now in the docs: applying Two design questions still open, both genuine:
On docs: I've left the envelope options for you on debezium#7725, since you know the reasoning behind each one — What I did change there was two sentences of my own that your commit made false — they claimed the sink implemented |
|
@wagnercsantos, thank you for the detailed review and for validating the envelope paths against a live workspace. I reviewed I pushed
Verification passed: For documentation PR #7725, my current contribution scope covers @Naros, once CI completes on this commit, this should be ready for maintainer review. |
|
@omkarmehta06 — re-ran the envelope contract end to end against
One observation, not a problem: with On the documentation: either path works for me, whichever is easier on your side. Send me the envelope option text and the row-contract wording and I'll apply it to #7725 with attribution to you in the commit message. Or, if you can get approval for the From my side this is done: 146 tests green locally, formatter, imports and checkstyle clean, and the envelope validated live in both formats. @Naros, ready for review whenever you have the time. |
|
@wagnercsantos, thank you. I would like to take the direct-contribution path so the documentation commit preserves my authorship and Once access is active, I will start from the branch current head and add one focused signed commit covering the missing envelope options, row contract, operation/filter semantics, and tombstone/filter precedence. I will preserve the existing commit history without rewriting or squashing it. |
|
@omkarmehta06 — done, invite sent for |
|
@kmos should I solve the conflict? |
Yes, thanks! |
|
@kmos done — merged |
|
Hi @wagnercsantos we can't accept PRs that have merge commits. Could you remove the merge commit and instead do a rebase against |
ead0a38 to
b4c8d42
Compare
|
@Naros / @kmos done — rebased against While resolving the rebase I picked up the |
| } | ||
| } | ||
|
|
||
| private void post(String table, String json) throws IOException, InterruptedException { |
There was a problem hiding this comment.
Could you please extract this method in another class like DefaultTokenHttpClient?
There was a problem hiding this comment.
@kmos done in 1dfa2322 — extracted the record-insert HTTP concern into a new ZerobusRestClient, the counterpart to the existing DefaultTokenHttpClient (token exchange). The consumer now holds the batch loop and delegates the request wiring; both share a single HttpClient.
| * This route needs neither the Kafka ingress flag nor a persistent connection, so it suits | ||
| * serverless / edge deployments. Like the other routes it is at-least-once; deduplicate downstream. | ||
| */ | ||
| @Named("zerobusrest") |
There was a problem hiding this comment.
it could be better to use naming zerobus-rest? @Naros @wagnercsantos WDYT?
There was a problem hiding this comment.
I agree, and we already have the precedent in the NATS consumer with nats-streaming and nats-jetstream.
There was a problem hiding this comment.
| import io.debezium.runtime.BatchEvent; | ||
|
|
||
| /** Applies optional sink-native filters before an envelope is routed or serialized. */ | ||
| final class ZerobusEventFilter { |
There was a problem hiding this comment.
If I understand correctly, you want to filter events. Did you try Debezium Filter SMT ? https://debezium.io/documentation/reference/stable/transformations/filtering.html
There was a problem hiding this comment.
@kmos the Filter SMT runs in the transformation chain, ahead of the sink, and needs a scripting engine (Groovy/GraalJS). This filter runs inside the sink's envelope path, where the record is still the full change-event envelope (op, source, headers) — the same reason ExtractNewRecordState can't be applied in envelope mode; an SMT there would alter or strip the envelope the sink is meant to preserve. It's a lightweight regex over the already-parsed event, with no extra dependency. @omkarmehta06 authored this path and can add anything I've missed.
There was a problem hiding this comment.
Interesting. Is there any possibility of making this filter reusable across other sinks? From what I can see in the code, there are some sink-specific rules. Could these be generalized or made more consistent with the Debezium specification?
There was a problem hiding this comment.
@kmos yeah — mechanically most of this is generic. The destination / operation / header / value-pointer regex rules don't know anything about Zerobus; the only sink-specific bits are that it reads ZerobusChangeConsumerConfig and asks the envelope mapper for the operation. So a shared sink-side filter is definitely feasible: lift the rule engine into debezium-server-core (or a shared helper), drive it off generic debezium.sink.filter.* keys, and let each consumer hand it the destination/op/headers/value.
The catch is that it's a cross-cutting change — it lands in server-core and effectively defines a small filtering contract for every Debezium Server sink, plus it overlaps with the Filter SMT you mentioned earlier. That feels like a maintainer's call on whether it belongs in the framework, rather than something to introduce through this sink's PR — I'd rather not stretch #294 into an SPI change on my own.
Proposal: keep the filter self-contained here (it's tested and localized) and take "generalize into a shared server-side filter" as a follow-up. @omkarmehta06 wrote this path, so he should weigh in first. Omkar — if you're swamped, just throw it my way and I'll pick it up in a separate PR, no worries; otherwise it's all yours. And @Naros — would a shared sink-side filter in server-core be something you'd want, or would you rather sinks lean on the Filter SMT for this?
| } | ||
| } | ||
|
|
||
| private void writeJsonEnvelopeGroup(String table, List<ZerobusEnvelope> records) throws Exception { |
There was a problem hiding this comment.
Could you please extract all the serialization logic into a different class using a strategy pattern and initialize only one based on cofiguration (protobuf/Json)
There was a problem hiding this comment.
@kmos good suggestion. There are already two serializer classes (ZerobusJsonEnvelopeSerializer / ZerobusProtobufEnvelopeSerializer); the current dispatch is a single if on record.format. A full strategy pattern would need to be generic over the payload type — JSON produces String, Protobuf byte[], and that split runs down to two separately-typed stream maps — so it's a mid-size refactor of the core write path, and the gain would be mostly stylistic since there's no functional change. Would it work to take it as a follow-up PR so this one stays focused? @omkarmehta06 owns this path — deferring to you on whether you'd prefer to fold it in here or handle it separately.
There was a problem hiding this comment.
@kmos done in 04449078. Extracted the serialization into a ZerobusEnvelopeSerializer<P> strategy: it serializes the envelope to its payload type, reports the encoded size, and opens the stream that accepts that encoding — the three things that varied by format, bound together by P. A factory instantiates only the one the configured record.format selects, so we no longer build both.
With that, the consumer holds a single envelopeSerializer and a single stream map instead of one field and one map per format, and the batch write path (ingestEnvelopeGroup<P>) is generic — serialization, size check, ingest, eviction and close no longer branch on JSON vs Protobuf. The one format decision left is the single create() at startup, which is unavoidable since the payload types (String vs byte[]) diverge into differently-typed streams.
No functional change. I re-validated end-to-end against a live workspace on both record.format=json and record.format=protobuf — snapshot read plus live create/update/delete/tombstone landing correctly in Delta.
@omkarmehta06 — heads-up, this touches the envelope serialization path you wrote. It's purely structural (nothing changes at runtime), and I mainly did it to get this thread cleared so the PR can move to review sooner rather than sit waiting. Totally fine to tweak it or take it over if you'd rather — just didn't want to leave your code changed without flagging it to you.
Naros
left a comment
Naros
left a comment
There was a problem hiding this comment.
Hi @wagnercsantos I found just one more item, not sure you think it's important.
New debezium-server-databricks-zerobus module: a sink that writes change events directly into Databricks Zerobus Ingest (managed Delta tables in Unity Catalog), with no intermediate broker. Delivery is at-least-once; the source LSN/offset is carried for downstream deduplication. Routes: - zerobus - gRPC via the Databricks Zerobus Java SDK (GA transport, primary); - zerobusrest - Zerobus REST API, one HTTP request per record; - a Kafka OAUTHBEARER login handler so the built-in kafka sink can authenticate against the Zerobus Kafka-compatible endpoint. ZerobusTypeSystem maps Connect/Debezium logical types to the Delta column contract (decimal -> exact string, temporals -> epoch, bytes -> base64, JSON/JSONB -> VARIANT, arrays -> ARRAY<T>), avoiding the "invalid type" class of ingestion errors without source-side *.handling.mode tuning. A small StringifyFields SMT covers the VARIANT-per-field case. The Zerobus SDK (a JNI wrapper shipping native libraries) is declared optional, so it is not pulled into the default distribution; deployments opt in for the gRPC route. Tests (59): unit (type encoding, RFC 9396 authorization_details, routing, tombstone-skip, SMT), REST route (mocked HttpClient), gRPC route (ZerobusJsonStream mocked with Mockito), and a credentials-gated end-to-end IT. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
Expose sink observability over JMX, per delivery route, under the object name debezium.zerobus:type=connector-metrics,context=sink,server=<grpc|rest>, task=0 — following the same convention as the JDBC sink connector and Debezium source connectors. Attributes cover throughput (TotalRecordsIngested, split by operation into TotalInserts/Updates/Deletes/Reads; TotalRecordsSkipped; TotalErrors; TotalFlushes), end-to-end freshness (MilliSecondsBehindSource = now - source.ts_ms of the last ingested event), flush latency (Last/MaxFlushDurationMillis), ActiveStreams and Route. An optional metrics.log.interval also mirrors the values to a periodic INFO log line (default 0 = off; JMX is always on). The operation / source-timestamp extractors handle both the full Debezium envelope (op, nested source.ts_ms) and an unwrapped record (the recommended ExtractNewRecordState / ExtractNewDocumentState setup, flattened __op / __source_ts_ms); missing fields degrade gracefully. The metrics are self-contained and do not implement the shared io.debezium.sink.spi.SinkProgressListener, mirroring how ZerobusTypeSystem owns its mapping rather than depending on io.debezium.sink.*, keeping the sink free of the debezium-sink module dependency. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
- Flush only the streams touched in a batch rather than every open stream, avoiding needless round-trips when few of N tables change per batch. - Extract the Databricks OAuth token exchange into a shared, stateless ZerobusTokenExchange used by both the REST route (ZerobusTokenProvider) and the Kafka OAUTHBEARER login handler, so the request shape and response parsing cannot drift; promote the TokenHttpClient seam to a top-level type. - Drop an unrelated pom.xml reordering (mysql/mariadb) and an unrelated debezium-quarkus-db2 assembly entry from this PR; the latter will be raised separately (candidate for a 3.6 backport). Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
…ter-flush - Define the OAuth client secret as ConfigDef.Type.PASSWORD in both the gRPC and REST consumer configs so it is masked rather than exposed as a plain string. - Commit the source offsets only after the touched streams are flushed to durability. Committing before the flush could advance the offset while records were still buffered, dropping them on a crash and breaking at-least-once. The REST route already held this (its POST is synchronous). Add a test asserting the ingest -> flush -> commit ordering. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
…ingifyFields upstream Let DefaultTokenHttpClient accept an existing HttpClient, so that the REST consumer builds a single client and shares it with the token provider. One client, and therefore one thread pool, now serves both the record POSTs and the OAuth token exchange. The no-argument constructor remains for the Kafka-route login handler, which has no client of its own to share. Remove the StringifyFields transformation, its test and its service registration. The transformation carries nothing sink-specific, so it moves to debezium-connect-plugins in debezium/debezium#7782. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
…sink MilliSecondsBehindSource freezes at its last value when the pipeline stops, so on its own it cannot distinguish an idle source from a stalled sink. Add MilliSecondsSinceLastEvent, which reports how long the sink has been silent and keeps growing while nothing is forwarded, mirroring the attribute of the same name on the source connectors. Add Connected as well, so that an operator can tell whether the sink holds a usable connection, again mirroring the source connectors. It is set when the consumer connects and cleared when it closes. Also cover the counters with a concurrency test: the class is annotated @threadsafe and the counters are updated from more than one thread, but nothing asserted that until now. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
The Kafka route writes through the generic kafka sink, so the Zerobus sink is never invoked and cannot instrument itself: the metrics that the gRPC and REST routes expose are simply absent there. A producer interceptor is the extension point the Kafka producer already provides, so this needs no change to the kafka sink. Registering it reports the same attributes under server=kafka, alongside the other two routes. Two attributes carry a route-specific meaning. TotalFlushes counts broker-acknowledged records, which is this route's durability boundary, and TotalRecordsSkipped stays 0, because the generic sink forwards every record it receives. Because onSend runs for every produced record, the metadata extraction stays off the allocation path. The operation and the source timestamp are read from the record headers when ExtractNewRecordState is configured with add.headers, and otherwise from the serialized value by a single-pass scan that builds no document tree. Latency is sampled rather than measured on every record: System.nanoTime() is an order of magnitude more expensive when the producer's network thread reads the clock concurrently, which otherwise dominates the cost of the interceptor. The first record is always timed, so a low-volume pipeline still reports a duration, and all counters remain exact. Add unit tests for the header and value paths, including nested and quoted keys that must not be mistaken for top-level ones, malformed JSON, the sampling boundary and concurrent sends, plus an integration test against a real workspace that is gated on credentials being present in the environment, like the existing gRPC one. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
The sink configured only maxInflightRecords and left the SDK's recovery settings untouched, so a stream that failed with a retriable error stopped the connector even though the SDK can recover it. Recovery is safe here because the SDK tracks which records a stream has not acknowledged and re-sends only those, so enabling it cannot duplicate records that already landed, and the sink still does not retry the batch itself. Expose recovery, recovery.retries, recovery.backoff.ms, recovery.timeout.ms and flush.timeout.ms. None of them declares a default: the value reaches the SDK only when it is set, so an unset option keeps whatever the SDK version in use defines instead of being pinned here, where it could drift across upgrades. Configuration.getInteger and getBoolean parse the raw value and throw on an absent one, so the fields are read through hasKey to keep null meaningful. An error that recovery cannot resolve still propagates, so the offset is not committed past it and the at-least-once guarantee is unchanged. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
Fixes four problems found reviewing the sink against a production workload: Records were counted as ingested when ingestRecordOffset() returned, which is before the flush that makes them durable, so a batch whose flush failed still advanced the counters. handle() now accumulates the operation and source timestamp of each enqueued record and counts them only once the flush succeeds. The per-table stream map was unbounded, so a source capturing many tables held one connection per table with no limit. The map is now access-ordered and bounded by max.open.streams (default 100, 0 keeps every stream open); above the limit the least recently used streams are flushed and closed, and reopened on demand. The sink declares zerobus-ingest-sdk as optional, so it was not resolved transitively into a distribution built with the sink-databricks-zerobus profile and the gRPC route failed at startup with NoClassDefFoundError. The profile now requests the SDK explicitly, with the version managed in debezium-server-bom. The sink offered no way to supply a pre-configured client, unlike the other sinks. Adds the @CustomConsumerBuilder injection point for ZerobusSdk. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
Groundwork for the raw-envelope contribution discussed on the issue, so that it can be added rather than refactored in. The SDK's ZerobusJsonStream and ZerobusProtoStream share only a package-private base class, so no SDK type names both, and the consumer's stream map was typed as ZerobusJsonStream. Introduces ZerobusStreamHandle<P> (ingest/flush/close) with a JSON adapter, so the batch handling no longer depends on how a record is encoded. Mocking this interface instead of the SDK class also keeps the consumer tests off the SDK's native loader, so they no longer skip where no native library is available. Declares payload.mode=typed|envelope and record.format=json|protobuf, defaulting to typed and json so current behaviour is unchanged. Two combinations are refused: typed with protobuf, which is a contradiction because the typed path writes JSON, and anything other than typed with json, which is reserved but not implemented - accepting it would silently write typed rows for a deployment that asked for an envelope. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
The SDK was declared optional, which left it out of a distribution that still contained this module. That was tolerable while the sink only referenced the type from its own fields, because the JVM links those lazily and connect() never runs unless the Zerobus sink is selected. Injecting Instance<ZerobusSdk> changed that: the type becomes part of the bean signature, so Quarkus emits a Class.forName for it in the generated bean constructor, which runs during Arc.initialize() before any sink is chosen. On a distribution without the SDK that throws ClassNotFoundException and the server fails to start with any sink, as the redis distribution check showed. Drops the optional flag, making the SDK a regular dependency as in every other sink module, and leaves inclusion to the existing sink.databricks-zerobus.scope property. The individual profile no longer has to request the SDK explicitly. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
Add JSON and Protobuf envelope serialization for the Debezium Server Zerobus sink while preserving typed ingestion. Include deterministic idempotency metadata, sink-native filtering, record-size validation, ordered grouping, and durability-gated offset commits. Add focused tests and example configuration for the new contract. Signed-off-by: Omkar Mehta <omehta@nvidia.com>
The Protobuf runtime is a real requirement of the Protobuf envelope path: the Zerobus SDK exposes Protobuf types on its public API (DescriptorProto, Message, Parser) but declares none of them at compile scope, so a module that opens a Protobuf stream has to bring the runtime itself. It was declared with an inline version in the sink module, which no other module in this repository does, and it pinned 4.33.0 while the distribution already resolves 4.33.2 transitively through debezium-server-qdrant. The sink therefore compiled against one version and ran against another. Managing it in debezium-server-bom at 4.33.2 removes that gap and makes the version deliberate rather than inherited by accident. Also drops the version.zerobus.sdk property that the sink module still carried. It has been unused since the SDK version moved to the BOM, and two sources for one version only invite drift. The connectors keep Protobuf 3.25.5, which is a separate concern: debezium-connector-postgres and debezium-connector-oracle generate code from .proto files with protoc pinned to that same property, two of them proto2, so moving them is not a version bump. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
…rite them The envelope payload mode documents tombstone.handling.mode=event, which writes a distinct operation=tombstone envelope, but the mode could never produce one: TombstoneSupportProducer defaults to withholding tombstones unless a consumer reports the capability, and this sink did not implement tombstoneSupport(). An end-to-end run with tombstones.on.delete=true and the event handling mode landed the delete and no tombstone at all. Reports the capability only when the envelope mode is configured to write tombstones, so the typed and REST paths keep receiving none; they discard a null payload in isJsonObject anyway, and the default handling mode must not leak across payload modes. The configuration is read directly rather than from the field the @PostConstruct populates, because the engine queries this capability while deciding what to capture and that is not ordered against bean initialisation. Verified end to end: the tombstone now lands with operation=tombstone, its key preserved and a null value. Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
Preserve Debezium operation codes including truncate and message, accept long aliases in sink filters, and compact the deterministic identity with SHA-256. Document and test the at-least-once table-group replay boundary, remove redundant envelope initialization from the batch hot path, and exercise the consumer's tombstone capability directly. Signed-off-by: Omkar Mehta <omehta@nvidia.com>
…he route to zerobus-rest Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
…uched Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
b4c8d42 to
f701386
Compare
Naros
left a comment
Naros
left a comment
There was a problem hiding this comment.
I think this LGTM from my PoV, just waiting on CI.
|
@Naros, if you agree, I can open a new issue to work with the serialization logic as we mentioned before |
…ategy interface Introduces ZerobusEnvelopeSerializer<P>, a per-format strategy that binds the envelope encoding to the stream that accepts it: it serializes the envelope to its payload type P, reports the encoded size, and opens a stream for that encoding. A factory instantiates only the serializer the configured record.format selects, rather than one of each. The JSON and Protobuf serializers implement it (String and byte[] respectively), and the change consumer now holds a single envelopeSerializer and a single wildcard stream map instead of one field and one map per format. The batch write path is generic over P, so serialization, size checking, ingestion, eviction and close no longer branch on the format; the only remaining format decision is the one-time factory selection. Behaviour is unchanged. Addresses the review request to extract the serialization logic into a strategy chosen by configuration. Verified end-to-end against a live workspace on both record.format=json and record.format=protobuf (snapshot read plus live create/update/delete/tombstone). Signed-off-by: Wagner Santos <wagner.santos@databricks.com>
4 participants
Adds a new Debezium Server sink that writes change events directly into Databricks Zerobus Ingest → managed Delta tables in Unity Catalog, with no intermediate broker.
Fixes debezium/dbz#2279
Fixes debezium/dbz#2378
What
The new module (
debezium-server-databricks-zerobus) ships three delivery routes and a schema-driven type mapping:zerobus— gRPC via the Databricks Zerobus Java SDK (GA transport, primary route);zerobusrest— Zerobus REST API, one HTTP request per record (no persistent connection);kafkasink authenticate against the Zerobus Kafka-compatible endpoint;ZerobusTypeSystem— maps Connect/Debezium logical types to the Delta column contract (decimal→exact string, temporals→epoch, bytes→base64, JSON/JSONB→VARIANT, arrays→ARRAY<T>), the per-dialect "type system" equivalent for this sink; plus a smallStringifyFieldsSMT for the VARIANT-per-field case.Delivery is at-least-once; the sink carries the source LSN/offset for downstream dedup.
Note on scope
When @Naros and I discussed this on debezium/dbz#2279, the plan was to land the gRPC route first and follow up with REST and the Kafka handler separately. Everything ended up ready in time, so this PR brings the complete sink as one coherent unit. Happy to split it into the phased gRPC-first sequence if you'd prefer to review it that way — the routes are cleanly separated and easy to peel apart; I brought them together only because the sink tells a more complete story whole.
SDK dependency
The gRPC route depends on
com.databricks:zerobus-ingest-sdk, a JNI wrapper over a native client that ships per-platform native libraries. It is declared<optional>true</optional>, so it is not propagated into the default Debezium Server distribution — the module compiles and unit-tests against it, but deployments opt in by adding the jar for the gRPC route. Thezerobusrestroute and the type system need nothing from the SDK.How it was validated
End-to-end against managed Delta tables, exercising snapshot + insert/update/delete:
DECIMAL), temporals (DATE/TIMESTAMP/TIMESTAMP_NTZ, tz→string), binary/base64→BINARY, arrays→ARRAY<T>, JSON/JSONB→VARIANT, CLOB/BLOB.Tests
59 tests, green (
mvn -pl debezium-server-databricks-zerobus test):ZerobusTypeSystem), OAuthauthorization_details(RFC 9396 — scoped grants, neverALL_PRIVILEGES), topic→table routing / qualified-name handling, tombstone-skip, and theStringifyFieldsSMT.HttpClientasserting request path, headers and JSON body, plus tombstone-skip and HTTP-error handling.ZerobusJsonStreamis mocked with Mockito (mirroring how the Zerobus SDK tests itself), covering ingest+flush, skip, and fail-fast; the live gRPC IT is gated behind@EnabledIfEnvironmentVariableso it is skipped in the default CI run and executed manually with credentials.Docs
The operations-guide section for this sink is submitted separately, against the docs in the main repo — debezium/debezium#7725 (config properties, managed-table prerequisites + RFC 9396 grants, table routing, and the data type mapping table). A module-level
application.propertiesexample is included in this PR.Related platform fixes (separate PRs, required to run this sink with some sources)
Building/validating this sink surfaced a few pre-existing platform bugs that block certain sources on 3.7; each is contributed separately and linked here for context:
initial_only. Fix: debezium/dbz#2305 Skip archive-log destination validation for snapshot-only Oracle runs debezium#7724.cc @Naros