Moving Databricks streams from the Event Hubs Spark connector to the Kafka endpoint: auth, offsets and a no-gap cutover

Many older Azure Databricks streams read Event Hubs through the com.microsoft.azure:azure-eventhubs-spark Maven library. You can't install Maven libraries on serverless, they need allowlisting on standard access mode, and the connector carries its own connection-string secrets. The built-in Kafka source talks to the Event Hubs Kafka endpoint, works with Unity Catalog service credentials and needs no library. This runbook moves one stream at a time without losing or doubling events.

Caution

The two sources use different checkpoint formats. You cannot point the Kafka reader at the old checkpoint. Every migrated stream gets a new checkpoint and an explicit starting position. Get that wrong and you either replay retention-days of data or skip events.

#Prerequisites

• An Event Hubs namespace on Standard tier or above. Basic tier has no Kafka endpoint.
• Each event hub is a Kafka topic with the same name. Its partitions are Kafka partitions.
• The sink must tolerate a short overlap: a MERGE on an event key, or append to a table you deduplicate downstream. If the current sink is a blind append with no usable key, fix that first.

#Step 1: Inventory the old stream

Record these for each stream before you touch it:

q = [s for s in spark.streams.active if s.name == "telemetry_bronze"][0]
print(q.lastProgress["sources"][0]["endOffset"])   # per-partition seqNo/offset
print(q.lastProgress["sink"])

Note the consumer group, maxEventsPerTrigger, the trigger interval, any startingPositions, and which of the old columns downstream code uses.

#Step 2: Set up keyless auth

Create a Unity Catalog service credential backed by an access connector (Catalog Explorer > External data > Credentials > Create credential > Service credential). Then give the connector's managed identity read access on the namespace or the single event hub:

PRINCIPAL_ID=$(az databricks access-connector show -g rg-dbx -n ac-dbx-prod \
  --query identity.principalId -o tsv)

az role assignment create --assignee-object-id "$PRINCIPAL_ID" \
  --assignee-principal-type ServicePrincipal \
  --role "Azure Event Hubs Data Receiver" \
  --scope $(az eventhubs eventhub show -g rg-ingest --namespace-name evhns-prod -n telemetry --query id -o tsv)

Grant the job's run-as principal use of the credential:

GRANT ACCESS ON SERVICE CREDENTIAL evh_reader TO `sp-telemetry-jobs`;

If you're on a runtime below 16.1, use OAuth with a client ID and secret instead (14.3 LTS+ on standard access mode). The JAAS config must use the kafkashaded. class names. Unshaded names fail with RESTRICTED_STREAMING_OPTION_PERMISSION_ENFORCED.

#Step 3: Map the columns

Downstream code written against the Event Hubs connector's columns will break. Map them once in the new reader so the bronze schema stays the same:

#Step 4: Build the new reader against a test sink

from pyspark.sql import functions as F

opts = {
    "kafka.bootstrap.servers": "evhns-prod.servicebus.windows.net:9093",
    "subscribe": "telemetry",
    "databricks.serviceCredential": "evh_reader",
    "includeHeaders": "true",
    "maxOffsetsPerTrigger": "500000",
    "failOnDataLoss": "true",
}

raw = spark.readStream.format("kafka").options(**opts).load()

bronze = raw.select(
    F.col("value").alias("body"),
    F.col("partition").cast("string").alias("partition"),
    F.col("offset").alias("sequenceNumber"),
    F.col("timestamp").alias("enqueuedTime"),
    F.map_from_entries(
        F.transform("headers", lambda h: F.struct(h.key, h.value.cast("string")))
    ).alias("properties"),
)

With service credentials, don't set kafka.security.protocol, kafka.sasl.mechanism or kafka.sasl.jaas.config. Databricks sets them. Run this against a scratch table with startingOffsets = latest and a throwaway checkpoint for an hour. Compare per-minute counts with the old stream's output.

Tip

maxOffsetsPerTrigger is a total across all partitions, like the old maxEventsPerTrigger. Start with the same number. The Kafka source plans one task per partition by default; set minPartitions only if a few hot partitions leave cores idle.

