PluginsApache PulsarTriggersRealtimeTrigger

RealtimeTrigger

yaml

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

Consume a message in real-time from Pulsar topics and create one execution per message.

If you would like to consume multiple messages processed within a given time frame and process them in batch, you can use the io.kestra.plugin.pulsar.Trigger instead.

Examples

Consume a message from a Pulsar topic in real-time.

yaml

id: pulsar
namespace: company.team

tasks:
  - id: log
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.value }}"

triggers:
  - id: realtime_trigger
    type: io.kestra.plugin.pulsar.RealtimeTrigger
    topic: kestra_trigger
    uri: pulsar:https://localhost:26650
    deserializer: JSON
    subscriptionName: kestra_trigger_sub

Properties

`deserializer`

Type: object
Dynamic: ❓
Required: ✔️

`initialPosition`

Type: string
Dynamic: ❌
Required: ✔️
Default: Earliest
Possible Values:
- Latest
- Earliest

The position of a subscription to the topic.

`subscriptionName`

Type: string
Dynamic: ✔️
Required: ✔️

The subscription name.

Using subscription name, we will fetch only records that haven't been consumed yet.

`subscriptionType`

Type: string
Dynamic: ❌
Required: ✔️
Default: Exclusive
Possible Values:
- Exclusive
- Shared
- Failover
- Key_Shared

The subscription type.

`topic`

Type: object
Dynamic: ✔️
Required: ✔️

Pulsar topic(s) where to consume messages from.

Can be a string or a list of strings to consume from multiple topics.

`uri`

Type: string
Dynamic: ✔️
Required: ✔️

Connection URLs.

You need to specify a Pulsar protocol URL.

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