Event Hubs

Microsoft Azure

Synopsis

Creates a collector that connects to Azure Event Hubs and consumes messages from specified event hubs. Reads over AMQP, with checkpoints in a blob container, or through Event Hubs' Kafka endpoint, with no storage account. Supports multiple authentication methods, TLS encryption, and multiple workers for high-throughput scenarios.

Schema

- id: <numeric>
  name: <string>
  description: <string>
  type: eventhubs
  tags: <string[]>
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    event_hub: <string>
    client_connection_string: <string>
    tenant_id: <string>
    client_id: <string>
    client_secret: <string>
    namespace: <string>
    consumer_group: <string>
    protocol: <string>
    start_position: <string>
    container_connection_string: <string>
    container_url: <string>
    container_name: <string>
    timeout: <numeric>
    batch_size: <numeric>
    reuse: <boolean>
    workers: <numeric>
    event_split_rules: <string[]>
    tls:
      status: <boolean>
      cert_name: <string>
      key_name: <string>
      passphrase: <string>
      ca_name: <string>
      server_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>
      insecure_skip_verify: <boolean>

Configuration

The following fields are used to define the device:

Device

Field	Required	Default	Description
id	Y		Unique identifier
name	Y		Device name
description	N	-	Optional description
type	Y		Must be eventhubs
tags	N	-	Optional tags
pipelines	N	-	Optional pre-processor pipelines
status	N	true	Enable/disable the device

Connection

Event Hubs supports three authentication methods:

Method 1: Connection String Authentication

Field	Required	Default	Description
client_connection_string	Y*		Event Hubs connection string (required for connection string authentication)
event_hub	Y		Event hub name to consume from

Method 2: Service Principal Authentication

Field	Required	Default	Description
tenant_id	Y*		Azure tenant ID (required for service principal authentication)
client_id	Y*		Azure service principal client ID
client_secret	Y*		Azure service principal client secret
namespace	Y*		Event Hubs namespace (required if not using connection string)
event_hub	Y		Event hub name to consume from

Method 3: Managed Identity Authentication

With Managed identity switched on in the console, the director signs in with the Azure default credential, which is the managed identity of the Azure resource it runs on. Leave tenant_id, client_id, client_secret and client_connection_string unset.

Field	Required	Default	Description
namespace	Y*		Fully qualified Event Hubs namespace. The director needs it whenever there is no connection string
event_hub	Y		Event hub name to consume from
container_url	Y*		Checkpoint storage container URL. Required for protocol: amqp only (see Storage Configuration)

The identity needs the same Azure RBAC roles as a service principal (see IAM Permissions).

Consumer Configuration

Field	Required	Default	Description
consumer_group	N	"$Default"	Consumer group name

Reading

Field	Required	Default	Description
protocol	N	amqp	What the device reads the event hub through: amqp or kafka (see Kafka Endpoint Mode). Any other value is a configuration error
start_position	N	latest	Where a partition with no saved position starts, in both protocols: latest or earliest (see Start Position). Any other value is a configuration error

protocol	Reads through	Tier	Checkpoint storage	Splits partitions between directors
absent or amqp (default)	AMQP	Every tier	Required: container_url, or container_connection_string with container_name	Ownership blobs in the container
kafka	<namespace host>:9093	Standard and above, not Basic	Not used; container_* keys are ignored	Event Hubs (Kafka consumer group protocol)

Nothing changes for an existing device: with no protocol, it reads over AMQP.

Storage Configuration

With protocol: amqp, the default, EventHubs requires checkpoint storage, reached with the same sign-in as the event hub: Method 1 with connection string authentication, Method 2 with service principal or managed identity authentication. The container must exist beforehand; nothing creates it. With protocol: kafka, Event Hubs keeps the consumer group's offsets itself, and no checkpoint storage is used.

Sign-in	Checkpoint storage fields
Connection string (client_connection_string)	container_connection_string and container_name
Service principal (tenant_id, client_id, client_secret, namespace)	container_url
Managed identity (namespace, no client credentials)	container_url

The director takes the connection string path only when both client_connection_string and container_connection_string are set. Any other combination builds the checkpoint container client from container_url with the device's Azure credential, so a mixed pair fails when the device starts. A service principal or managed identity with a container connection string has no container URL to reach. A connection string with a container URL has no namespace to consume from.