#Step 5: Pick the restart position

You have two reliable options. Both deliberately overlap a little.

Option A, timestamp (simplest). Stop the old stream, take the latest enqueuedTime it wrote, subtract a safety margin (5-10 minutes), and start from there:

cutover_ms = spark.sql("""
  SELECT unix_millis(max(enqueuedTime) - INTERVAL 10 MINUTES)
  FROM prod.bronze.telemetry
""").first()[0]

opts["startingTimestamp"] = str(cutover_ms)

startingTimestamp applies to all partitions. Use startingOffsetsByTimestamp for per-partition values. Test timestamp lookup against your namespace in non-prod first. If it's rejected, use Option B.

Option B, explicit offsets. Build startingOffsets JSON from the highest sequenceNumber per partition already in bronze. First confirm on a sample that the Kafka offset for a known event equals its Event Hubs sequence number:

import json
rows = spark.sql("""
  SELECT partition, max(sequenceNumber) AS seq
  FROM prod.bronze.telemetry
  WHERE enqueuedTime > now() - INTERVAL 2 DAYS
  GROUP BY partition
""").collect()
opts["startingOffsets"] = json.dumps(
    {"telemetry": {str(r.partition): int(r.seq) - 1000 for r in rows}})

Subtracting a margin gives you the overlap. Either way, the overlap is handled by the sink.

#Step 6: Dedupe the overlap in the sink

If bronze is append-only, write through foreachBatch with a MERGE on (partition, sequenceNumber), or on a business event ID if the producer sets one:

from delta.tables import DeltaTable

def upsert(batch, batch_id):
    batch = batch.dropDuplicates(["partition", "sequenceNumber"])
    (DeltaTable.forName(spark, "prod.bronze.telemetry").alias("t")
       .merge(batch.alias("s"),
              "t.partition = s.partition AND t.sequenceNumber = s.sequenceNumber "
              "AND t.enqueuedTime >= current_timestamp() - INTERVAL 1 DAY")
       .whenNotMatchedInsertAll()
       .execute())

(bronze.writeStream
   .foreachBatch(upsert)
   .option("checkpointLocation", "/Volumes/prod/ops/checkpoints/telemetry_bronze_kafka/")
   .queryName("telemetry_bronze_kafka")
   .start())

The time predicate keeps the MERGE from scanning all of bronze. After the overlap window has passed and the migration is stable, you can switch back to a plain append.

#Step 7: Cut over

1. Stop the old query: q.stop(), or cancel the job run. Confirm it's gone from spark.streams.active.
2. Pause the old job's schedule so a retry can't restart it.
3. Deploy the new job with a new checkpointLocation, for example /Volumes/prod/ops/checkpoints/telemetry_bronze_kafka/.
4. Start it and watch the first batches: numInputRows, inputRowsPerSecond vs processedRowsPerSecond, and per-partition lag in sources[0].metrics if your runtime reports it.

#Step 8: Validate, then remove the library

SELECT partition, count(*) AS n,
       max(sequenceNumber) - min(sequenceNumber) + 1 AS expected
FROM prod.bronze.telemetry
WHERE enqueuedTime BETWEEN '2026-11-13T08:00:00' AND '2026-11-13T10:00:00'
GROUP BY partition
HAVING n <> expected;

No rows means no gaps or duplicates across the cutover. Sequence numbers are contiguous within a partition. Then remove the azure-eventhubs-spark library and the connection-string secret from the job and the secret scope.

Warning

Keep the old checkpoint and the old job definition (paused) until validation passes. Rolling back means stopping the new stream and restarting the old one from its old checkpoint, which resumes exactly where it stopped.

#Prevent it next time

• Use service credentials for all new Event Hubs streams so there are no connection strings to rotate.
• Give every stream a sink-side idempotency key. It makes migrations, replays and checkpoint resets safe.
• Keep Event Hubs retention well above your longest expected outage plus migration time.
• Alert on consumer lag per stream, not only on job failures.