Connect
Self-Managed Only

pulsar

Reads messages from an Apache Pulsar server.

Introduced in version 3.43.0.

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.

  • Common

  • Advanced

input:
  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: ""
input:
  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: ""

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.

Requires version 4.29.0 or later.

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.

Requires version 4.29.0 or later.

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 the broker TLS service.

Requires version 4.2.0 or later.

Type: object

tls.root_cas_file

The path of a PEM file containing the certificate authority certificates that the client trusts when it verifies the TLS certificate of the broker. Leave empty to trust the certificate authorities of the host system.

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<string>

topics_pattern

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

Requires version 4.19.0 or later.

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