PLUGINSPLUGIN-PULSARTASKSIO.KESTRA.PLUGIN.PULSAR.CONSUME

Consume

yaml

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

Consume messages from Pulsar topic(s)

Examples

yaml

id: "consume"
type: "io.kestra.plugin.pulsar.Consume"
uri: pulsar://localhost:26650
topic: test_kestra
deserializer: JSON
subscriptionName: kestra_flow

Properties

`deserializer`

Type: object
Dynamic: ❓
Required: ✔️

`initialPosition`

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

Add all the properties in the provided map to the consumer.

`pollDuration`

Type: string
Dynamic: ✔️
Required: ✔️
Default: 2.000000000
Format: duration

Duration waiting for record to be polled

If no records are available, the max wait to wait for a new records.

`subscriptionName`

Type: string
Dynamic: ✔️
Required: ✔️

The subscription name

Using subscription name, we will fetch only records not already consumed

`subscriptionType`

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

Add all the properties in the provided map to the consumer.

`topic`

Type: object
Dynamic: ✔️
Required: ✔️

Pulsar topic(s) where to consume message

Can be a string or a List of string to consume from multiple topic

`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:6550,localhost:6651,localhost:6652
If you use TLS authentication: pulsar+ssl://pulsar.us-west.example.com:6651
ssl.truststore.location