Azure Event Hubs component

Q: Can I use Kafka clients instead?

Event Hubs also has a Kafka endpoint; the kafka component can use it with SASL PLAIN and the connection string as password. This component uses the native AMQP protocol.

The azure-eventhubs component sends events to and consumes events from Azure Event Hubs over AMQP 1.0, with partitions, checkpoints and load balancing between consumers.

Scheme: azure-eventhubs: · Crate: cameleer-amqp10 · Cargo feature: amqp10 · Consumer: yes · Producer: yes · Apache Camel: Azure Event Hubs

When to use it

Stream telemetry from devices and gateways into Azure.
Consume high-volume event streams partition by partition, resuming from checkpoints.
Run several consumer instances that share the partitions (like the Azure SDK's event processor).
Read Event Grid events (for example Key Vault or Storage events) that are forwarded to an event hub.

URI format

text

azure-eventhubs:namespace/eventHubName[?options]
azure-eventhubs:?connectionString=...;EntityPath=eventHubName

Example

Send device telemetry; events with the same partition key go to the same partition:

yaml example: azure-eventhubs-send-telemetry

- route:
    id: send-telemetry
    from:
      uri: "timer:telemetry?repeatCount=1"
      steps:
        - setBody:
            constant: '{"device":"edge-7","temperature":21.5}'
        - to: "azure-eventhubs:?connectionString=RAW({{env:EVENTHUBS_CONNECTION_STRING:Endpoint=sb://my-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=change-me;EntityPath=telemetry}})&partitionKey=edge-7"

Consume the event hub from the oldest retained event:

yaml example: azure-eventhubs-consume-telemetry

- route:
    id: read-telemetry
    from:
      uri: "azure-eventhubs:?connectionString=RAW({{env:EVENTHUBS_CONNECTION_STRING:Endpoint=sb://my-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=change-me;EntityPath=telemetry}})&initialPosition=earliest"
      steps:
        - convertBodyTo: String
        - log: "partition ${header.CamelAzureEventHubsPartitionId} offset ${header.CamelAzureEventHubsOffset}: ${body}"
        - to: "log:telemetry"

The event body arrives as bytes. Checkpoints are kept in memory unless you configure a checkpoint store (see the options).

Rust DSL

rust

ctx.add_routes(|b| {
    b.from("azure-eventhubs:?connectionString=RAW({{env:EVENTHUBS_CONNECTION_STRING:Endpoint=sb://my-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=change-me;EntityPath=telemetry}})&initialPosition=earliest")
        .convert_body_to("String")
        .log("partition ${header.CamelAzureEventHubsPartitionId} offset ${header.CamelAzureEventHubsOffset}: ${body}")
        .to("log:telemetry");
})?;

Options

option	default	description
`connectionString`		Connection string; `EntityPath` names the event hub when the uri has none.
`sharedAccessName`, `sharedAccessKey`, `fullyQualifiedNamespace`		Instead of a connection string (with `namespace/eventHubName` in the uri).
`credentialType`	`CONNECTION_STRING`	`CONNECTION_STRING`, `AZURE_IDENTITY` or `TOKEN_CREDENTIAL` (see Azure Service Bus).
`amqpTransportType`	`AMQP`	`AMQP` or `AMQP_WEB_SOCKETS`.
`consumerGroupName`	`$Default`	Consumer group.
`partitionId`		Producer: send to this partition. Consumer: read only this partition.
`partitionKey`		Producer: partition key.
`prefetchCount`	`500`	Events to prefetch per partition.
`initialPosition`	`latest`	Cameleer: where to start without a checkpoint, `latest` or `earliest`.
`eventPosition`		Start position per partition, for example `0=earliest,1=offset:42,2=sequenceNumber:7:inclusive` (or `#bean`).
`checkpointStore`	in memory	`#bean` of a `CheckpointStore`.
`blobAccountName`, `blobContainerName`, `blobAccessKey`, `blobEndpoint`		Checkpoints and ownership in Azure Blob Storage (layout of the Azure SDK's blob checkpoint store).
`checkpointBatchSize`, `checkpointBatchTimeout`	`500`, `5000`	Checkpoint every N events or T ms (and when stopping).
`loadBalancing`	`true`	Share partitions with other instances using the same checkpoint store.
`loadBalancingUpdateInterval`, `partitionOwnershipExpirationInterval`	10 s, 60 s	Load balancing timings (Cameleer).
`ownerId`, `ownerLevel`	random,	Owner id of ownership records and the receiver epoch.
`sasTokenTtl`, `idleTimeout`, `requestTimeout`, `recoveryInterval`		Connection timings.

Message headers

header	direction	description
`CamelAzureEventHubsPartitionId`	in/out	Partition (input: target partition).
`CamelAzureEventHubsPartitionKey`	in/out	Partition key.
`CamelAzureEventHubsOffset`, `CamelAzureEventHubsSequenceNumber`	out	Position of the event.
`CamelAzureEventHubsEnqueuedTime`	out	Enqueued time (epoch millis), also `CamelMessageTimestamp`.

Other headers are sent as application properties, and received application properties become headers.

Run it with Docker

docker run --rm -e EVENTHUBS_CONNECTION_STRING \
  -v "$PWD:/routes:ro" pmsudhi/cameleer run /routes/telemetry.camel.yaml

Common questions

How do I resume after a restart?

Configure a persistent checkpoint store, for example Azure Blob Storage with blobAccountName, blobContainerName and blobAccessKey; the consumer continues after the last checkpoint.

How do I scale out consumers?

Start more instances with the same consumer group and the same blob checkpoint store: partitions are balanced between them, and a new owner takes over the partitions of a stopped instance.

Can I use Kafka clients instead?

Event Hubs also has a Kafka endpoint; the kafka component can use it with SASL PLAIN and the connection string as password. This component uses the native AMQP protocol.

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.