Apache Kafka Consume

Apache Kafka Consume
Certified
Read Kafka records into internal storage

Consumes from configured topics or regex with manual offset commits (auto-commit disabled) and committed-only reads by default. Writes all fetched records to Kestra internal storage as ION at uri and returns the count. Defaults: pollDuration PT5S, STRING deserializers, Avro logical type converters enabled. Use groupType: CONSUMER (default, backward compatible) for classic consumer groups, or groupType: SHARE for queue semantics with share groups and explicit acknowledgements. In SHARE mode, topic and groupId are required, and topicPattern, partitions, and since are not supported. Use since, maxRecords, maxDuration, or header filters to stop early.

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

Examples

Consume data from a Kafka topic

yaml
id: kafka_consume
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.kafka.Consume
    topic: test_kestra
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      schema.registry.url: http://localhost:8085
    keyDeserializer: STRING
    valueDeserializer: AVRO
    schemaRegistryVendor:
      type: io.kestra.plugin.kafka.registry.ConfluentSchemaRegistry
      schemaRegistryUrl: http://localhost:8085

Connect to a Kafka cluster with SSL.

yaml
id: kafka_consume
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.kafka.Consume
    properties:
      security.protocol: SSL
      bootstrap.servers: localhost:19092
      ssl.key.password: "{{ secret('SSL_KEY_PASSWORD') }}"
      ssl.keystore.type: PKCS12
      ssl.keystore.location: my-base64-encoded-keystore
      ssl.keystore.password: "{{ secret('SSL_KEYSTORE_PASSWORD') }}"
      ssl.truststore.location: my-base64-encoded-truststore
      ssl.truststore.password: "{{ secret('SSL_TRUSTSTORE_PASSWORD') }}"
    topic:
      - kestra_workerinstance
    keyDeserializer: STRING
    valueDeserializer: STRING

Consume data from a Kafka topic and write it to a JSON file

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: JSON

  - id: write_json
    type: io.kestra.plugin.serdes.json.IonToJson
    newLine: true
    from: "{{ outputs.consume.uri }}"

Consume only records whose headers match

yaml
id: consume_with_headers
namespace: company.team

tasks:
  - id: consume_filtered
    type: io.kestra.plugin.kafka.Consume
    topic: orders
    properties:
      bootstrap.servers: localhost:9092
      auto.offset.reset: earliest
    keyDeserializer: STRING
    valueDeserializer: JSON
    headerFilters:
      event-type: order_created
      region: us-east

Consume queue-style with Kafka share groups

yaml
id: consume_kafka_share_group
namespace: company.team

tasks:
  - id: consume_queue
    type: io.kestra.plugin.kafka.Consume
    topic: orders
    properties:
      bootstrap.servers: localhost:9092
    groupId: orders-share-group
    groupType: SHARE
    acknowledgeType: ACCEPT
    keyDeserializer: STRING
    valueDeserializer: JSON

Properties

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) marks a record as processed, RELEASE returns it to the queue, REJECT negatively acknowledges it, and RENEW extends the acquisition lock timeout for the current delivery attempt without changing record state. Ignored when groupType is CONSUMER.

groupId string

Kafka consumer group ID

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

groupType string

DefaultCONSUMER

Possible Values

CONSUMERSHARE

Group protocol to consume with

CONSUMER (default) keeps backward-compatible Kafka consumer-group behavior. SHARE enables Kafka share-group queue semantics with explicit acknowledgements. In SHARE mode, groupId and topic are required, and 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.

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.

pluginDefaultsRef string

Reference (ref) of the pluginDefaults to apply to this task.

pollDuration string

DefaultPT5S

How often to poll for records

Maximum wait per poll when no records are available.

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.

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.

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

Metrics

records counter

Unitrecords

Number of records consumed from Kafka topic.

Liquibase

Tasks that run Liquibase change management commands.

Data

PostgreSQL

Tasks that connect to PostgreSQL via JDBC to run queries and loads.

Data

Apache Kafka ConsumeCertifiedRead Kafka records into internal storage

Apache Kafka Consume

More Plugins in this Category

Liquibase

PostgreSQL

Apache Kafka Consume
Certified
Read Kafka records into internal storage