Databricks Zerobus Sink Node

The zerobussink node in ZephFlow streams processed records directly into a Databricks Unity Catalog managed Delta table over the Zerobus gRPC ingest API, without a SQL warehouse or a staging volume.

Overview

The Zerobus sink opens a single long-lived gRPC stream to a Databricks Zerobus ingest endpoint and pushes records onto it as they flow through the pipeline. Unlike the Databricks Sink, which stages Parquet files on a Databricks Volume and loads them with a COPY INTO on a SQL Warehouse, Zerobus writes straight into the table. There is no warehouse to keep running and no staging volume, so neither warehouseId nor volumePath exists on this config.

Key Features

Direct Streaming Ingest: Writes to a Unity Catalog managed table over gRPC, with no SQL warehouse or staging volume
Protobuf or JSON Encoding: Send records as protobuf built from an Avro schema, or as JSON
Ordered, Durable Delivery: Each flush blocks until Zerobus confirms the batch is durable
Store and Forward: Optionally buffer records to a local durable queue during an outage and replay them on recovery
Automatic Reconnect: A failed flush marks the stream unhealthy and the next flush rebuilds it

Parameters

Parameter	Type	Description
`databricksCredentialId`	`String`	Key used to look up the Databricks credential in the job context
`zerobusEndpoint`	`String`	gRPC ingest endpoint, e.g. `https://<id>.zerobus.<region>.cloud.databricks.com`
`tableName`	`String`	Fully-qualified Unity Catalog table name: `catalog.schema.table`
`avroSchema`	`Map<String, Object>`	Avro schema of the target table, used to build the protobuf descriptor
`encodingType`	`String`	Wire encoding: `protobuf` (default) or `json`

Set these fields in the config of a node with commandName: "zerobussink" — see DAG Examples.

Config Object

The full config object for the Zerobus sink node (ZerobusSinkDto.Config):

Field	Type	Required	Default	Description
`databricksCredentialId`	`String`	Yes	—	Key used to look up the `DatabricksCredential` in the job context. The credential must exist, or config validation fails.
`zerobusEndpoint`	`String`	Yes	—	Zerobus gRPC ingest endpoint. Must be a well-formed `https` URL with no surrounding whitespace.
`tableName`	`String`	Yes	—	Fully-qualified Unity Catalog table name as `catalog.schema.table`, with no surrounding whitespace.
`avroSchema`	`Map<String, Object>`	Only for `protobuf`	—	Avro record schema of the target table. Required and validated in `protobuf` mode; neither required nor used in `json` mode.
`encodingType`	`String`	No	`protobuf`	Wire encoding: `protobuf` or `json` (matched case-insensitively). An explicitly empty value fails validation.
`storeAndForwardEnabled`	`boolean`	No	`false`	Buffer records to a local durable queue on a connectivity failure and replay them once the connection recovers.
`localStorePath`	`String`	No	`null`	Directory for the local store-and-forward queue. Defaults to a temp directory keyed by node id.
`localStoreMaxBytes`	`long`	No	`1073741824` (1 GiB)	Hard cap on local store size. Once reached, incoming records are dropped. Must be positive when store-and-forward is enabled.
`forwardRetryIntervalMillis`	`long`	No	`5000`	How often the background forwarder retries draining the local queue during an outage. Must be positive when store-and-forward is enabled.
`ackTimeoutMillis`	`int`	No	`20000`	Max time to wait for a server ack before treating the connection as down. Deliberately lower than the SDK default of 60s.
`flushTimeoutMillis`	`int`	No	`60000`	Max time a flush blocks waiting for queued records to be acknowledged. Deliberately lower than the SDK default of 5 minutes.

Endpoint and Table Validation

Both values are passed to the Databricks SDK exactly as written, so they are validated exactly as the SDK will receive them rather than being silently trimmed:

zerobusEndpoint must be an https URL with a host, and must not have leading or trailing whitespace. The host shape is deliberately not constrained, so private-link and custom-DNS endpoints are accepted.
tableName must be three non-blank dot-separated parts. Backtick-quoted identifiers that themselves contain dots are not supported by this check.

Encoding

`protobuf` (default)

The node parses avroSchema, converts it to a protobuf descriptor, and opens a proto stream. Each record is converted to a DynamicMessage against that descriptor and sent as bytes.

The Avro schema must be a record. Field types map as follows:

Avro type	Protobuf type
`string`, `enum`	`string`
`int`	`int32`
`long`	`int64`
`float`	`float`
`double`	`double`
`boolean`	`bool`
`bytes`, `fixed`	`bytes` (including `decimal` logical type)
`record`	nested message
`array`, `map`	repeated / map field

