Apache Pulsar Produce

Apache Pulsar Produce

Certified

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.

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

Read a CSV file, transform it to the right format, and publish it to Pulsar topic.

yaml
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

Message source

Kestra internal storage URI, or map/list objects with optional key, value, eventTime, properties, deliverAt, deliverAfter, and sequenceId fields.

DefaultSTRING
Possible Values
STRINGJSONBYTES

Serializer for message value

Defaults to STRING. Choose a serializer compatible with consumers and topic schema.

Target Pulsar topic

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.

Possible Values
SharedExclusiveExclusiveWithFencingWaitForExclusive

Producer access mode

Shared (default Pulsar behavior) allows multiple producers; Exclusive fails if another producer is connected; WaitForExclusive waits for exclusivity.

Authentication token

Token used when the broker requires token-based auth (e.g., hosted providers).

Possible Values
NONELZ4ZLIBZSTDSNAPPY

Producer compression type

Default NONE. Other options: LZ4, ZLIB, ZSTD, SNAPPY. Use to reduce payload size at the cost of CPU.

Public encryption key

PEM-encoded key used to encrypt the data key for message payload encryption.

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

Custom producer name

Optional name reused on reconnects; may affect exclusive access checks.

Producer properties

Key/value properties passed to the Pulsar producer builder.

Topic schema definition

JSON representation of the topic schema when schema enforcement is enabled.

DefaultNONE
Possible Values
NONEAVROJSON

Topic schema type

One of NONE (default, no enforcement), AVRO, or JSON.

TLS options

Certificate/key material for TLS client authentication. Requires a pulsar+ssl:// URL.

Definitions
castring

CA certificate

Base64-encoded PEM of the trusted CA chain.

certstring

Client certificate

Base64-encoded PEM content for the client certificate.

keystring

Client key

Base64-encoded PEM private key matching the client certificate.

Number of messages produced

More Plugins in this Category

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

PostgreSQL
PostgreSQL

Tasks that connect to PostgreSQL via JDBC to run queries and loads.

Data

6 tasks

25 blueprints