= pulsar
:type: input
:status: experimental
:categories: ["Services"]



////
     THIS FILE IS AUTOGENERATED!

     To make changes, edit the corresponding source file under:

     https://github.com/redpanda-data/connect/tree/main/internal/impl/<provider>.

     And:

     https://github.com/redpanda-data/connect/tree/main/cmd/tools/docs_gen/templates/plugin.adoc.tmpl
////

// © 2024 Redpanda Data Inc.


component_type_dropdown::[]


Reads messages from an Apache Pulsar server.

Introduced in version 3.43.0.


[tabs]
======
Common::
+
--

```yml
# Common config fields, showing default values
input:
  label: ""
  pulsar:
    url: pulsar://localhost:6650 # 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: ""
```

--
Advanced::
+
--

```yml
# All config fields, showing default values
input:
  label: ""
  pulsar:
    url: pulsar://localhost:6650 # 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:

```text
- 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
xref:configuration:interpolation.adoc#bloblang-queries[function interpolation].


== Fields

=== `url`

A URL to connect to.


*Type*: `string`


```yml
# Examples

url: pulsar://localhost:6650

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

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

=== `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`


=== `subscription_name`

Specify the subscription name for this consumer.


*Type*: `string`


=== `subscription_type`

Specify the subscription type for this consumer.

> NOTE: 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 https://pulsar.apache.org/docs/en/2.8.1/concepts-messaging/#negative-acknowledgement[Pulsar documentation^] and https://github.com/apache/pulsar/issues/12208[this Github issue^] for more details.


*Type*: `string`

*Default*: `"shared"`

Options:
`shared`
, `key_shared`
, `failover`
, `exclusive`
.

=== `subscription_initial_position`

Specify the subscription initial position for this consumer.


*Type*: `string`

*Default*: `"latest"`

Options:
`latest`
, `earliest`
.

=== `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*: `""`

```yml
# Examples

root_cas_file: ./root_cas.pem
```

=== `auth`

Optional configuration of Pulsar authentication methods.


*Type*: `object`

Requires version 3.60.0 or newer

=== `auth.oauth2`

Parameters for Pulsar OAuth2 authentication.


*Type*: `object`


=== `auth.oauth2.enabled`

Whether OAuth2 is enabled.


*Type*: `bool`

*Default*: `false`

=== `auth.oauth2.audience`

OAuth2 audience.


*Type*: `string`

*Default*: `""`

=== `auth.oauth2.issuer_url`

OAuth2 issuer URL.


*Type*: `string`

*Default*: `""`

=== `auth.oauth2.scope`

OAuth2 scope to request.


*Type*: `string`

*Default*: `""`

=== `auth.oauth2.private_key_file`

The path to a file containing a private key.


*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*: `""`