RealtimeTrigger
RealtimeTrigger Certified

Trigger flows from Azure Event Hubs in real time

yaml
type: io.kestra.plugin.azure.eventhubs.RealtimeTrigger
yaml
id: azure_eventhubs_realtime_trigger
namespace: company.team

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

triggers:
  - id: read_from_eventhub
    type: io.kestra.plugin.azure.eventhubs.RealtimeTrigger
    eventHubName: my_eventhub
    namespace: my_eventhub_namespace
    connectionString: "{{ secret('EVENTHUBS_CONNECTION') }}"
    bodyDeserializer: JSON
    consumerGroup: "$Default"
    checkpointStoreProperties:
      containerName: kestra
      connectionString: "{{ secret('BLOB_CONNECTION') }}"
yaml
    id: eventhubs_realtime_trigger
    namespace: company.team

    tasks:
      - id: insert_into_storagetable
        type: io.kestra.plugin.azure.storage.table.Bulk
        endpoint: https://yourstorageaccount.blob.core.windows.net
        connectionString: "{{ secret('STORAGETABLE_CONNECTION') }}"
        table: orders
        from:
          - partitionKey: order_id
            rowKey: "{{ trigger.body | jq('.order_id') | first }}"
            properties:
              customer_name: "{{ trigger.body | jq('.customer_name') | first }}"
              customer_email: "{{ trigger.body | jq('.customer_email') | first }}"
              product_id: "{{ trigger.body | jq('.product_id') | first }}"
              price: "{{ trigger.body | jq('.price') | first }}"
              quantity: "{{ trigger.body | jq('.quantity') | first }}"
              total: "{{ trigger.body | jq('.total') | first }}"

    triggers:
      - id: realtime_trigger
        type: io.kestra.plugin.azure.eventhubs.RealtimeTrigger
        eventHubName: orders
        namespace: kestra
        connectionString: "{{ secret('EVENTHUBS_CONNECTION') }}"
        bodyDeserializer: JSON
        consumerGroup: $Default
        checkpointStoreProperties:
          containerName: kestra
          connectionString: "{{ secret('BLOB_CONNECTION') }}"
Properties
Defaultfalse
DefaultSTRING
Possible Values
STRINGBINARYIONJSON
Default{}
SubTypestring
Default{}
Default5
Default500
Definitions
Example
yaml
id: schedule_condition_datetimebetween
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute once every 5 minutes after the date 2025-12-31T23:59:59Z"

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DateTimeBetween
        date: "{{ trigger.date }}"
        after: "2025-12-31T23:59:59Z"
yaml
id: schedule_condition_datetimebetween
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will be executed once every 5 minutes between the before and after dates"

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DateTimeBetween
        date: "{{ trigger.date }}"
        after: "2025-01-01T00:00:00Z"
        before: "2025-12-31T23:59:59Z"
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.DateTimeBetweenio.kestra.core.models.conditions.types.DateTimeBetweenConditionio.kestra.plugin.core.condition.DateTimeBetweenCondition
afterstring
beforestring
datestring
Default{{ trigger.date }}
Example
yaml
id: schedule_condition_dayweek
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on Mondays at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DayWeek
        dayOfWeek: "MONDAY"
dayOfWeek*Requiredstring
Possible Values
MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAY
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.DayWeekio.kestra.core.models.conditions.types.DayWeekConditionio.kestra.plugin.core.condition.DayWeekCondition
datestring
Default{{ trigger.date }}
Example
yaml
id: schedule_condition_dayweekinmonth
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on the first Monday of the month at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.DayWeekInMonth
        dayOfWeek: "MONDAY"
        dayInMonth: FIRST
dayInMonth*Requiredstring
Possible Values
FIRSTLASTSECONDTHIRDFOURTH
dayOfWeek*Requiredstring
Possible Values
MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAY
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.DayWeekInMonthio.kestra.plugin.core.condition.DayWeekInMonthConditionio.kestra.core.models.conditions.types.DayWeekInMonthCondition
datestring
Default{{ trigger.date }}
Example
yaml
id: flow_condition_executionflow
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when flow `flow_a` of namespace `company.team` enters RUNNING state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionFlow
        flowId: flow_a
        namespace: company.team
    states:
      - RUNNING
flowId*Requiredstring
namespace*Requiredstring
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.ExecutionFlowio.kestra.plugin.core.condition.ExecutionFlowConditionio.kestra.core.models.conditions.types.ExecutionFlowCondition
Example
yaml
id: flow_condition_executionlabels
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when flow with specified labels enters FAILED state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionLabels
        labels:
          owner: john.doe
          env: prod
    states:
      - FAILED
labels*Requiredarrayobject
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.ExecutionLabelsio.kestra.core.models.conditions.types.ExecutionLabelsConditionio.kestra.plugin.core.condition.ExecutionLabelsCondition
Example
yaml
id: flow_condition_executionnamespace
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when any flow within `company.engineering` namespace enters RUNNING state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionNamespace
        namespace: company.engineering
        comparison: PREFIX
    states:
      - RUNNING
