Apache Kafka Trigger

Apache Kafka Trigger
Certified
Start a Flow on scheduled Kafka pulls

Polls Kafka on a fixed interval (default PT1M, pollDuration PT5S) to batch records into one Execution. In groupType: CONSUMER (default), behavior is classic consumer groups with manual offset commits and committed-only reads. In groupType: SHARE, behavior is queue semantics with share groups and explicit acknowledgements. Records are stored in internal storage at {{ trigger.uri }}; defaults use STRING deserializers. Use header filters to drop mismatching records or switch to RealtimeTrigger for one-execution-per-record.

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

Examples

yaml
id: kafka_trigger
namespace: company.team

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

triggers:
  - id: trigger
    type: io.kestra.plugin.kafka.Trigger
    topic: test_kestra
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      schema.registry.url: http://localhost:8085
    keyDeserializer: STRING
    valueDeserializer: AVRO
    interval: PT30S
    maxRecords: 5
    groupId: kafkaConsumerGroupId

Use a Kafka share group for queue semantics

yaml
id: kafka_trigger_share_group
namespace: company.team

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

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

Properties

groupId *string

Kafka consumer group ID

Determines offset management; required when using topicPattern and mandatory for groupType: SHARE (share group).

properties *object

Kafka client properties

Must include bootstrap.servers; accepts any Kafka consumer or producer config. Provide base64-encoded content for ssl.keystore.location and ssl.truststore.location when using SSL.

acknowledgeType string

DefaultACCEPT

Possible Values

ACCEPTRELEASEREJECTRENEW

Acknowledgement action for SHARE group type

Used only when groupType is SHARE. ACCEPT (default) acknowledges processed records, RELEASE returns records to the queue, REJECT negatively acknowledges records, and RENEW extends the acquisition lock timeout for the current delivery attempt without changing record state. Ignored when groupType is CONSUMER.

allowConcurrent boolean

Defaultfalse

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

groupType string

DefaultCONSUMER

Possible Values

CONSUMERSHARE

Group protocol to consume with

CONSUMER (default) polls with classic consumer-group behavior. SHARE polls with Kafka share-group queue semantics and explicit acknowledgements. In SHARE mode, use topic with groupId; topicPattern, partitions, and since are not supported.

headerFilters object

Filter messages by Kafka headers

Consume records only when all header key/value pairs match exactly (last header wins, UTF-8 comparison)

interval string

DefaultPT1M

Formatduration

Interval between polling.

The interval between 2 different polls of schedule, this can avoid to overload the remote system with too many calls. For most of the triggers that depend on external systems, a minimal interval must be at least PT30S. See ISO_8601 Durations for more information of available interval values.

keyDeserializer string

DefaultSTRING

Possible Values

STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONPROTOBUF

Deserializer used for the key

Default STRING. Options: STRING, INTEGER, FLOAT, DOUBLE, LONG, SHORT, BYTE_ARRAY, BYTE_BUFFER, BYTES, UUID, VOID, AVRO, JSON.

maxDuration string

Maximum duration to wait before stopping

Soft limit checked on each poll.

maxRecords integerstring

Maximum records to consume before stopping

Soft limit checked on each poll.

onSerdeError

Behavior on serde error

Applies when valueDeserializer is JSON: SKIPPED (default), STORE to internal storage, or DLQ to the configured topic.

Definitions

io.kestra.plugin.kafka.KafkaConsumerInterface-OnSerdeError

topicstring

Topic used when type is DLQ

typestring

DefaultSKIPPED

Possible Values

SKIPPEDDLQSTORE

Action to take on serde error

partitions array

SubTypeinteger

Specific partitions to consume

Manually assign partitions; bypasses consumer group rebalancing.

pollDuration string

DefaultPT5S

How often to poll for records

Maximum wait per poll when no records are available.

serdeProperties object

Default{}

Serializer or deserializer properties

Passed to serdes; avro.use.logical.type.converters is forced to true by default.

since string

Timestamp to start consuming from

ISO-8601 instant used when no consumer group offsets exist; ignored when a consumer group controls offsets.

stopAfter array

SubTypestring

Possible Values

CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

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

topic object

Kafka topic(s) to consume from

String or list of strings; mutually exclusive with topicPattern.

topicPattern string

Regex pattern of topics to consume from

Subscribes to topics matching the pattern and receives dynamic partition assignments; mutually exclusive with topic.

valueDeserializer string

DefaultSTRING

Possible Values

STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONPROTOBUF

Deserializer used for the value

Default STRING. Options: STRING, INTEGER, FLOAT, DOUBLE, LONG, SHORT, BYTE_ARRAY, BYTE_BUFFER, BYTES, UUID, VOID, AVRO, JSON.

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

messagesCount integer

Number of messages consumed from a Kafka topic

uri string

Formaturi

URI of a file in Kestra's internal storage containing the messages