Apache Pulsar Produce
CertifiedPublish messages to a Pulsar topic
Apache Pulsar Produce
Publish messages to a Pulsar topic
Reads records from Kestra storage or inline maps/lists and sends them with a Pulsar producer. Uses STRING serialization and no compression by default; Pulsar's shared producer mode applies unless another access mode is set.
type: io.kestra.plugin.pulsar.ProduceExamples
Read a CSV file, transform it to the right format, and publish it to Pulsar topic.
id: produce
namespace: company.team
inputs:
- type: FILE
id: file
tasks:
- id: csv_reader
type: io.kestra.plugin.serdes.csv.CsvToIon
from: "{{ inputs.file }}"
- id: file_transform
type: io.kestra.plugin.graalvm.js.FileTransform
from: "{{ outputs.csv_reader.uri }}"
script: |
var result = {
"key": row.id,
"value": {
"username": row.username,
"tweet": row.tweet
},
"eventTime": row.timestamp,
"properties": {
"key": "value"
}
};
row = result
- id: produce
type: io.kestra.plugin.pulsar.Produce
from: "{{ outputs.file_transform.uri }}"
uri: pulsar://localhost:26650
serializer: JSON
topic: test_kestra
Properties
from *object
Message source
Kestra internal storage URI, or map/list objects with optional key, value, eventTime, properties, deliverAt, deliverAfter, and sequenceId fields.
serializer *string
STRINGSTRINGJSONBYTESSerializer for message value
Defaults to STRING. Choose a serializer compatible with consumers and topic schema.
topic *string
Target Pulsar topic
uri *string
Pulsar service URL
One or more Pulsar protocol URLs, e.g. pulsar://localhost: 6650 or pulsar://host1: 6650,host2: 6651. Use pulsar+ssl:// when enabling TLS.
accessMode string
SharedExclusiveExclusiveWithFencingWaitForExclusiveProducer access mode
Shared (default Pulsar behavior) allows multiple producers; Exclusive fails if another producer is connected; WaitForExclusive waits for exclusivity.
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.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset 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.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150authenticationToken string
Authentication token
Token used when the broker requires token-based auth (e.g., hosted providers).
compressionType string
NONELZ4ZLIBZSTDSNAPPYProducer compression type
Default NONE. Other options: LZ4, ZLIB, ZSTD, SNAPPY. Use to reduce payload size at the cost of CPU.
encryptionKey string
Public encryption key
PEM-encoded key used to encrypt the data key for message payload encryption.
producerName string
Custom producer name
Optional name reused on reconnects; may affect exclusive access checks.
producerProperties object
Producer properties
Key/value properties passed to the Pulsar producer builder.
schemaString string
Topic schema definition
JSON representation of the topic schema when schema enforcement is enabled.
schemaType string
NONENONEAVROJSONTopic schema type
One of NONE (default, no enforcement), AVRO, or JSON.
tlsOptions
TLS options
Certificate/key material for TLS client authentication. Requires a pulsar+ssl:// URL.
io.kestra.plugin.pulsar.AbstractPulsarConnection-TlsOptions
CA certificate
Base64-encoded PEM of the trusted CA chain.
Client certificate
Base64-encoded PEM content for the client certificate.
Client key
Base64-encoded PEM private key matching the client certificate.
Outputs
messagesCount integer
Number of messages produced
