# 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](https://learn.microsoft.com/azure/event-hubs/azure-event-hubs-apache-kafka-overview). Microsoft [recommends the Standard tier](https://learn.microsoft.com/azure/event-hubs/event-hubs-create) 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](https://learn.microsoft.com/azure/event-hubs/event-hubs-get-connection-string) on the namespace with the Send right, and put it in a [credential](/docs/secrets). Use the namespace's, not one that ends in `EntityPath=`.
    -   a service principal: set `tenantId`, `clientId` and `clientSecret`, with the secret in a [credential](/docs/secrets).
    -   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](https://learn.microsoft.com/entra/identity/managed-identities-azure-resources/overview), or as an AKS pod's [workload identity](https://learn.microsoft.com/azure/aks/workload-identity-overview). 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](https://learn.microsoft.com/azure/event-hubs/authorize-access-azure-active-directory) 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](https://learn.microsoft.com/azure/event-hubs/apache-kafka-configurations).

## Settings

It also takes the settings [every destination](/docs/destinations#settings) 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-logs`
- `acme-logs.servicebus.usgovcloudapi.net`
- `localhost:9092`

`defaultEventHub`  string required

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

Examples:

- `alerts`
- `insights-clean`

`connectionString`  credential default none

The [credential](/docs/secrets) 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](/docs/secrets) 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.us`
- `https://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:

- `60`
- `120`

`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 MiB`
- `16 MiB`
- `1 GiB`

## Fields it reads

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

| Field | Type | Holds |
| --- | --- | --- |
| `__destination.topic` | `str` | The 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](/docs/destinations/kafka) 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](/docs/delivery#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

-   [Destinations](/docs/destinations)
-   [Azure Event Hubs source](/docs/sources/event-hubs)
-   [Handle refused records](/docs/delivery#refused)