Apache Kafka RealtimeTrigger

Apache Kafka RealtimeTrigger

Certified

Start a Flow for each Kafka record

Consumes Kafka messages as they arrive and starts one Execution per record. In groupType: CONSUMER (default), behavior is classic consumer groups with manual offset commits (auto-commit disabled) and STRING deserializers by default. Configure groupId, serdeProperties, or since to control offsets and schema handling. Without groupId, partitions are assigned manually and no offsets are committed: every restart (worker restart, flow update) reads from the beginning again and starts one Execution per old record. groupId is required for a plain topic, unless topicPattern or since is set. Set since (e.g. "{{ now() }}") to read only new records when no groupId is used. since is rendered again at every trigger start, so with "{{ now() }}" records produced while the trigger was down (restart, flow update, deploy) are skipped. Use groupId if no record can be lost. In groupType: SHARE, behavior is queue semantics with share groups and explicit acknowledgements. In SHARE mode, use topic with groupId; topicPattern, partitions, and since are not supported. Use header filters or recordFilter to drop unmatched records before they start an execution. Prefer the batch Kafka Trigger for interval-based pulls. Emits DEBUG logs on startup, subscription, connection confirmation, and shutdown — grep by triggerId to verify Kafka connectivity.

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

Consume a message from a Kafka topic in real time.

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

Start a flow only for approved change requests.

yaml
id: kafka_realtime_record_filter
namespace: company.team

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

triggers:
  - id: approved_change_requests
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: change_requests
    properties:
      bootstrap.servers: localhost:9092
    valueDeserializer: JSON
    groupId: kestra_change_requests
    recordFilter: "{{ trigger.value.event == 'change-request.approved' }}"

Use Kafka Realtime Trigger to push events into MongoDB

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

Use Kafka share group queue semantics in realtime

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

Read only new records without a consumer group (records produced while the trigger is down are skipped, use groupId to avoid losing them)

yaml
id: kafka_realtime_no_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
    topicPattern: "orders-.*"
    since: "{{ now() }}"
    properties:
      bootstrap.servers: localhost:9092

Propagate OpenTelemetry trace context from Kafka records to flow executions

yaml
id: kafka_realtime_trace_context
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "Received record with headers: {{ trigger.headers }}"

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.kafka.RealtimeTrigger
    topic: orders
    properties:
      bootstrap.servers: localhost:9092
    groupId: orders-consumer-group
    traceContext:
      enabled: true
Installation

Our standard Docker images already include this plugin. If you use a -slim image or another setup, install it with the following command: 

kestractl plugins get io.kestra.plugin:plugin-kafka --compatible-for $KESTRA_VERSION

For kestractl, use your Kestra version with --compatible-for, or specify an exact plugin version.

For Kestra CLI, replace LATEST with a specific version or version range to pin the plugin version.

Properties

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.

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.

Defaultfalse

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

Defaultfalse

Deduplicate records

When enabled, duplicate records retrieved during the same execution are filtered out using the Kafka topic, partition, and offset.

Kafka consumer group ID

Determines offset management; mandatory for groupType: SHARE (share group) and for the polling Trigger. Optional for Consume. Required for RealtimeTrigger with a plain topic, unless topicPattern or since is set. Without it, records are read from the beginning (or from since), no offsets are committed, and with topicPattern the matching topics are resolved once at startup, so topics created later are not picked up.

DefaultCONSUMER
Possible Values
CONSUMERSHARE

Group protocol to consume with

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

Filter messages by Kafka headers

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

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.

Behavior on serde error

Applies to value deserialization errors only: SKIPPED (default), STORE to internal storage, or DLQ to the configured topic. Key deserialization errors are not handled. For RealtimeTrigger, STORE starts an execution whose trigger.value is the stored file URI.

Definitions
topicstring

Topic used when type is DLQ

typestring
DefaultSKIPPED
Possible Values
SKIPPEDDLQSTORE

Action to take on serde error

SubTypeinteger

Specific partitions to consume

Manually assign partitions; bypasses consumer group rebalancing.

