Use Kafka-compatible APIs with Zerobus Ingest

Important

This feature is in Beta and is available on AWS and Azure only.

Zerobus Ingest offers Kafka-compatible producer APIs that let you ingest using any Apache Kafka producer client without a Databricks SDK. You point an existing Kafka producer at the Zerobus endpoint and produce to a topic named after your target table, and the records land in a Unity Catalog Delta table. The Kafka-compatible APIs are a good fit when you already have a Kafka producer, a collector that speaks the Kafka protocol, or tooling that emits to Kafka, and you want to route that data into Delta with minimal code changes.

The Kafka-compatible APIs run over TCP with SASL_SSL and the OAUTHBEARER mechanism and implement the producer-side subset of the Kafka protocol, which covers Produce, Metadata, ApiVersions, and the SASL handshake APIs. Consumer, admin, and transactional APIs are not available. The APIs are write-only.

When to use the Kafka-compatible APIs

The Kafka-compatible APIs are the best fit in the following scenarios:

  • You want to send data to Delta without adopting a Zerobus SDK, and you already run a Kafka producer or an application, agent, or collector that emits to Kafka.
  • You want to reuse your existing Kafka producer configuration, batching, and operational tooling.
  • You send JSON records and don't need Protocol Buffers or Apache Arrow.

If you are building a new client from scratch and want the highest throughput, per-record acknowledgements, and automatic recovery, use a Zerobus SDK over gRPC instead of the Kafka-compatible APIs. See Choose an interface. For columnar or batched workloads, see Use Arrow Flight with Zerobus Ingest.

How the ingestion model works

The Kafka-compatible APIs map Kafka concepts onto Zerobus Ingest as follows:

  • Topics. A Kafka topic name is the full three-level Unity Catalog table name (catalog.schema.table). The target table must already exist, because Zerobus never creates topics.
  • Records. Zerobus ingests only the record value, which must be a UTF-8 encoded JSON object that matches the target Delta table schema. It ignores record keys, headers, and the client-supplied partition and timestamp, and doesn't persist them.
  • Authentication. Each connection authenticates with a Databricks OAuth token scoped to the target table, presented through SASL/OAUTHBEARER. See Authentication.
  • Acknowledgements. Zerobus returns a Produce response only after it durably persists the records. Configure your producer with acks=all.

Zerobus is partitionless by design. The Zerobus endpoint is a single logical broker and partition, so every record for a topic is acknowledged on partition 0. Producers don't need to account for this. Zerobus Ingest scales horizontally to handle incoming load.

Also, Zerobus Ingest provides at-least-once delivery. A single connection sustains roughly 50,000 messages per second, which is lower than the SDK gRPC path. For the highest throughput, use a Zerobus SDK instead of the Kafka-compatible APIs. Latency, quota, record-size, and partitioned-table limits are shared with the rest of Zerobus Ingest. See Zerobus Ingest connector quotas.

Authentication

The Kafka-compatible APIs use SASL/OAUTHBEARER. The bearer token is a Databricks OAuth token that you obtain with the client credentials of a service principal with access to the target table. The token is scoped to that table through OAuth authorization_details and uses the zerobusDirectWriteApi resource, the same flow as the Zerobus REST API.

OAuth tokens expire after one hour, so supply the token through your Kafka client's token-provider callback rather than as a static string. The client then re-fetches a fresh token whenever it reconnects. Connections also have a bounded server-side lifetime. When a connection reaches that limit, Zerobus closes it, and the producer reconnects and re-authenticates on its own. Re-authentication on a live connection is not supported.

Grant the service principal the required Unity Catalog privileges on the target table before you connect. See Create a service principal and grant permissions.

Write a client

The example below produces to the same air_quality table used in the Use the Zerobus Ingest connector examples. It uses kafka-python, but any Kafka producer client that supports SASL_SSL with the OAUTHBEARER mechanism works. Adapt the token-provider pattern to your client library.

The producer connects to the Zerobus bootstrap server on port 9092. Find your workspace ID and region as described in Get your workspace URL and Zerobus Ingest endpoint.

  • Bootstrap server: <workspace-id>.zerobus.<region>.azuredatabricks.net:9092
pip install kafka-python requests

Step 1: Build a token provider

Zerobus authenticates each connection with a short-lived, table-scoped Databricks OAuth token. Because the token expires, pass a callback that mints a fresh token on demand rather than a static token.

The fetch_zerobus_token() function exchanges your service principal credentials for a token scoped to the target table, and ZerobusTokenProvider wraps it in the callback interface kafka-python expects.

import json

import requests
from kafka.sasl.oauth import AbstractTokenProvider

# See "Get your workspace URL and Zerobus Ingest endpoint" in zerobus-ingest.md.
WORKSPACE_ID = "1234567890123456"
WORKSPACE_URL = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
TABLE_NAME = "main.default.air_quality"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"


