Apache Kafka TopicCreate

Apache Kafka TopicCreate
Certified
Create a Kafka topic

Creates a topic with the given partition count and replication factor using the Kafka AdminClient. Fails with TopicExistsException if the topic already exists, unless ifNotExists is set to true.

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

Examples

Provision a topic for a tenant namespace

yaml
id: kafka_topic_create
namespace: company.team

tasks:
  - id: create_topic
    type: io.kestra.plugin.kafka.TopicCreate
    properties:
      bootstrap.servers: localhost:9092
    topic: tenant_acme_orders
    partitions: 6
    replicationFactor: 3
    ifNotExists: true
    configs:
      retention.ms: "604800000"

Properties

partitions *integerstring

Number of partitions

No default is provided: choose an explicit partition count based on expected throughput.

properties *object

Kafka AdminClient properties

Must include bootstrap.servers; accepts any Kafka AdminClient config. Provide base64-encoded content for ssl.keystore.location and ssl.truststore.location when using SSL.

topic *string

Topic name

assets

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

io.kestra.core.models.assets.AssetsDeclaration

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.

io.kestra.core.models.assets.AssetIdentifier

id*string

Min length1

typestring

outputs

The assets produced as outputs.

io.kestra.plugin.ee.assets.Dataset

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.plugin.ee.assets.File

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.plugin.ee.assets.Table

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.plugin.ee.assets.VM

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.core.models.assets.External

id*string

Min length1

Max length150

type*object

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

io.kestra.core.models.assets.Custom

id*string

Min length1

Max length150

type*string

Min length1

Custom asset type

descriptionstring

displayNamestring

metadataobject

Default{}

namespacestring

Min length1

Max length150

callTimeout string

DefaultPT30S

AdminClient call timeout

Maximum duration to wait for each AdminClient operation to complete before failing the task. Defaults to PT30S (30 seconds). Distinct from the task-level timeout, which lets the worker kill the task without retrying it.

configs object

Default{}

Additional topic-level configs

For example retention.ms, retention.bytes, cleanup.policy, min.insync.replicas.

ifNotExists booleanstring

Defaultfalse

Don't fail when the topic already exists

When true, an existing topic is left untouched and the task succeeds, matching kafka-topics.sh --if-not-exists semantics. Defaults to false.

replicationFactor integerstring

Default1

Replication factor

Defaults to 1, which is unsuitable for production — use at least 3 on a production cluster for durability.

Outputs

created boolean

Whether the topic was created by this task run

false when ifNotExists was true and the topic already existed — partitions/replicationFactor then reflect the existing topic, not the requested values.