Skip to main content

Azure Event Hub Source Node

The azureeventhubsource node consumes events from an Azure Event Hub. It runs Azure's EventProcessorClient, which balances partitions across all workers in the consumer group and persists partition offsets to a blob checkpoint store, so a restarted pipeline resumes where it left off.

Authentication uses either a SAS connection string or Microsoft Entra ID. The checkpoint store is required — it is what makes partition ownership and offset recovery durable.

Key Features​

  • Automatic partition balancing: the processor client claims and releases partitions cooperatively, so scaling workers up or down redistributes load without configuration changes
  • Durable checkpointing: partition offsets live in an Azure Blob Storage container, so restarts resume from the last committed offset rather than re-reading the hub
  • Backpressure: Azure delivers events by callback; they are queued into a bounded buffer, and a full buffer blocks the callback, slowing Event Hub delivery instead of growing memory
  • Never checkpoints ahead of the pipeline: only offsets for events actually drained and returned downstream are committed
  • Dual authentication: SAS connection string, or Entra ID via an explicit service principal or the ambient DefaultAzureCredential
  • Partition key preserved: an event's Event Hub partition key is carried through as the record key
  • Infinite streaming: never exhausts — runs until the pipeline is terminated

Configuration​

FieldTypeRequiredDefaultDescription
eventHubNameStringYes—Name of the Event Hub (topic) to consume from
consumerGroupStringYes$DefaultConsumer group to read under. Each group gets an independent view of the stream
connectionStringStringConditional—SAS connection string for the namespace or entity. Mutually exclusive with fullyQualifiedNamespace
fullyQualifiedNamespaceStringConditional—Namespace host for Entra ID auth, e.g. my-namespace.servicebus.windows.net. Mutually exclusive with connectionString
tenantIdStringNo—Entra ID service principal tenant. Requires clientId, clientSecret and fullyQualifiedNamespace
clientIdStringNo—Entra ID service principal client id
clientSecretStringNo—Entra ID service principal secret
checkpointContainerNameStringYes—Blob container holding checkpoint and partition-ownership data
checkpointStorageConnectionStringStringConditional—Connection string for the checkpoint storage account. Mutually exclusive with checkpointStorageEndpoint
checkpointStorageEndpointStringConditional—Blob endpoint for the checkpoint storage account, authenticated with Entra ID. Mutually exclusive with checkpointStorageConnectionString
encodingTypeStringYes—Encoding of the event body. See Encoding Types
initialPositionStringNoEARLIESTWhere to start when a partition has no persisted checkpoint: EARLIEST or LATEST. Ignored once a checkpoint exists
maxBufferedEventsintNo1024Bound on events buffered between Azure's callback and the pull loop. A full buffer applies backpressure to Event Hub
maxEventsPerFetchintNo500Maximum events drained from the buffer per fetch cycle
commitStrategyStringNoBATCHCheckpoint strategy: PER_RECORD, BATCH, or NONE
commitBatchSizeintNo1000Records between checkpoints when commitStrategy is BATCH
commitIntervalMslongNo5000Milliseconds between checkpoints when commitStrategy is BATCH

Authentication​

Exactly one Event Hub authentication mode must be configured, and the checkpoint store is authenticated separately.

SAS connection string​

Set connectionString and leave fullyQualifiedNamespace unset. The connection string may target the namespace or a single Event Hub entity.

config:
  connectionString: "Endpoint=sb://my-namespace.servicebus.windows.net/;SharedAccessKeyName=listen;SharedAccessKey=..."
  eventHubName: "telemetry"

Microsoft Entra ID​

Set fullyQualifiedNamespace and leave connectionString unset. Supplying tenantId, clientId and clientSecret uses that service principal; all three are required together. Omit them to fall back to DefaultAzureCredential, which picks up a managed identity, environment variables, or a developer login.

config:
  fullyQualifiedNamespace: "my-namespace.servicebus.windows.net"
  eventHubName: "telemetry"
  tenantId: "00000000-0000-0000-0000-000000000000"
  clientId: "11111111-1111-1111-1111-111111111111"
  clientSecret: "..."

Checkpoint store​

checkpointContainerName is always required. Authenticate the storage account with either checkpointStorageConnectionString or checkpointStorageEndpoint (Entra ID) — providing both is rejected at validation time.

How It Works​

Azure's EventProcessorClient is push-based, while ZephFlow sources are pull-based. The node bridges the two:

  1. The processor client claims a share of the hub's partitions and starts one delivery callback thread per owned partition.
  2. Each callback enqueues its event into a bounded buffer sized by maxBufferedEvents. When the buffer is full the callback blocks, which propagates backpressure to Event Hub rather than accumulating unbounded memory.
  3. The pipeline's polling thread drains up to maxEventsPerFetch events per cycle, deserializes each body using encodingType, and emits the records downstream.
  4. On commit, the offset of the latest drained event per partition is checkpointed to the blob container.

Because only dequeued events are eligible for checkpointing, the stored offset never runs ahead of what the pipeline actually received. On restart, each partition resumes from its checkpoint; initialPosition applies only to partitions that have none.

Commit strategies​

  • BATCH (default): checkpoint after commitBatchSize records or commitIntervalMs milliseconds, whichever comes first. Best throughput.
  • PER_RECORD: checkpoint after every record. Minimizes reprocessing on restart at a significant throughput cost.
  • NONE: never checkpoint. Every restart re-reads from initialPosition.

Delivery is at-least-once: events drained but not yet checkpointed are redelivered after a restart.

DAG Example​

jobContext:
  otherProperties: {}
  metricTags: {}
  dlqConfig:

dag:
  - id: "source"
    commandName: "azureeventhubsource"
    config:
      connectionString: "Endpoint=sb://my-namespace.servicebus.windows.net/;SharedAccessKeyName=listen;SharedAccessKey=..."
      eventHubName: "telemetry"
      consumerGroup: "$Default"
      checkpointStorageConnectionString: "DefaultEndpointsProtocol=https;AccountName=mystorage;AccountKey=..."
      checkpointContainerName: "eventhub-checkpoints"
      encodingType: "JSON_OBJECT"
      initialPosition: "EARLIEST"
      maxBufferedEvents: 1024
      maxEventsPerFetch: 500
      commitStrategy: "BATCH"
      commitBatchSize: 1000
      commitIntervalMs: 5000
    outputs:
      - "sink"

  - id: "sink"
    commandName: "stdout"
    config:
      encodingType: "JSON_OBJECT"

Azure Setup​

  1. Create an Event Hub in an Event Hubs namespace and note its name.
  2. Create a consumer group for this pipeline, or use $Default. Give each independent consumer its own group so they do not compete for the same offsets.
  3. Create a storage account and blob container for checkpoints. Do not share one container across unrelated pipelines consuming the same hub and group.
  4. Grant permissions: the Event Hub identity needs listen rights on the hub (Azure Event Hubs Data Receiver for Entra ID, or a Listen SAS policy); the storage identity needs read/write on the container (Storage Blob Data Contributor for Entra ID).