Azure Event Hubs Trigger

Azure Event Hubs Trigger
Certified
Poll Azure Event Hubs and trigger flows

Periodically consumes events in batches, checkpoints to Blob Storage, and triggers one execution per batch. Defaults: interval=PT60S, consumerGroup=$Default, partitionStartingPosition=EARLIEST, maxBatchSizePerPartition=50, maxWaitTimePerPartition=PT5S, maxDuration=PT10S. Use RealtimeTrigger for per-event executions.

yaml
type: io.kestra.plugin.azure.eventhubs.Trigger

Examples

Trigger flow based on events received from Azure Event Hubs in batch.

yaml
id: azure_eventhubs_trigger
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: Hello there! I received {{ trigger.eventsCount }} from Azure EventHubs!

triggers:
  - id: read_from_eventhub
    type: io.kestra.plugin.azure.eventhubs.Trigger
    interval: PT30S
    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

The event hub to read from

namespace *string

Namespace name of the event hub to connect to

allowConcurrent boolean

Defaultfalse

Specifies whether a trigger is allowed to start a new execution even if a previous run is still in progress.

bodyDeserializer string

DefaultSTRING

Possible Values

STRINGBINARYIONJSON

Body deserializer

Serde used to decode event bodies; defaults to STRING

bodyDeserializerProperties object

Default{}

Deserializer properties

Key/value options passed to the selected serde

clientMaxRetries integerstring

Default5

The maximum number of retry attempts before considering a client operation to have failed

clientRetryDelay integerstring

Default500

The maximum permissible delay between retry attempts in milliseconds

connectionString string

Connection string of the Storage Account.

consumerGroup string

Default$Default

Consumer group

Event Hubs consumer group; defaults to $Default

customEndpointAddress string

Custom endpoint address when connecting to the Event Hubs service

enqueueTime string

Start from enqueue time

ISO-8601 datetime applied only when partitionStartingPosition is set to INSTANT; ignored for EARLIEST and LATEST

interval string

DefaultPT1M

Formatduration

Polling interval

Time between poll cycles; defaults to PT60S

maxBatchSizePerPartition integerstring

Default50

Max batch size per partition

Maximum events pulled per partition read; defaults to 50

maxDuration string

DefaultPT10S

Overall max duration

Stop consuming after this duration each poll; defaults to PT10S

maxWaitTimePerPartition string

DefaultPT5S

Max wait per partition

Maximum wait for a partition batch before returning; defaults to PT5S

partitionStartingPosition string

DefaultEARLIEST

Possible Values

EARLIESTLATESTINSTANT

Starting position

Initial position strategy per partition; defaults to EARLIEST

sasToken string

The SAS token to use for authenticating requests.

This string should only be the query parameters (with or without a leading '?') and not a full URL.

sharedKeyAccountAccessKey string

Shared Key access key for authenticating requests.

sharedKeyAccountName string

Shared Key account name for authenticating requests.

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

List of execution states after which a trigger should be stopped (a.k.a. disabled).

when string

Defaulttrue

A condition that determines whether the trigger should run.

A Pebble expression evaluated at trigger time. The trigger fires only when the expression evaluates to a truthy value (true, a non-empty string, a non-zero number). Use this to gate trigger execution on dynamic runtime values such as execution labels, flow variables, or environment conditions.

Outputs

eventsCount integer

Events consumed

uri string

Formaturi

Consumed events URI

kestra:// URI for the ION file containing consumed events