RealtimeTrigger 
RealtimeTrigger
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: kafkaConsumerGroupIdyaml
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
Properties
groupId *Requiredstring
properties *Requiredobject
SubTypestring
allowConcurrent Non-dynamicboolean
Default
falseconditions Non-dynamic
Definitions
Condition to allow events between two specific datetime values.
type*Requiredobject
afterstring
Format
date-timebeforestring
Format
date-timedatestring
Default
{{ trigger.date }}Condition to allow events on a particular day of the week.
dayOfWeek*Requiredstring
Possible Values
MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAYtype*Requiredobject
datestring
Default
{{ trigger.date }}Condition to execute tasks on a specific day of the week relative to the current month (first, last, ...)
dayInMonth*Requiredstring
Possible Values
FIRSTLASTSECONDTHIRDFOURTHdayOfWeek*Requiredstring
Possible Values
MONDAYTUESDAYWEDNESDAYTHURSDAYFRIDAYSATURDAYSUNDAYtype*Requiredobject
datestring
Default
{{ trigger.date }}Condition for a specific flow of an execution.
flowId*Requiredstring
namespace*Requiredstring
type*Requiredobject
Condition that checks labels of an execution.
labels*Requiredarrayobject
type*Requiredobject
Condition for an execution namespace.
namespace*Requiredstring
type*Requiredobject
comparisonstring
Possible Values
EQUALSPREFIXSUFFIXprefixbooleanstring
Default
falseCondition based on the outputs of an upstream execution.
expression*Requiredbooleanstring
type*Requiredobject
Condition based on execution status.
type*Requiredobject
inarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTEDnotInarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTEDCondition based on variable expression.
expression*Requiredstring
type*Requiredobject
Condition for a specific flow. Note that this condition is deprecated, use `io.kestra.plugin.core.condition.ExecutionFlow` instead.
flowId*Requiredstring
namespace*Requiredstring
type*Requiredobject
Condition for a flow namespace.
namespace*Requiredstring
type*Requiredobject
prefixboolean
Default
falseCondition that matches if any taskRun has retry attempts.
type*Requiredobject
inarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTEDnotInarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTEDRun a flow if the list of preconditions is met in a time window.
conditions*Requiredobject
id*Requiredstring
Validation RegExp
^[a-zA-Z0-9][a-zA-Z0-9_-]*Min length
1type*Requiredobject
resetOnSuccessboolean
Default
truetimeWindow
Default
{
"type": "DURATION_WINDOW"
}io.kestra.core.models.triggers.TimeWindow
deadlinestring
Format
partial-timeendTimestring
Format
partial-timestartTimestring
Format
partial-timetypestring
Default
DURATION_WINDOWPossible Values
DAILY_TIME_DEADLINEDAILY_TIME_WINDOWDURATION_WINDOWSLIDING_WINDOWwindowstring
Format
durationwindowAdvancestring
Format
durationwindowDeprecatedstring
Format
durationwindowAdvanceDeprecatedstring
Format
durationCondition to exclude other conditions.
conditions*Required
Min items
1type*Requiredobject
Condition to have at least one condition validated.
conditions*Required
Min items
1type*Requiredobject
Condition to allow events on public holidays.
type*Requiredobject
countrystring
datestring
Default
{{ trigger.date}}subDivisionstring
Condition to allow events between two specific times.
type*Requiredobject
afterstring
Format
timebeforestring
Format
timedatestring
Default
{{ trigger.date }}Condition to allow events on weekend.
type*Requiredobject
datestring
Default
{{ trigger.date }}headerFilters object
SubTypestring
keyDeserializer string
Default
STRINGPossible Values
STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONonSerdeError Non-dynamic
Definitions
io.kestra.plugin.kafka.KafkaConsumerInterface-OnSerdeError
topicstring
typestring
Default
SKIPPEDPossible Values
SKIPPEDDLQSTOREpartitions array
SubTypeinteger
serdeProperties object
SubTypestring
Default
{}since string
stopAfter Non-dynamicarray
SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTEDtopic Non-dynamicobject
topicPattern string
valueDeserializer string
Default
STRINGPossible Values
STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONOutputs
headers array
Definitions
org.apache.commons.lang3.tuple.Pair
key object
offset integer
partition integer
timestamp string
Format
date-time
