Produce

yaml

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

Produce message to a Pulsar topic.

Examples

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.scripts.nashorn.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`

Type: object
Dynamic: ✔️
Required: ✔️

Source of the sent message.

Can be a Kestra internal storage URI, a map or a list in the following format: key, value, eventTime, properties, deliverAt, deliverAfter and sequenceId.

`serializer`

Type: object
Dynamic: ❓
Required: ✔️

`topic`

Type: string
Dynamic: ✔️
Required: ✔️

Pulsar topic to send a message to.

`uri`

Type: string
Dynamic: ✔️
Required: ✔️

Connection URLs.

You need to specify a Pulsar protocol URL.

Example of localhost: pulsar://localhost:6650
If you have multiple brokers: pulsar://localhost:6650,localhost:6651,localhost:6652
If you use TLS authentication: pulsar+ssl://pulsar.us-west.example.com:6651

`accessMode`

Type: string
Dynamic: ❌
Required: ❌
Possible Values:
- Shared
- Exclusive
- ExclusiveWithFencing
- WaitForExclusive

Configure the type of access mode that the producer requires on the topic.

Possible values are:

Shared: By default, multiple producers can publish to a topic.
Exclusive: Require exclusive access for producer. Fail immediately if there's already a producer connected.
WaitForExclusive: Producer creation is pending until it can acquire exclusive access.