namespace*Requiredstring
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.ExecutionNamespaceio.kestra.core.models.conditions.types.ExecutionNamespaceConditionio.kestra.plugin.core.condition.ExecutionNamespaceCondition
comparisonstring
Possible Values
EQUALSPREFIXSUFFIX
prefixbooleanstring
Defaultfalse
Example
yaml
id: flow_condition_executionoutputs
namespace: company.team

tasks:
  - id: upstream_outputs
    type: io.kestra.plugin.core.log.Log
    message: hello from a downstream flow

triggers:
  - id: condition_on_flow_execution_outputs
    type: io.kestra.plugin.core.trigger.Flow
    states:
      - SUCCESS
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionOutputs
        expression: "{{ trigger.outputs.flow_a_output == 'hello' }}"
expression*Requiredbooleanstring
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.ExecutionOutputsio.kestra.core.models.conditions.types.ExecutionOutputsConditionio.kestra.plugin.core.condition.ExecutionOutputsCondition
Example
yaml
id: flow_condition_executionstatus
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when any flow enters FAILED or KILLED state."

triggers:
  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionStatus
        in:
          - FAILED
          - KILLED
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.ExecutionStatusio.kestra.core.models.conditions.types.ExecutionStatusConditionio.kestra.plugin.core.condition.ExecutionStatusCondition
inarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED
notInarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED
Example
yaml
id: myflow
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: Average value has gone below 10

triggers:
  - id: expression_trigger
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/1 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Expression
        expression: "{{ kv('average_value') < 10 }}"
expression*Requiredstring
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.Expressionio.kestra.plugin.core.condition.ExpressionConditionio.kestra.core.models.conditions.types.VariableCondition
Example
yaml
id: flow_condition_hasretryattempt
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute when any flow task on retry enters a specific state(s)."

triggers:
  - id: flow_condition
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.HasRetryAttempt
        in:
          - FAILED
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.HasRetryAttemptio.kestra.plugin.core.condition.HasRetryAttemptConditionio.kestra.core.models.conditions.types.HasRetryAttemptCondition
inarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED
notInarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED
Example
yaml
id: schedule_condition_not
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute on all days except Sunday at 11am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Not
        conditions:
          - type: io.kestra.plugin.core.condition.DayWeek
            dayOfWeek: "SUNDAY"
conditions*Required
Min items1
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.Notio.kestra.core.models.conditions.types.NotConditionio.kestra.plugin.core.condition.NotCondition
Example
yaml
id: schedule_condition_or
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute on Sundays and Mondays at 11am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Or
        conditions:
          - type: io.kestra.plugin.core.condition.DayWeek
            dayOfWeek: "MONDAY"
          - type: io.kestra.plugin.core.condition.DayWeek
            dayOfWeek: "SUNDAY"
conditions*Required
Min items1
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.Orio.kestra.core.models.conditions.types.OrConditionio.kestra.plugin.core.condition.OrCondition
Example
yaml
id: schedule_condition_public-holiday
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on the public holidays of France at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.PublicHoliday
        country: FR
yaml
id: schedule-condition-work-days
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on the work days of France at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Not
        conditions:
          - type: io.kestra.plugin.core.condition.PublicHoliday
            country: FR
          - type: io.kestra.plugin.core.condition.Weekend
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.PublicHolidayio.kestra.plugin.core.condition.PublicHolidayConditionio.kestra.core.models.conditions.types.PublicHolidayCondition
countrystring
datestring
Default{{ trigger.date}}
subDivisionstring
Example
yaml
id: schedule_condition_timebetween
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute every 5 minutes between 4pm and 8pm."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.TimeBetween
        after: "16:00:00+02:00"
        before: "20:00:00+02:00"
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.TimeBetweenio.kestra.plugin.core.condition.TimeBetweenConditionio.kestra.core.models.conditions.types.TimeBetweenCondition
afterstring
beforestring
datestring
Default{{ trigger.date }}
Example
yaml
id: schedule_condition_weekend
namespace: company.team

tasks:
  - id: log_message
    type: io.kestra.plugin.core.log.Log
    message: "This flow will execute only on weekends at 11:00 am."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 11 * * *"
    conditions:
      - type: io.kestra.plugin.core.condition.Weekend
type*Requiredobject
Possible Values
io.kestra.plugin.core.condition.Weekendio.kestra.plugin.core.condition.WeekendConditionio.kestra.core.models.conditions.types.WeekendCondition
datestring
Default{{ trigger.date }}
Default$Default
DefaultEARLIEST
Possible Values
EARLIESTLATESTINSTANT
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

More Plugins in this Category

Plugin Weaviate for Kestra

Connect and query Apache Pinot databases using Kestra's JDBC plugin.

Plugin DataHub for Kestra
OVERVIEW

Created by

Kestra Core Team

Managed by

Kestra Core Team
Plugin Type
CloudDataInfrastructure
Links & Resources
GitHub