In the console, the Checkpoint storage group on the Advanced settings step is shown, and required, only when Read through is AMQP. It shows Container URL for a managed identity or a service principal, and Container name with Connection string for a connection string. A device saved earlier with a mixed pair opens asking for the missing field, and saving it removes the stored fields of the other method.

Method 1: Storage Account Connection String

Field	Required	Default	Description
container_connection_string	Y*		Azure Storage connection string
container_name	Y*		Blob container name for checkpoints

Method 2: Storage Account URL

Field	Required	Default	Description
container_url	Y*		Azure Storage container URL

* = Conditionally required (see authentication and storage methods above). The storage method follows the sign-in and is not a separate choice. With protocol: kafka, none of these fields is required.

Performance

Field	Required	Default	Description
timeout	N	15	Seconds a single receive may wait for events before the cycle ends and a new one begins
batch_size	N	10000	Maximum events requested per receive
reuse	N	true	Caps workers at the number of CPU cores the host offers
workers	N	4	Number of consumers to start, whether reuse is on or off. With reuse on, fewer start when the host has fewer CPU cores. 1 runs a single consumer
max_buffer_size	N	128MB	Ingest buffer admission cap. Also sizes this listener's ingest buffers: 3/8 of this value at start, growing to at most 3x

note

max_buffer_size is the ingest-buffer admission cap, and is distinct from any buffer_size field above, which sizes the network read buffer. Left unset it is 128MB, whose buffers take 48 MiB of heap per listener worker at start and grow to at most 384 MiB as bursts need the room — and with reuse: true a device runs one worker per listener. Size strings are binary, and 32MB and 32MiB are exact synonyms.

See Listener Memory Sizing for the arithmetic, the sizing table, and what happens when the cap is exhausted.

Event Split Rules

Field	Required	Default	Description
event_split_rules	N	-	Ordered list of rule set keys that split each message into separate logs, such as azure_records for a {"records":[...]} batch. See Event Split Rules

Azure sends Diagnostic Settings, Activity Log, Entra ID and Defender XDR data to Event Hubs in batches, many logs to a message. Without rules, the device keeps each message whole, as one log, unless its split_mode splits it. With them, each record becomes a log of its own as the device receives it, in both protocols, and every later count, the device's own included, counts the logs. A message is still acknowledged only after all of its logs are delivered. A condition reads the event hub as source. Changing the list restarts the device; editing a rule set it uses does not.

TLS

The connection to Event Hubs is always encrypted, in both protocols. tls.status: true only adds a client certificate and a CA to it. With protocol: kafka, tls.insecure_skip_verify is refused: the device's credential would go to a broker whose certificate nobody checked, so the device reports a configuration error.

In the console, the TLS switch is labelled Client certificate: when enabled, the director also presents a client certificate (mTLS); when disabled, the connection is still encrypted, with no client certificate. The form has no CA field, so tls.ca_name can be set only in the device's configuration. Skip certificate verification is hidden when Read through is Kafka.

Field	Required	Default	Description
tls.status	N	false	Enable TLS encryption
tls.cert_name	Y*	cert.pem	Client certificate for mutual TLS
tls.key_name	Y*	key.pem	Client private key for mutual TLS
tls.passphrase	N	-	Passphrase for an encrypted private key
tls.ca_name	N	-	CA bundle used to verify the server certificate. When unset, the host trust store is used; when set, it replaces the host trust store rather than adding to it.
tls.server_name	N	-	SNI hostname override for the TLS handshake
tls.min_tls_version	N	tls1.2	Minimum accepted TLS version (tls1.0, tls1.1, tls1.2, tls1.3)
tls.max_tls_version	N	-	Maximum accepted TLS version. When unset, the highest mutually supported version is negotiated.
tls.insecure_skip_verify	N	false	Skip server certificate verification. Use only for testing.

* = Required when tls.status is true.

note

TLS material fields (cert_name, key_name, ca_name, client_ca_name) accept any of the following:

• File name — resolved relative to the service root directory. Nested paths such as certs/prod/server.pem are supported.
• Absolute path — honored only if it resolves inside the service root. Any path that escapes the root is refused.
• Inline PEM content — used verbatim when the value contains -----BEGIN.
• Environment variable — ${ENV_VAR}.
• Vault reference — $secret{id=...} or $secret{store=...,ref=...}.

