Azure Event Hubs

Writes each record to the Event Hubs namespace through its Kafka endpoint only.

Example

Writes to the event hub alerts in the namespace acme-logs, with the connection string from the worker-host variable EVENT_HUBS_CONNECTION_STRING.

{
  "id": "alerts-out",
  "type": "eventHubs",
  "namespace": "acme-logs",
  "defaultEventHub": "alerts",
  "connectionString": "EVENT_HUBS_CONNECTION_STRING"
}

Before you start

  • Use a namespace with a Kafka endpoint. Microsoft recommends the Standard tier for Kafka.
  • Create each event hub it writes to.
  • Choose how the destination signs in, with exactly one of:
    • a connection string: copy one from a shared access policy on the namespace with the Send 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 destination 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 Sender role on the namespace.
  • Allow workers to reach the namespace on port 9093, such as acme-logs.servicebus.windows.net:9093.
  • Set compression to gzip only on a namespace that takes it. See Microsoft's Kafka client settings for Event Hubs.

Settings

It also takes the settings every destination 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

defaultEventHub string required

The event hub for a record with no __destination.topic.

Examples
alertsinsights-clean

connectionString credential default none

The credential holding the namespace's connection string, with the Send right. 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

format string default json

What each message holds.

json
Writes the record as JSON.
text
Writes one field's text, from field.
bytes
Writes each source record's bytes as read, before pipeline processing, once per record: __source.bytes unless field names another bytes field.
avro
Writes Avro messages, registering the schema in the schema registry.

schemaRegistry object default none

The schema registry the Avro schema is registered in. 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}}

field string default _raw for text, __source.bytes for bytes

The field each text or bytes record writes. Only with format text or bytes.

Example
message

Advanced: Delivery

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

compression string default none

How it compresses each batch of events. Use gzip only on a namespace that takes it.

none
Sends it uncompressed.
gzip
Compresses it with gzip.

stallAfterSecs integer default 30

Seconds one batch may take to deliver before it counts as stalled. 1 to 120.

Examples
60120

bufferBytes size default 64 MiB

How much undelivered output it may hold. The Kafka client holds as much again. 4 MiB to 1 GiB.

Examples
256 MiB16 MiB1 GiB

Fields it reads

__destination.topic names the event hub a record goes to.

FieldTypeHolds
__destination.topicstrThe Kafka topic or Event Hub a destination writes the record to. Null means the default value for that destination.

How it works

The destination writes through the namespace's Kafka endpoint, as the Kafka destination does, over TLS, signed in with the connection string, the service principal or the host's identity. A batch is acknowledged once Event Hubs has every record in it. A record Event Hubs will not take is refused.

Verify

Send a record you can recognize through the source, then find it in the event hub.

Troubleshooting

Authentication fails
Check that the namespace has a Kafka endpoint. With a connection string, check that it is the namespace's and that its policy has the Send right. Otherwise, check that the service principal or identity has the Azure Event Hubs Data Sender 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.
The destination stalls saying the namespace does not take gzip
The namespace does not take gzip. Set compression to none.

See also