Nullable fields are expressed as a union with null. A union of more than one non-null type is rejected. date logical types accept a string value.

avroSchema is validated at config time: an unparseable schema, or one that cannot be converted to a protobuf descriptor, fails the node before it starts.

`json`

The node opens a JSON stream and serializes each record with Jackson. The stream is created without a descriptor, so avroSchema is neither required nor used in this mode.

Delivery Model

The sink pushes every record of a flush batch onto the stream with ingestRecordsOffset, then blocks on the last offset of that batch with waitForOffset. Zerobus guarantees ordered, durable delivery per stream, so once the last offset is confirmed every earlier offset on the same stream is durable too.

Delivery is at-least-once. On reconnect the SDK re-sends unacknowledged records, so downstream consumers should dedupe if they need exactly-once.
Flush batches are capped at 10,000 records. Larger upstream batches are partitioned into chunks of this size, and each chunk is fully drained before the next is ingested. This bounds both heap use (every encoded payload of a chunk is held in memory) and in-flight records against the SDK's 50,000-record window.
A per-record encoding failure is isolated. Every payload is encoded before the stream is touched, so a record that fails to encode becomes an error output for that record only and the rest of the batch still commits. This is where a record/schema mismatch surfaces.
An ingest or durability failure fails the node. If the batch has already been handed to the SDK and the ingest or the durability wait then fails, the commit state is unknown — the records may or may not have been written. The node raises an unknown-commit-state error rather than reporting the batch as clean per-record failures, which would silently drop records that may actually be committed.

The stream is not thread-safe, so all stream access is serialized. A failed flush marks the stream unhealthy, and the next flush closes and rebuilds it before sending.

Store and Forward

Setting storeAndForwardEnabled to true makes a connectivity failure buffer records to a local durable queue instead of dropping them. A background forwarder replays them through the normal encode-and-flush path once the connection recovers, reconnecting the stream as needed.

A failure is treated as transient connectivity — and therefore buffered — when the cause chain contains any of:

an IOException or an internal retriable-connection error
a gRPC status of UNAVAILABLE, DEADLINE_EXCEEDED, or RESOURCE_EXHAUSTED
any ZerobusException other than NonRetriableException

NonRetriableException is the SDK's authoritative "permanent" marker (schema, auth, or other non-recoverable errors) and is never buffered. Anything unrecognized is also treated as permanent.

info

When the connection is down, the SDK's native calls surface a ZerobusException carrying only a message — gRPC status objects do not cross the JNI boundary. That is why any non-permanent ZerobusException counts as transient.

The node emits these counters:

Metric	Meaning
`store_forward_buffered_count`	Records written to the local store during an outage
`store_forward_replayed_count`	Records successfully replayed after recovery
`store_forward_dropped_count`	Records dropped because the local store was full

When to Use

Choose Zerobus Sink for continuous, low-latency streaming into a Unity Catalog managed table when you do not want to run a SQL Warehouse or manage a staging volume.
Choose Databricks Sink when you need the COPY INTO staging path — for example to use COPY INTO options, or to write to a table that is not a Zerobus target.
Choose Delta Lake Sink for a raw landing zone on storage that is not coupled to a Databricks workspace.

DAG Examples

Basic Usage (protobuf)

jobContext:
  otherProperties:
    my-databricks-credential:
      host: "https://my-workspace.cloud.databricks.com"
      clientId: "my-service-principal-client-id"
      clientSecret: "my-service-principal-secret"
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "broker:9092"
      topic: "device-events"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "zerobus_sink"

  - id: "zerobus_sink"
    commandName: "zerobussink"
    config:
      databricksCredentialId: "my-databricks-credential"
      zerobusEndpoint: "https://1234567890.zerobus.us-west-2.cloud.databricks.com"
      tableName: "prod.telemetry.device_events"
      encodingType: "protobuf"
      avroSchema:
        type: "record"
        name: "DeviceEvent"
        fields:
          - name: "device_id"
            type: "string"
          - name: "reading"
            type: "double"
          - name: "recorded_at"
            type: "long"
    outputs: []

JSON Encoding

In json mode the stream is opened without a protobuf descriptor, so no Avro schema is needed:

jobContext:
  otherProperties:
    my-databricks-credential:
      host: "https://my-workspace.cloud.databricks.com"
      clientId: "my-service-principal-client-id"
      clientSecret: "my-service-principal-secret"
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "broker:9092"
      topic: "device-events"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "zerobus_sink"

  - id: "zerobus_sink"
    commandName: "zerobussink"
    config:
      databricksCredentialId: "my-databricks-credential"
      zerobusEndpoint: "https://1234567890.zerobus.us-west-2.cloud.databricks.com"
      tableName: "prod.telemetry.device_events"
      encodingType: "json"
    outputs: []

