Azure Event Hubs

Reads events from the Event Hubs namespace through its Kafka endpoint only.

Workers reading in one consumer group share the work, and a source in a different group reads the same events independently. It reads the Azure logs a diagnostic setting streams to an event hub.

Example

Reads the sign-in logs Entra ID streams to the hub insights-logs-signinlogs, with the connection string from the worker-host variable EVENT_HUBS_CONNECTION_STRING.

{
  "id": "entra-logs",
  "type": "eventHubs",
  "namespace": "acme-logs",
  "eventHubs": [
    "insights-logs-signinlogs"
  ],
  "consumerGroup": "smeltr",
  "connectionString": "EVENT_HUBS_CONNECTION_STRING"
}
The Azure Event Hubs form on the Sources page: id entra-logs, namespace acme-logs, event hub insights-logs-signinlogs, consumer group smeltr.
Open complete screenshot ↗
The Azure Event Hubs form on the Sources page: id entra-logs, namespace acme-logs, event hub insights-logs-signinlogs, consumer group smeltr.
Open complete screenshot ↗
The same source in the Sources page's add dialog, before you pick a connection string.

Before you start

  • Use a namespace with a Kafka endpoint. Microsoft recommends the Standard tier for Kafka.
  • To collect Azure's logs, add a diagnostic setting that streams them to the namespace. Each log category goes to an event hub named after it, such as insights-logs-signinlogs, unless the setting names one hub for all of them.
  • Choose a consumer group name for this source. The leader refuses two sources that read one event hub in one consumer group of one namespace.
  • Choose how the source signs in, with exactly one of:
    • a connection string: copy one from a shared access policy on the namespace with the Listen right, and put it in a credential. Use the namespace's, not one that ends in EntityPath=.
    • a service principal: set tenantId, clientId and clientSecret, with the secret in a credential.
    • a user-assigned managed identity: set managedIdentityClientId to its client id. Set it when the host has more than one managed identity.
    • none: the source signs in as the host's managed identity, or as an AKS pod's workload identity. The managed identities of App Service, Container Apps and Functions are not supported.
    Give a service principal or identity the Azure Event Hubs Data Receiver role on the namespace.
  • Allow workers to reach the namespace on port 9093, such as acme-logs.servicebus.windows.net:9093.

Settings

It also takes the settings every source takes.

namespace string required

The Event Hubs namespace: its name in Azure's public cloud, or its host name in another cloud, such as acme-logs.servicebus.usgovcloudapi.net. A host:port names another address, such as the emulator's localhost:9092, which is reached without TLS when it is on this host.

Examples
acme-logsacme-logs.servicebus.usgovcloudapi.netlocalhost:9092

eventHubs list of strings required

The event hubs to read.

Examples
insights-logs-auditeventinsights-logs-signinlogsinsights-logs-auditlogs

consumerGroup string required

The consumer group. Readers in one group share the work and resume from the group's saved positions. The leader refuses two sources that read one event hub in one consumer group of one namespace.

Examples
smeltr$Default

connectionString credential default none

The credential holding the namespace's connection string, with the Listen right. An event hub's own connection string, which ends in EntityPath=, does not work with the Kafka endpoint. Set at most one of connectionString, clientSecret and managedIdentityClientId; with none, the host's managed or workload identity is used.

Examples
EVENTHUBS_CONNECTION_STRING{"secret": "eventhubs-connection-string","version": 1}

tenantId string default none

A service principal's tenant. Set tenantId, clientId and clientSecret together, or none of them.

Example
00000000-0000-0000-0000-000000000000

clientId string default none

A service principal's client id.

Example
11111111-1111-1111-1111-111111111111

clientSecret credential default none

The credential holding a service principal's client secret.

Examples
AZURE_CLIENT_SECRET{"secret": "azure-client-secret","version": 1}

managedIdentityClientId string default the host's managed or workload identity

The client id of the user-assigned managed identity to sign in as. Set it when the worker's host has more than one managed identity. Left out, it signs in as the host's managed identity, or an AKS pod's workload identity.

Example
22222222-2222-2222-2222-222222222222

start position default {"type": "earliest"}

Where this source reads from. Changing it will move offsets to the new position.

earliest
Reads from the oldest record each partition holds.
latest
Reads only what is written after it starts.
timestamp
Reads from the first record at or after millis, in milliseconds since the Unix epoch.

