Connect
Self-Managed Only

pulsar

Reads messages from an Apache Pulsar server.

Introduced in version 3.43.0.

  • Common

  • Advanced

inputs:
  label: ""
  pulsar:
    url: "" # No default (required)
    topics: [] # No default (optional)
    topics_pattern: "" # No default (optional)
    subscription_name: "" # No default (required)
    subscription_type: shared
    subscription_initial_position: latest
    tls:
      root_cas_file: ""
inputs:
  label: ""
  pulsar:
    url: "" # No default (required)
    topics: [] # No default (optional)
    topics_pattern: "" # No default (optional)
    subscription_name: "" # No default (required)
    subscription_type: shared
    subscription_initial_position: latest
    tls:
      root_cas_file: ""
    auth:
      oauth2:
        enabled: false
        audience: ""
        issuer_url: ""
        scope: ""
        private_key_file: ""
      token:
        enabled: false
        token: ""

Metadata

This input adds the following metadata fields to each message:

  • pulsar_message_id

  • pulsar_key

  • pulsar_ordering_key

  • pulsar_event_time_unix

  • pulsar_publish_time_unix

  • pulsar_topic

  • pulsar_producer_name

  • pulsar_redelivery_count

  • All properties of the message

You can access these metadata fields using function interpolation.

Fields

auth

Optional configuration of Pulsar authentication methods.

Requires version 3.60.0 or later.

Type: object

auth.oauth2

Parameters for Pulsar OAuth2 authentication.

Type: object

auth.oauth2.audience

OAuth2 audience.

Type: string

Default: ""

auth.oauth2.enabled

Whether OAuth2 is enabled.

Type: bool

Default: false

auth.oauth2.issuer_url

OAuth2 issuer URL.

Type: string

Default: ""

auth.oauth2.private_key_file

The path to a file containing a private key.

Type: string

Default: ""

auth.oauth2.scope

OAuth2 scope to request.

Type: string

Default: ""

auth.token

Parameters for Pulsar Token authentication.

Type: object

auth.token.enabled

Whether Token Auth is enabled.

Type: bool

Default: false

auth.token.token

Actual base64 encoded token.

Type: string

Default: ""

subscription_initial_position

Specify the subscription initial position for this consumer.

Type: string

Default: latest

Options: latest, earliest

subscription_name

Specify the subscription name for this consumer.

Type: string

subscription_type

Specify the subscription type for this consumer.

Using a key_shared subscription type will allow out-of-order delivery since nack-ing messages sets non-zero nack delivery delay - this can potentially cause consumers to stall. See Pulsar documentation and this Github issue for more details.

Type: string

Default: shared

Options: shared, key_shared, failover, exclusive

tls

Specify the path to a custom CA certificate to trust broker TLS service.

Type: object

tls.root_cas_file

An optional path of a root certificate authority file to use. This is a file, often with a .pem extension, containing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Type: string

Default: ""

# Examples:
root_cas_file: ./root_cas.pem

topics[]

A list of topics to subscribe to. This or topics_pattern must be set.

Type: array

topics_pattern

A regular expression matching the topics to subscribe to. This or topics must be set.

Type: string

url

A URL to connect to.

Type: string

# Examples:
url: pulsar://localhost:6650

# ---

url: pulsar://pulsar.us-west.example.com:6650

# ---

url: pulsar+ssl://pulsar.us-west.example.com:6651