Run a live smoke test

This guide shows how to prove that Event Hubs messages sent by the included sender arrive in the Databricks target table through EvBricks.

Before you start

You need:

Python 3.13 and uv.
Azure CLI authenticated as an identity that can receive from and send to the test event hub.
A Databricks workspace in a Zerobus-supported region.
A Unity Catalog managed Delta table that matches sql/create_eventhub_events.sql.
A Databricks service principal with USE CATALOG, USE SCHEMA, MODIFY, and SELECT on the target table.
Lakebase Postgres connection settings for the checkpoint tables.

Create local files

Copy the examples:

cp config/evbricks.example.toml config/evbricks.toml
cp .env.example .env

Edit config/evbricks.toml with your non-secret shape:

eventhub_namespace = "eventhub1.servicebus.windows.net"
environment = "development"
region = "eastus2"
client_id = "local-dev"

[control]
target_db = "databricks_postgres"
target_schema = "evbricks"
target_table = "ingestion_status"
backend = "postgres"

[eventhub_defaults]
credential_mode = "azure_cli"
max_wait_time = 60
prefetch_count = 300
starting_position_on_no_checkpoint = "-1"

[event_hubs.EVENTHUBNAME_1]
name = "topic1"
namespace = "eventhub1.servicebus.windows.net"
consumer_group = "$Default"

[databricks_configs.DATABRICKS_1]
catalog = "evbricks_catalog"
schema_name = "default"
table_name = "eventhub_events"
batch_size = 100
record_type = "json"
wait_for_offsets = true

[[mappings]]
event_hub_key = "EVENTHUBNAME_1"
databricks_key = "DATABRICKS_1"
source_id_pattern = "{event_hub}-{env}-{region}-{client_id}"

Edit .env with secrets and runtime endpoints:

DATABRICKS_WORKSPACE_URL=https://<workspace-host>
ZEROBUS_SERVER_ENDPOINT=https://<workspace-id>.zerobus.<region>.azuredatabricks.net
DATABRICKS_CLIENT_ID=<service-principal-client-id>
DATABRICKS_CLIENT_SECRET=<service-principal-secret>

CONTROL_PG_HOST=<lakebase-read-write-host>
CONTROL_PG_PORT=5432
CONTROL_PG_USER=evbricks_runtime
CONTROL_PG_PASSWORD=<lakebase-native-postgres-password>
CONTROL_PG_SSLMODE=require
CONTROL_PG_AUTH_MODE=password

Do not commit either file.

Validate configuration

Install dependencies and validate the resolved config:

uv sync
uv run evbricks validate-config \
  --config-file config/evbricks.toml \
  --env-file .env \
  --require-runtime-secrets

Check the process plan without opening live clients:

uv run evbricks run \
  --config-file config/evbricks.toml \
  --env-file .env \
  --dry-run

Start EvBricks

Open terminal 1 and start the bridge:

uv run evbricks run \
  --config-file config/evbricks.toml \
  --env-file .env

Wait until the console shows that the Zerobus client and Event Hubs receiver started.

Expected startup lines include:

Starting Zerobus client for <catalog>.<schema>.<table>
Control table verified/created: <database>.<schema>.<table>
Successfully obtained token for EventHub scope

Send a small batch

Open terminal 2. Set a unique run ID:

RUN_ID="evbricks-smoke-$(date -u +%Y%m%dT%H%M%SZ)"
START_ID=1000000
COUNT=12

Send messages:

uv run python tools/eventhub_sender/main.py \
  --namespace eventhub1.servicebus.windows.net \
  --eventhub topic1 \
  --count "$COUNT" \
  --start-id "$START_ID" \
  --batch-size 4 \
  --payload "{\"run_id\":\"$RUN_ID\",\"expected_count\":$COUNT}"

Use the namespace and event hub name from your local config.

Watch the EvBricks console

The application should log successful writes:

Zerobus batch ingested: table=<catalog>.<schema>.<table> partition=<id> messages=<n> offset=<offset> total_messages=<n> total_batches=<n> source_id=<source>
Event Hub batch processed: mapping=EVENTHUBNAME_1->DATABRICKS_1 table=<catalog>.<schema>.<table> messages=<n> partitions=[...] total_messages=<n> total_batches=<n>

If the last batch has fewer messages than batch_size, stop EvBricks with ctrl+c after the sender finishes. The shutdown path flushes the pending in-memory batch.

Check Databricks

Run this query against the configured target table. Replace placeholders with your values:

WITH actual AS (
  SELECT
    eventhub_namespace,
    eventhub_name,
    consumer_group,
    partition_id,
    offset,
    sequence_number,
    CAST(get_json_object(body, '$.sequence_id') AS BIGINT) AS payload_sequence_id
  FROM <catalog>.<schema>.eventhub_events
  WHERE body LIKE concat('%', '<run-id>', '%')
),
expected AS (
  SELECT explode(sequence(<start-id>, <start-id> + <count> - 1)) AS payload_sequence_id
),
payload_counts AS (
  SELECT payload_sequence_id, count(*) AS row_count
  FROM actual
  GROUP BY payload_sequence_id
)
SELECT
  (SELECT count(*) FROM actual) AS rows_arrived,
  (SELECT count(*) FROM payload_counts) AS distinct_payload_sequence_ids,
  (
    SELECT count(*)
    FROM expected e
    LEFT ANTI JOIN payload_counts a
      ON e.payload_sequence_id = a.payload_sequence_id
  ) AS missing_payload_sequence_ids,
  (
    SELECT coalesce(sum(row_count - 1), 0)
    FROM payload_counts
    WHERE row_count > 1
  ) AS duplicate_payload_rows,
  count(DISTINCT concat(partition_id, ':', offset)) AS distinct_eventhub_offsets,
  count(*) - count(DISTINCT concat(
    eventhub_namespace,
    '|',
    eventhub_name,
    '|',
    consumer_group,
    '|',
    partition_id,
    '|',
    offset,
    '|',
    sequence_number
  )) AS duplicate_identity_rows
FROM actual;

Expected result:

rows_arrived equals the number sent.
distinct_payload_sequence_ids equals the number sent.
missing_payload_sequence_ids is 0.
duplicate_payload_rows is 0.
distinct_eventhub_offsets equals the number sent.
duplicate_identity_rows is 0.

Stop the bridge

Return to terminal 1 and press ctrl+c. A normal shutdown prints:

Pipeline stopped by user