ESC
Ask Kestra AI
ESC
Kestra AI

Hi! I'm your Kestra AI assistant.
Ask me anything about workflows.

EXAMPLE QUESTIONS

How to set up CI/CD for kestra flows?
How to receive an alert on flow failure?
How to sync my flows with Git?

Apache Kafka Produce

Apache Kafka Produce

Certified

Publish records to Kafka topics

Reads messages from from data (key/value/topic/partition/timestamp/headers) and sends them with configurable serializers (default STRING) and transactional sends enabled by default. Requires bootstrap.servers; topic can come from task config or each payload. Avro serializers require the matching schema (keyAvroSchema / valueAvroSchema) and serdeProperties (for example schema.registry.url). Produced records can be consumed by both classic consumer groups (groupType: CONSUMER) and share groups (groupType: SHARE) in Kafka consume and trigger tasks.

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

Send a string to a Kafka topic

yaml
id: kafka_producer
namespace: company.team

tasks:
  - id: kafka_producer
    type: io.kestra.plugin.kafka.Produce
    properties:
      bootstrap.servers: localhost:9092
    topic: example_topic
    from:
      key: "{{ execution.id }}"
      value: "Hello, World!"
      timestamp: "{{ execution.startDate }}"
      headers:
        x-header: some value
    keySerializer: STRING
    valueSerializer: STRING
    serdeProperties:
      schema.registry.url: http://localhost:8085

Read a CSV file, transform it and send it to Kafka.

yaml
id: send_message_to_kafka
namespace: company.team

inputs:
  - id: file
    type: FILE
    description: "A CSV file with columns: id, username, tweet, and timestamp."

tasks:
  - id: csv_to_ion
    type: io.kestra.plugin.serdes.csv.CsvToIon
    from: "{{ inputs.file }}"

  - id: ion_to_avro_schema
    type: io.kestra.plugin.graalvm.js.FileTransform
    from: "{{ outputs.csv_to_ion.uri }}"
    script: |
      var result = {
        "key": row.id,
        "value": {
          "username": row.username,
          "tweet": row.tweet
        },
        "timestamp": row.timestamp,
        "headers": {
          "key": "value"
        }
      };
      row = result

  - id: avro_to_kafka
    type: io.kestra.plugin.kafka.Produce
    from: "{{ outputs.ion_to_avro_schema.uri }}"
    keySerializer: STRING
    properties:
      bootstrap.servers: localhost:9092
    serdeProperties:
      schema.registry.url: http://localhost:8085
    topic: example_topic
    valueAvroSchema: |
      {"type":"record","name":"twitter_schema","namespace":"io.kestra.examples","fields":[{"name":"username","type":"string"},{"name":"tweet","type":"string"}]}
    valueSerializer: AVRO
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.

Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.

Definitions
assetFailureBehaviorstring
Possible Values
IGNOREFAILWARN

Asset failure behavior

Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.

enableAutobooleanstring

Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.

inputsarray

The assets consumed as inputs.

id*string
Min length1
typestring
outputs

The assets produced as outputs.

id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*string
Min length1

Custom asset type

descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150

Structured data items, either as a map, a list of map, a URI, or a JSON string.

Structured data items can be defined in the following ways:

  • A single item as a map (a document).
  • A list of items as a list of maps (a list of documents).
  • A URI, supported schemes are kestra for internal storage files, file for host local files, and nsfile for namespace files.
  • A JSON String that will then be serialized either as a single item or a list of items.

Avro schema when the key serializer is AVRO

Required if keySerializer is AVRO.

DefaultSTRING
Possible Values
STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONPROTOBUF

Serializer used for the key

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

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.

Target Kafka topic when not provided in payload

Can be overridden per record by setting topic inside from.

Defaulttrue

Enable transactional producer

Defaults to true; generates a transactional.id automatically so sends are committed atomically.

Avro schema when the value serializer is AVRO

Required if valueSerializer is AVRO.

DefaultSTRING
Possible Values
STRINGINTEGERFLOATDOUBLELONGSHORTBYTE_ARRAYBYTE_BUFFERBYTESUUIDVOIDAVROJSONPROTOBUF

Serializer used for the value

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

Number of messages sent to a Kafka topic

Unitrecords

Number of records sent to Kafka topic.

More Plugins in this Category

Twilio
Twilio

Tasks that integrate with the Twilio ecosystem.

BusinessData

9 tasks

4 blueprints

AMQP
AMQP

Tasks that connect to RabbitMQ brokers to declare exchanges and queues, bind routing keys, publish messages, and consume them via tasks or triggers (batch or real time), plus AMQP 1.0 tasks for brokers such as Azure Service Bus and ActiveMQ Artemis.

DataInfrastructure

10 tasks

7 blueprints