Filter records before creating executions

Boolean expression rendered for each record, with the record available as trigger (key, value, headers, topic, partition, offset and timestamp), for example {{ trigger.value.event == 'approved' }}. When unset, every record starts an execution. Records that render to false are skipped: committed in CONSUMER mode, acknowledged with ACCEPT in SHARE mode. A record whose expression fails to render, or does not render to a boolean, is skipped the same way with a warning. It runs after headerFilters and deduplication. Records recovered by onSerdeError are not filtered. Unlike conditions, which are evaluated once for the trigger, it sees each record.

Schema registry vendor

Definitions
Example

Consume AVRO data from a Kafka topic using AWS Glue Schema Registry

yaml
id: consume-kafka-messages
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.kafka.Consume
    topic: topic_test
    properties:
      bootstrap.servers: localhost:9093
      auto.offset.reset: earliest
    pollDuration: PT20S
    maxRecords: 50
    keyDeserializer: STRING
    valueDeserializer: AVRO
    schemaRegistryVendor:
      type: io.kestra.plugin.kafka.registry.AwsGlueSchemaRegistry
      region: us-east-1
      # endpoint, accessKey, secretKey are optional when using IAM roles
      accessKey: "{{ secret('AWS_ACCESS_KEY_ID') }}"
      secretKey: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
region*string

The AWS region where the schema registry is located

type*object
accessKeystring

The AWS access key ID. Optional when using IAM roles or environment credentials

endpointstring

The AWS Glue Schema Registry endpoint. Optional when using IAM roles or standard regional endpoints

secretKeystring

The AWS secret access key. Optional when using IAM roles or environment credentials

Example

Consume data from a Kafka topic with Confluent Schema Registry

yaml
id: consume-kafka-messages
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.kafka.Consume
    topic: topic_test
    properties:
      bootstrap.servers: localhost:9092
    pollDuration: PT20S
    maxRecords: 50
    keyDeserializer: STRING
    valueDeserializer: AVRO
    schemaRegistryVendor:
      type: io.kestra.plugin.kafka.registry.ConfluentSchemaRegistry
      schemaRegistryUrl: http://localhost:8081
schemaRegistryUrl*string

The Confluent Schema Registry URL

type*object
Default{}

Serializer or deserializer properties

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

Timestamp to start consuming from

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

SubTypestring
Possible Values
CREATEDSUBMITTEDRUNNINGPAUSEDRESTARTEDKILLINGSUCCESSWARNINGFAILEDKILLEDCANCELLEDQUEUEDRETRYINGRETRIEDSKIPPEDBREAKPOINTRESUBMITTED

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

Kafka topic(s) to consume from

String or list of strings; mutually exclusive with topicPattern.

Regex pattern of topics to consume from

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

Extract OpenTelemetry trace context from Kafka records and propagate to the execution

Propagation takes effect on Kestra 1.3.x; support on Kestra 2.0.x is pending a core update.

Definitions
bodyExtractorstring

Optional field path to extract the traceparent from the message body (e.g. traceparent or trace.parent)

Supported when message value is a JSON string or a Map.

enabledbooleanstring
Defaultfalse

Whether to extract trace context from the Kafka message and propagate it to the execution

Disabled by default. When enabled, W3C traceparent will be extracted and propagated to the execution. Note: Propagation takes effect on Kestra 1.3.x; support on Kestra 2.0.x is pending a core update.

traceparentHeaderstring
Defaulttraceparent

Kafka header name for the W3C traceparent

Defaults to traceparent.

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.

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.

The message headers

Definitions

The message key

The message offset

The partition the message belongs to

Formatdate-time

The message timestamp

The topic the message belongs to

The message value

More Plugins in this Category

Firecrawl
Firecrawl

Tasks that scrape and crawl websites into clean, LLM-ready content with Firecrawl.

Data

2 tasks

Canonical (EE)
Canonical
Enterprise

Tasks to interact with Canonical services, enabling automation and infrastructure management through API-driven workflows.

CloudInfrastructure

5 tasks