Azure Consume
CertifiedConsume events from Azure Event Hubs
Azure Consume
Consume events from Azure Event Hubs
Polls Event Hubs partitions in batches, checkpoints to Azure Blob Storage, and writes events to internal storage as ION. Defaults: consumerGroup=$Default, partitionStartingPosition=EARLIEST, maxBatchSizePerPartition=50, maxWaitTimePerPartition=PT5S, maxDuration=PT10S. Requires checkpointStoreProperties.connectionString and .containerName.
type: io.kestra.plugin.azure.eventhubs.ConsumeExamples
Consume data events from Azure EventHubs.
id: azure_eventhubs_consume_data_events
namespace: company.team
tasks:
- id: consume_from_eventhub
type: io.kestra.plugin.azure.eventhubs.Consume
eventHubName: my_eventhub
namespace: my_eventhub_namespace
connectionString: "{{ secret('EVENTHUBS_CONNECTION') }}"
bodyDeserializer: JSON
consumerGroup: "$Default"
checkpointStoreProperties:
containerName: kestra
connectionString: "{{ secret('BLOB_CONNECTION') }}"
Properties
checkpointStoreProperties *object
Checkpoint store properties
Blob container config for checkpoints (connectionString, containerName required)
eventHubName *string
Event Hub name
Target Event Hub entity
namespace *string
Namespace
Event Hubs namespace when using AAD or shared key auth
assets
Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset failure behavior
Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150bodyDeserializer string
STRINGSTRINGBINARYIONJSONBody deserializer
Serde used to decode event bodies; defaults to STRING
bodyDeserializerProperties object
{}Deserializer properties
Key/value options passed to the selected serde
clientMaxRetries integerstring
5Client max retries
Max retry attempts for Event Hubs client operations; default 5
clientRetryDelay integerstring
500Retry delay (ms)
Delay between client retries in milliseconds; default 500
connectionString string
Event Hubs connection string
Namespace or Event Hub–level connection string; overrides key/sas fields when set
consumerGroup string
$DefaultConsumer group
Event Hubs consumer group name; defaults to $Default
customEndpointAddress string
Custom endpoint address
Custom endpoint for Event Hubs (e.g., for private link); optional
enqueueTime string
Start from enqueue time
ISO-8601 datetime applied only when partitionStartingPosition is set to INSTANT; ignored for EARLIEST and LATEST
maxBatchSizePerPartition integerstring
50Max batch size per partition
Maximum events pulled per partition read; defaults to 50
maxDuration string
PT10SOverall max duration
Stop consuming after this duration; defaults to PT10S
maxWaitTimePerPartition string
PT5SMax wait per partition
Maximum wait for a partition batch before returning; defaults to PT5S
partitionStartingPosition string
EARLIESTEARLIESTLATESTINSTANTStarting position
Initial position strategy per partition; defaults to EARLIEST
sasToken string
SAS token
Precomputed SAS token for Event Hubs; optional alternative to shared key/connection string
Outputs
eventsCount integer
Events consumed
uri string
uriConsumed events URI
kestra:// URI for the ION file containing consumed events
Metrics
records.consumed counter
The total number of events consumed.
