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. 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 to drop unmatched records. 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

Examples

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

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

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) 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.

headerFilters object

Filter messages by Kafka headers

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

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.

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.

schemaRegistryVendor

Schema registry vendor

Definitions

AWS Glue Schema Registry

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

Confluent Schema Registry

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

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

headers array

The message headers

Definitions

org.apache.commons.lang3.tuple.Pair

key object

The message key

offset integer

The message offset

partition integer

The partition the message belongs to

timestamp string