RealtimeTrigger

RealtimeTrigger

Start a Flow for each Kafka record

yaml
type: io.kestra.plugin.kafka.RealtimeTrigger

Examples

yaml
id: kafka_realtime_trigger
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.value ?? '' }}"

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: test_kestra
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      schema.registry.url: http://localhost:8085
    keyDeserializer: STRING
    valueDeserializer: AVRO
    groupId: kafkaConsumerGroupId

yaml
id: kafka_realtime_trigger
namespace: company.team

tasks:
  - id: insert_into_mongodb
    type: io.kestra.plugin.mongodb.InsertOne
    connection:
      uri: mongodb://mongoadmin:secret@localhost:27017/?authSource=admin
    database: kestra
    collection: products
    document: |
      {
        "product_id": "{{ trigger.value ?? '' | jq('.product_id') | first }}",
        "product_name": "{{ trigger.value ?? '' | jq('.product_name') | first }}",
        "category": "{{ trigger.value ?? '' | jq('.product_category') | first }}",
        "brand": "{{ trigger.value ?? '' | jq('.brand') | first }}"
      }

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: products
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      valueDeserializer: JSON
    groupId: kestraConsumer

yaml
id: kafka_realtime_share_group
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.value ?? '' }}"

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: orders
    properties:
      bootstrap.servers: localhost:9092
    groupId: orders-share-group
    groupType: SHARE
    acknowledgeType: ACCEPT

Properties

groupId *string

properties *object

SubTypestring

acknowledgeType string

DefaultACCEPT

Possible Values

ACCEPTRELEASEREJECTRENEW

allowConcurrent boolean

Defaultfalse

conditions

Definitions

Allow events only between two datetimes.

type*object

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 }}

Allow events on a specific weekday.

dayOfWeek*string

Possible Values

MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAY

type*object

Possible Values

io.kestra.plugin.core.condition.DayWeekio.kestra.core.models.conditions.types.DayWeekConditionio.kestra.plugin.core.condition.DayWeekCondition

datestring

Default{{ trigger.date }}

Allow events on an nth weekday within the month.

dayInMonth*string

Possible Values

FIRSTLASTSECONDTHIRDFOURTH

dayOfWeek*string

Possible Values

MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAY

type*object

Possible Values

io.kestra.plugin.core.condition.DayWeekInMonthio.kestra.plugin.core.condition.DayWeekInMonthConditionio.kestra.core.models.conditions.types.DayWeekInMonthCondition

datestring

Default{{ trigger.date }}

Match events from a specific flow.

flowId*string

namespace*string

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionFlowio.kestra.plugin.core.condition.ExecutionFlowConditionio.kestra.core.models.conditions.types.ExecutionFlowCondition

Match executions by label set.

labels*arrayobject

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionLabelsio.kestra.core.models.conditions.types.ExecutionLabelsConditionio.kestra.plugin.core.condition.ExecutionLabelsCondition

Match executions by namespace.

namespace*string

type*object

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

Condition based on the outputs of an upstream execution.

expression*booleanstring

type*object

Possible Values

io.kestra.plugin.core.condition.ExecutionOutputsio.kestra.core.models.conditions.types.ExecutionOutputsConditionio.kestra.plugin.core.condition.ExecutionOutputsCondition

Match executions by status.

type*object

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

Condition based on variable expression.

expression*string

type*object

Possible Values

io.kestra.plugin.core.condition.Expressionio.kestra.plugin.core.condition.ExpressionConditionio.kestra.core.models.conditions.types.VariableCondition

Match executions where a task was retried.

type*object

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

Invert one or more conditions.

conditions*

Min items1

type*object

Possible Values

io.kestra.plugin.core.condition.Notio.kestra.core.models.conditions.types.NotConditionio.kestra.plugin.core.condition.NotCondition

Pass when any condition is true.

conditions*

Min items1

type*object

Possible Values

io.kestra.plugin.core.condition.Orio.kestra.core.models.conditions.types.OrConditionio.kestra.plugin.core.condition.OrCondition

Allow events on public holidays.

type*object

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

Allow events between two times of day.

type*object

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 }}

Allow events on weekends.

type*object

Possible Values

io.kestra.plugin.core.condition.Weekendio.kestra.plugin.core.condition.WeekendConditionio.kestra.core.models.conditions.types.WeekendCondition

datestring

Default{{ trigger.date }}

groupType string

DefaultCONSUMER

Possible Values

CONSUMERSHARE

headerFilters object

SubTypestring

keyDeserializer string

DefaultSTRING

Possible Values

STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSON

onSerdeError

Definitions

io.kestra.plugin.kafka.KafkaConsumerInterface-OnSerdeError

topicstring

typestring

DefaultSKIPPED

Possible Values

SKIPPEDDLQSTORE

partitions array

SubTypeinteger

serdeProperties object

SubTypestring

Default{}

since string

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

topic object

topicPattern string

valueDeserializer string

DefaultSTRING

Possible Values

STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSON

Outputs

headers array

Definitions

org.apache.commons.lang3.tuple.Pair

key object

offset integer

partition integer

timestamp string

Formatdate-time

topic string

value object

Apache Cassandra

Tasks that integrate Apache Cassandra into Kestra workflows for querying and event-driven triggers.

Data

Git

Tasks that clone, fetch, and interact with Git repositories.

InfrastructureBusiness

MongoDB

Tasks that query and manipulate MongoDB collections.

Data

RealtimeTrigger Start a Flow for each Kafka record

More Plugins in this Category

Apache Cassandra

Git

MongoDB

RealtimeTrigger

Start a Flow for each Kafka record