Details

IAM Permissions

When using service principal or managed identity authentication, the following Azure RBAC roles are required:

Azure Role	Scope	Purpose	protocol
Azure Event Hubs Data Receiver	Event Hubs Namespace or Event Hub	Consume events and read hub properties	amqp
Storage Blob Data Contributor	Storage Account or Container	Read, write, and list checkpoint blobs	amqp
Azure Event Hubs Data Receiver	Event Hubs Namespace, Event Hub or consumer group	Join the consumer group, fetch, and commit offsets. The only role Kafka mode needs	kafka

With protocol: amqp, the checkpoint storage requires Contributor (not just Reader) because the device writes checkpoint state and manages ownership blobs for partition load balancing. With protocol: kafka, no storage role is needed.

When using connection string authentication, Azure RBAC roles are not needed for Event Hubs access. The Shared Access Policy embedded in the connection string governs access (typically Listen claim for consumers); in Kafka mode, the Listen claim is enough. With protocol: amqp, checkpoint storage is then reached with the storage account connection string in container_connection_string.

Kafka Endpoint Mode

With protocol: kafka, the device reads through Event Hubs' Kafka endpoint, on port 9093 of the namespace host, instead of over AMQP. Over AMQP, the device keeps each partition's reading position and partition ownership in a blob container, which must exist beforehand and needs Storage Blob Data Contributor. Through the Kafka endpoint, Event Hubs keeps the consumer group's offsets itself, so the device needs no storage account at all. The Kafka endpoint is available on the Standard tier and above, not on Basic.

The mode applies to the Event Hubs device only. No other device type changes, the Kafka device included.

Sign-in:

• client_connection_string is used as SASL PLAIN, with the user $ConnectionString and the connection string as the password. The broker comes from its Endpoint.
• A service principal (tenant_id, client_id, client_secret) or a managed identity signs in with a Microsoft Entra token for https://<namespace host>/.default. The broker is namespace plus port 9093.

The event hub must exist. Event Hubs does not create it on demand, and answers a missing one the same way it answers a missing role. The consumer group does not: through the Kafka endpoint it is a Kafka consumer group, which Event Hubs creates on first use, apart from the event hub's own consumer groups and not shown in the Azure portal.

Connection test. The device runs a connection test before it reports Connected:

• AMQP: it reads the event hub's properties and lists the checkpoint container. A missing container, a wrong URL or a missing storage role fails the test, not the started device.
• Kafka: it signs in, then asks for the event hub and the consumer group's coordinator. It never joins the group, so readers already in the group are not disturbed. A failure carries a hint naming the tier, the role, and that the event hub and the consumer group must exist.

Changing protocol on an existing device does not carry the reading position over. AMQP keeps it in the blob container, Kafka mode in Event Hubs. The device starts again from start_position, which means duplicates or a gap. For security data, plan the cutover, and keep the AMQP container until rollback is no longer needed.

One consumer group per application. Whichever protocol the device uses, every director replica runs it, and the partitions are split between the replicas. Over AMQP, the positions and partition ownership are in the blob container; through the Kafka endpoint, Event Hubs keeps them for the consumer group. An AMQP consumer group and a Kafka consumer group of the same name are different groups, so the director over AMQP and another application over Kafka do not compete for partitions. Kafka consumer groups span the namespace, though: several Kafka-mode devices or applications sharing one group name, for example all left at $Default, are rebalanced together whenever one of them is. Give each application its own consumer group.

In the console, the Advanced settings step has a Reading group above Checkpoint storage:

• Read through: two tiles, AMQP and Kafka. A new device starts on AMQP, and a device saved without protocol opens as AMQP.
• Start position: Latest or Earliest.

On a saved device's page, switching Read through shows the warning "The saved reading position does not carry over". Switching a saved device that signs in with a connection string to Kafka clears its stored container connection string when the device is saved.

Start Position

start_position sets where a partition with no saved position starts. It applies to both protocols.

start_position	Result
absent or latest (default)	Events sent after the device starts
earliest	Everything the event hub still retains: up to 1 day on Basic, 7 on Standard, 90 on Premium. This can be a large backlog
anything else	Refused (configuration error)

It applies only when there is no saved position. Once the device has read, it resumes from the saved position whatever start_position says.