def fetch_zerobus_token():
    catalog, schema, table = TABLE_NAME.split(".")
    authorization_details = [
        {
            "type": "unity_catalog_privileges",
            "privileges": ["USE CATALOG"],
            "object_type": "CATALOG",
            "object_full_path": catalog,
        },
        {
            "type": "unity_catalog_privileges",
            "privileges": ["USE SCHEMA"],
            "object_type": "SCHEMA",
            "object_full_path": f"{catalog}.{schema}",
        },
        {
            "type": "unity_catalog_privileges",
            "privileges": ["SELECT", "MODIFY"],
            "object_type": "TABLE",
            "object_full_path": TABLE_NAME,
        },
    ]

    response = requests.post(
        f"{WORKSPACE_URL}/oidc/v1/token",
        auth=(CLIENT_ID, CLIENT_SECRET),
        data={
            "grant_type": "client_credentials",
            "scope": "all-apis",
            "resource": f"api://databricks/workspaces/{WORKSPACE_ID}/zerobusDirectWriteApi",
            "authorization_details": json.dumps(authorization_details),
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["access_token"]


# kafka-python calls token() whenever it needs a fresh OAuth token.
class ZerobusTokenProvider(AbstractTokenProvider):
    def token(self):
        return fetch_zerobus_token()

Step 2: Configure the producer and send records

Point the producer at the bootstrap server, configure SASL_SSL with the OAUTHBEARER mechanism, and pass the token provider from Step 1. Use acks="all" so each batch is acknowledged only after it is durably persisted, and send records uncompressed.

from kafka import KafkaProducer

BOOTSTRAP_SERVERS = "1234567890123456.zerobus.us-west-2.cloud.databricks.com:9092"

producer = KafkaProducer(
    bootstrap_servers=BOOTSTRAP_SERVERS,
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    sasl_oauth_token_provider=ZerobusTokenProvider(),
    # Wait for durable acknowledgement before treating a record as ingested.
    acks="all",
    # Compression is not supported by the endpoint; send records uncompressed.
    compression_type=None,
)

# Each send() returns a future immediately. The topic is the full table name.
futures = [
    producer.send(
        topic=TABLE_NAME,
        value=json.dumps(
            {"device_name": f"sensor-{i}", "temp": 20 + i % 15, "humidity": 50 + i % 40}
        ).encode("utf-8"),
    )
    for i in range(1000)
]

producer.flush()

# Block on each future to confirm every record was durably acknowledged.
for future in futures:
    future.get(timeout=30)

producer.close()
print("All records ingested successfully")

Configuration options

The Kafka-compatible APIs implement the producer subset of the Kafka protocol. Configure your producer according to the following options.

Option Details
Record format JSON only. Each record value must be a UTF-8 encoded JSON object that matches the target table schema. To send Protocol Buffers or Avro, use a Zerobus SDK.
Compression Not supported. Send uncompressed batches, for example compression.type=none. Zerobus rejects gzip, snappy, lz4, and zstd batches with the UNSUPPORTED_COMPRESSION_TYPE error code.
Record fields Value only. Zerobus ingests the record value and doesn't persist keys, headers, partition assignments, or timestamps.
API support Write-only. Zerobus accepts Produce requests plus the metadata and SASL requests needed to establish a session. Consumer, admin, and transactional APIs are not supported.
Front-end Private Link Not supported. Connect over the public endpoint instead.
Schema Enforced. Zerobus rejects records with fields that don't match the target table schema, and treats extra nullable Delta columns as a non-breaking change. To capture unmatched fields instead of rejecting them, configure a rescue column.

For more information about front-end Private Link, see Azure Private Link concepts.

For latency, quota, record-size, and partitioned-table limits, see Zerobus Ingest connector quotas.

Best practices

Follow these guidelines to get the best performance and reliability from the Kafka-compatible APIs. A single connection can sustain roughly 50,000 messages per second.

  • Reuse a long-lived producer across many records rather than creating one per batch, because producer creation and the SASL handshake carry setup cost.
  • Let the producer accumulate records into batches, for example by tuning linger.ms and batch.size, instead of flushing after every record. Batching is the single biggest lever for throughput.
  • Use acks=all to get a durable acknowledgement for each batch, matching Zerobus at-least-once semantics.
  • Fetch OAuth tokens through your client's token-provider callback so they refresh automatically on reconnect, rather than passing a static token that expires.
  • Run the producer in the same cloud region as the Zerobus endpoint for maximum throughput.

Error handling

Zerobus reports failures using standard Kafka error codes on the affected topic and partition. Common codes include:

Kafka error Meaning
SASL_AUTHENTICATION_FAILED The OAuth token is missing, invalid, or lacks the required Unity Catalog privileges on the table.
UNKNOWN_TOPIC_OR_PARTITION The target table doesn't exist, has been dropped, or the token isn't authorized to write to it.
INVALID_RECORD A record failed schema validation or could not be decoded as UTF-8 JSON.
MESSAGE_TOO_LARGE A single record exceeds the 10 MB record-size limit. See Record size.
UNSUPPORTED_COMPRESSION_TYPE The batch was compressed. Send records uncompressed.

After a failed Produce request, Zerobus returns the per-partition error codes and closes the connection. Kafka producers reconnect automatically, but design your client to surface send failures, for example by inspecting the result of each send, so records aren't silently dropped.

Additional resources

  • Last updated on