stop position default never

When to stop reading. Absent, never stops.

timestamp
Stops at the first record at or after millis, in milliseconds since the Unix epoch.

format string default each payload is text, in _raw

How each payload becomes the record.

avro
Reads Avro messages using the schema registry. Fields you don't declare are kept.

schemaRegistry object default none

The schema registry that holds your Avro schemas. Set url. If the registry needs a sign-in, also set username and password. Needed for format avro.

Examples
{"url": "https://registry.example.com"}{"url": "https://registry.example.com","username": "smeltr","password": {"secret": "registry","version": 1}}

fields list of fields default none

The Avro fields to read, each with a name and a type such as i64 or str. Name a nested field with dots, such as user.id. Top-level fields you don't list are kept. Only with format avro.

Example
[{"name": "user_id","type": "i64"},{"name": "user.name","type": "str"}]

Advanced

authEndpoint string default its cloud's own

Where a client secret is exchanged for a token. Only with clientSecret.

Examples
https://login.microsoftonline.ushttps://login.chinacloudapi.cn

fetchBytes size default 64 MiB

How much fetched input may wait in memory before it is processed. 4 MiB to 512 MiB.

Examples
128 MiB16 MiB256 MB

encoding string default auto

How a record's bytes become _raw's text: auto, none, or a WHATWG encoding label other than the replacement ones, such as utf-8, utf-16le or shift_jis.

auto
Detects each record's encoding.
utf-8
Decodes every record as UTF-8, as any other label decodes as its own encoding. Bytes that do not decode become U+FFFD.
none
Decodes nothing: the record has no _raw, and its payload is __source.bytes alone. A pipeline reads it as text with decode(__source.bytes, encoding).

Fields it gives

__source.topic holds the event hub's name.

FieldTypeHolds
__source.namestrThe name of the source that read the record.
__source.typestrThe kind of source that read it, its type, such as kafka, s3List, httpApi or hecReceiver.
__source.topicstrThe Kafka topic, or event hub, the record came from. A metrics record carries __smeltr_metrics and a logs record __smeltr_logs.
__source.partitioni64The Kafka or Event Hubs partition the record came from.
__source.offseti64The record's offset in its Kafka or Event Hubs partition, or its record number in its object, from 0.
__source.timedatetimeThe time a sender gave the event, null otherwise.
__source.headersvariantThe Kafka record's headers, each key to its value as text. A key the record repeats holds its last value, and a header with no value is not there. Read one as text with __source.headers.trace as str.
__source.bytesbytesThe record's payload as the source read it, before any decoding. The source decodes _raw from it. A bytes destination writes it unless the destination names another field. Text functions refuse it; decode(__source.bytes, encoding) reads it as text.

Where it starts and stops

start and stop apply to each partition of every event hub the source reads. The group's workers share the partitions, whatever the start and the stop.

  • Start. Where the source reads from: beginning, end, or a specific time.
  • Stop. When to stop reading: never, or a specific time. Once every partition has reached it, the source waits, idle, until it is changed.
  • Progress is saved and restarted sources continue from where they left off.

To read the event hubs as a stream, leave both out. To read between 4 and 8 AM yesterday, start at 4 AM and stop at 8 AM.

Moving a source

To reread or skip events, change start and deploy. Each partition moves there once, then reads on.

  • Setting the same start again moves nothing.
  • Rolling back a deploy does not move the source back.

How it works

The source reads through the namespace's Kafka endpoint, as the Kafka source does, over TLS, signed in with the connection string, the service principal or the host's identity. Smeltr commits the consumer group's progress as it delivers events, so a restarted worker resumes where its group left off.

Event Hubs keeps events only for the hub's retention period, so a group that falls further behind than that loses the oldest.

Verify

Send an event you can recognize, such as with the event hub's Data Explorer in the portal, then find it at the destination.

Troubleshooting

Authentication fails
Check that the namespace has a Kafka endpoint. With a connection string, check that it is the namespace's, not an event hub's, and that its policy has the Listen right. Otherwise, check that the service principal or identity has the Azure Event Hubs Data Receiver role.
Connecting times out
Check that the worker host reaches port 9093 on the namespace, through its firewall and any network rules on the namespace.
Nothing arrives
Check that the diagnostic setting streams to this namespace and that the event hub's metrics show incoming messages.

See also