An AMQP device on a geo-replicated namespace with no start_position used to start new partitions at the earliest event, the Azure SDK's default there. It now starts at the latest. The console has always shown Latest for such a device.

note

Over AMQP, the device claims every free partition at once. It used to claim one partition about every 10 seconds, and a partition with no checkpoint started at the latest event when it was claimed, so a new device on a 32-partition event hub took about 5 minutes to read all partitions and skipped what reached the last ones meanwhile. A failed replica's partitions are now also taken over together once their ownership expires, rather than one at a time.

Multiple Workers

The collector starts workers workers whether reuse is on or off. Each worker maintains its own Event Hubs consumer and processes messages independently, automatically balancing message volumes. reuse only bounds the count: on, it caps workers at the host's CPU cores, so fewer consumers can run than workers names. Set workers: 1 to run a single consumer.

In the console, a new device starts with Multi-worker mode enabled and Workers (Optional) at 4, the same settings a device that sets neither key runs with. A saved device without reuse opens with Multi-worker mode enabled, as Director runs it, and one saved with reuse: false keeps that value.

Messages

The collector supports reading over AMQP, with checkpoints in a blob container, or through the Kafka endpoint, with offsets kept by Event Hubs, consumer group load balancing, multiple Event Hub subscriptions, TLS-encrypted connections, connection string, service principal and managed identity authentication, and custom message-processing pipelines.

Examples

Basic with Connection String

Creating a simple EventHubs consumer with connection string...

- id: 1 name: basic_eventhubs type: eventhubs properties: client_connection_string: "Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=mykey;SharedAccessKey=myvalue" event_hub: "logs" container_connection_string: "DefaultEndpointsProtocol=https;AccountName=mystorage;AccountKey=mykey" container_name: "checkpoints"

Service Principal Authentication

Connecting with service principal authentication...

- id: 2 name: sp_eventhubs type: eventhubs properties: tenant_id: "12345678-1234-1234-1234-123456789012" client_id: "87654321-4321-4321-4321-210987654321" client_secret: "${AZURE_CLIENT_SECRET}" namespace: "mynamespace" event_hub: "security-logs" consumer_group: "datastream-group" container_url: "https://mystorage.blob.core.windows.net/checkpoints"

Kafka Endpoint Mode

Reading through the Kafka endpoint with a managed identity, from the earliest retained event, with no checkpoint storage...

- id: 7 name: kafka_eventhubs type: eventhubs properties: protocol: "kafka" start_position: "earliest" namespace: "mynamespace.servicebus.windows.net" event_hub: "security-logs" consumer_group: "datastream-group"

High-Volume Processing

Optimizing for throughput with multiple workers...

- id: 3 name: performant_eventhubs type: eventhubs properties: client_connection_string: "${EVENTHUBS_CONNECTION_STRING}" event_hub: "high-volume-logs" consumer_group: "processing-group" container_connection_string: "${STORAGE_CONNECTION_STRING}" container_name: "checkpoints" reuse: true workers: 8

Secure Connection

Secure EventHubs connection with TLS...

- id: 4 name: secure_eventhubs type: eventhubs properties: tenant_id: "${AZURE_TENANT_ID}" client_id: "${AZURE_CLIENT_ID}" client_secret: "${AZURE_CLIENT_SECRET}" namespace: "secure-namespace" event_hub: "secure-logs" consumer_group: "secure-group" container_url: "${STORAGE_CONTAINER_URL}" tls: status: true cert_name: "eventhubs.crt" key_name: "eventhubs.key"

Pipeline Processing

Applying custom processing to EventHubs messages...

- id: 5 name: pipeline_eventhubs type: eventhubs pipelines: - json_parser - field_extractor - normalize_timestamps properties: client_connection_string: "${EVENTHUBS_CONNECTION_STRING}" event_hub: "application-logs" consumer_group: "processing-group" container_connection_string: "${STORAGE_CONNECTION_STRING}" container_name: "checkpoints"

Multiple Consumer Groups

Configuring consumer groups for load distribution...

- id: 6 name: distributed_eventhubs type: eventhubs properties: client_connection_string: "${EVENTHUBS_CONNECTION_STRING}" event_hub: "distributed-logs" consumer_group: "instance-1" container_connection_string: "${STORAGE_CONNECTION_STRING}" container_name: "checkpoints-instance1" reuse: true workers: 4