With Filtering and Transformation

jobContext:
  otherProperties:
    my-databricks-credential:
      host: "https://my-workspace.cloud.databricks.com"
      clientId: "my-service-principal-client-id"
      clientSecret: "my-service-principal-secret"
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "broker:9092"
      topic: "raw-events"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "ok_only"

  - id: "ok_only"
    commandName: "filter"
    config:
      expression: '$.status == "ok"'
    outputs:
      - "to_table_row"

  - id: "to_table_row"
    commandName: "eval"
    config:
      expression: "dict(device_id=$.id, reading=$.value, recorded_at=$.ts)"
    outputs:
      - "zerobus_sink"

  - id: "zerobus_sink"
    commandName: "zerobussink"
    config:
      databricksCredentialId: "my-databricks-credential"
      zerobusEndpoint: "https://1234567890.zerobus.us-west-2.cloud.databricks.com"
      tableName: "prod.telemetry.device_events"
      encodingType: "protobuf"
      avroSchema:
        type: "record"
        name: "DeviceEvent"
        fields:
          - name: "device_id"
            type: "string"
          - name: "reading"
            type: "double"
          - name: "recorded_at"
            type: "long"
    outputs: []

With Store and Forward

jobContext:
  otherProperties:
    my-databricks-credential:
      host: "https://my-workspace.cloud.databricks.com"
      clientId: "my-service-principal-client-id"
      clientSecret: "my-service-principal-secret"
  metricTags: {}

dag:
  - id: "kafka_source"
    commandName: "kafkasource"
    config:
      broker: "broker:9092"
      topic: "device-events"
      groupId: "consumer-group"
      encodingType: "JSON_OBJECT"
    outputs:
      - "zerobus_sink"

  - id: "zerobus_sink"
    commandName: "zerobussink"
    config:
      databricksCredentialId: "my-databricks-credential"
      zerobusEndpoint: "https://1234567890.zerobus.us-west-2.cloud.databricks.com"
      tableName: "prod.telemetry.device_events"
      encodingType: "json"
      storeAndForwardEnabled: true
      localStorePath: "/var/lib/zephflow/zerobus-queue"
      localStoreMaxBytes: 2147483648   # 2 GiB
      forwardRetryIntervalMillis: 5000
      ackTimeoutMillis: 20000
      flushTimeoutMillis: 60000
    outputs: []

Common Issues and Solutions

Configuration Errors

All configuration problems are collected and reported together as a single Zerobus sink configuration errors: ... message. Common causes:

databricksCredentialId is missing, or names a credential that is not present in the job context
zerobusEndpoint is not an https URL, or has leading/trailing whitespace
tableName is not catalog.schema.table, or has a blank part or surrounding whitespace
encodingType is neither protobuf nor json
avroSchema is missing, unparseable, or cannot be converted to a protobuf descriptor (protobuf mode only)

Records Rejected at Encode Time

A record that does not match the Avro schema fails to encode and is reported as an error output for that record. The first such failure in a batch is logged at warn with the reason; the rest are logged at debug and the total appears in the flush summary. Check for fields missing from the schema, or values whose type cannot be converted.

Stream Fails to Open

If the stream cannot be opened — bad or unsupported Avro schema, descriptor build failure, auth failure, or the open itself — the node logs the endpoint, table, and encoding along with the cause, then fails. Without store-and-forward the node will crash-loop on init until the cause is fixed.

Connectivity Outages

Enable storeAndForwardEnabled so a transient outage buffers records instead of failing them. The default ackTimeoutMillis (20s) and flushTimeoutMillis (60s) are set well below the SDK defaults so an outage is detected in seconds rather than minutes.

databricksSink: Load data into Unity Catalog tables via Parquet staging and COPY INTO
deltalakeSink: Write data to Delta Lake tables on cloud storage
kafkaSink: Write data to Kafka topics

Overview​

Key Features​

Parameters​

Config Object​

Endpoint and Table Validation​

Encoding​

protobuf (default)​

json​

Delivery Model​

Store and Forward​

When to Use​

DAG Examples​

Basic Usage (protobuf)​

JSON Encoding​

With Filtering and Transformation​

With Store and Forward​

Common Issues and Solutions​

Configuration Errors​

Records Rejected at Encode Time​

Stream Fails to Open​

Connectivity Outages​

Related Nodes​