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

Examples

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

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.

from stringarrayobject

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.

keyAvroSchema string

Avro schema when the key serializer is AVRO

Required if keySerializer is AVRO.

keySerializer string

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.

pluginDefaultsRef string

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

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.

topic string

Target Kafka topic when not provided in payload

Can be overridden per record by setting topic inside from.

transactional booleanstring

Defaulttrue

Enable transactional producer

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

valueAvroSchema string

Avro schema when the value serializer is AVRO

Required if valueSerializer is AVRO.

valueSerializer string

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.