Kafka Produce

Publishes the incoming event to a Kafka topic and emits the delivery result (topic, partition, offset) downstream.

Configuration

- kafka_produce:
    name: publish_customer
    credentials_path: /etc/kafka/credentials.json
    brokers: "{{env.KAFKA_BROKERS}}"
    topic: customers
    message_key: "customer-{{event.data.name}}"
    create_or_update: true

Fields

Field	Type	Default	Description
`name`	string	required	Task name.
`credentials_path`	string		Path to Kafka credentials file. Omit to connect without authentication. See Credentials.
`brokers`	string	`localhost:9092`	Comma-separated bootstrap broker addresses.
`topic`	string	required	Topic to publish to. Supports templating.
`message_key`	string		Message key template (e.g. `key-{{event.id}}`). See Templating.
`create_or_update`	bool	`false`	When `true`, the topic is created from `topic_options` if it does not exist. When `false`, an error is returned if the topic is absent from the cluster.
`topic_options`	object		Settings for a topic created by `create_or_update`. See Topic creation.
`ack_timeout`	duration	`30s`	How long to wait for the broker to acknowledge a message before the attempt fails. Must be greater than zero.
`depends_on`	list		Upstream task names.
`retry`	object		Retry configuration.

Output

Format: JSON

Field	Type	Description
`topic`	string	Topic the message was written to.
`partition`	int	Partition the message was written to.
`offset`	int	Offset of the written message.

Incoming events do not always carry an id (e.g. NATS messages without a Nats-Msg-Id header, generate tasks). The producer patches a UUID v7 fallback into the render context, so {{event.id}} in message_key always resolves to a value:

- kafka_produce:
    name: publish_orders
    credentials_path: /etc/kafka/credentials.json
    topic: orders
    message_key: "{{event.id}}"

Topic creation

With create_or_update: true, a topic that does not exist is created from topic_options:

- kafka_produce:
    name: publish_orders
    topic: orders
    create_or_update: true
    topic_options:
      partitions: 6
      replication_factor: 3
      retention: 7d
      config:
        cleanup.policy: delete

Field	Type	Default	Description
`partitions`	int	`1`	Number of partitions.
`replication_factor`	int	`1`	Replication factor. Cannot exceed the number of brokers.
`retention`	duration		How long the topic retains a message. Sets `retention.ms`.
`config`	map		Any other topic-level setting, passed to the broker verbatim. A key here overrides the equivalent typed field.

The defaults suit a local broker only — a single partition caps throughput at one consumer, and no replication loses the topic with its broker. Set both explicitly for anything else.

topic_options applies only when the topic is created. An existing topic is left as it is, so changing these values does not reshape a live topic.

On a cluster with auto.create.topics.enable (the broker default), a topic is otherwise created on first use with the broker’s own defaults. The task checks for the topic in the metadata of every topic, which does not trigger that, so topic_options is what the topic is actually created with.

Behaviour

The message payload is the incoming event’s data — JSON is serialized as-is, bytes/Avro payloads are sent raw, and Arrow record batches are serialized as an Arrow IPC stream. Delivery is acknowledged before the result event is emitted downstream; if the broker does not acknowledge within ack_timeout, the task fails and retries per the retry configuration.

A message with a message_key goes to the partition the Kafka Java client’s default partitioner chooses for that key (murmur2 hash). A message without message_key is sent with an empty key to the topic’s partitions in turn. The topic’s partitions are looked up again every five minutes, so partitions added to the topic are used.

Errors that cannot succeed on a second attempt — a missing topic with create_or_update: false, a template that fails to render, a payload that cannot be serialized — fail immediately instead of consuming the retry budget.