Skip to main content

Apache Pulsar

Type: pubsub.pulsar

Status: stable

Reference: https://docs.dapr.io/reference/components-reference/supported-pubsub/setup-pulsar/

Example​

apiVersion: cra.diagrid.io/v1beta1
kind: Component
metadata:
  name: <name>
spec:
  type: pubsub.pulsar
  version: v1
  metadata:
    # Shared JWT token for authentication (Optional)
    #- name: token
    #  value: ""
    # Address of the Pulsar broker.
    - name: host
      value: "mycompany.com:6650, http://pulsar-pj54qwwdpz4b-pulsar.ap-sg.public.pulsar.com:8080"
    # Enforces Avro schema validation for the configured topic. The value is a string containing a JSON object. (Optional)
    #- name: <topic-name>.avroschema
    #  value: | 
    #    {
    #      "type": "record",
    #      "name": "Example",
    #      "namespace": "test",
    #      "fields": [
    #        {"name": "ID","type": "int"},
    #        {"name": "Name","type": "string"}
    #      ]
    #    }
    #    
    # Enforces JSON schema validation for the configured topic. The value is a string containing a JSON object. (Optional)
    #- name: <topic-name>.jsonschema
    #  value: | 
    #    {
    #      "type": "record",
    #      "name": "Example",
    #      "namespace": "test",
    #      "fields": [
    #        {"name": "ID","type": "int"},
    #        {"name": "Name","type": "string"}
    #      ]
    #    }
    #    
    # Sets the maximum number of messages permitted in a batch. If set to a value greater than 1, messages will be queued until this threshold is reached, batchingMaxSize has been reached, or the batch interval has elapsed. (Optional)
    #- name: batchingMaxMessages
    #  value: "1000"
    # If batching is enabled, this sets the time window within messages are batched. If set to a non-zero value, messages will be queued until this time interval has passed, or the batchingMaxMessages or batchingMaxSize conditions have been met. (Optional)
    #- name: batchingMaxPublishDelay
    #  value: "10ms"
    # Sets the maximum number of bytes permitted in a batch If set to a value greater than 1, messages will be queued until this threshold is reached, batchingMaxMessages has been reached, or the batch interval has elapsed. (Optional)
    #- name: batchingMaxSize
    #  value: "131072" (128 KB)"
    # Used to set the subscription name or consumer ID. (Optional)
    #- name: consumerID
    #  value: "channel1"
    # When enabled, the producer will send messages in a batch. (Optional)
    #- name: disableBatching
    #  value: "false"
    # Enable TLS. (Optional)
    #- name: enableTLS
    #  value: "false"
    # A comma delimited string containing names of Pulsar session keys. Used in conjunction with publicKey for publisher encryption. (Optional)
    #- name: keys
    #  value: ""
    # Sets the maximum number of concurrent messages sent to the application. Default is 100. (Optional)
    #- name: maxConcurrentHandlers
    #  value: "100"
    # The administrative unit of the topic, which acts as a grouping mechanism for related topics. (Optional)
    #- name: namespace
    #  value: "default"
    # Pulsar supports two kinds of topics: persistent and non-persistent. With persistent topics, all messages are durably persisted on disks (if the broker is not standalone, messages are durably persisted on multiple disks), whereas data for non-persistent topics is not persisted to storage disks. (Optional)
    #- name: persistent
    #  value: "true"
    # A private key to be used for consumer encryption. Value can be one of two options: file path for a local PEM cert, or the cert data string value. (Optional)
    #- name: privateKey
    #  value: | 
    #    -----BEGIN PRIVATE KEY-----
    #    <base64-encoded DER>
    #    -----END PRIVATE KEY-----
    # A public key to be used for publisher and consumer encryption. Value can be one of two options: file path for a local PEM cert, or the cert data string value. (Optional)
    #- name: publicKey
    #  value: | 
    #    -----BEGIN PUBLIC KEY-----
    #    <base64-encoded DER>
    #    -----END PUBLIC KEY-----
    # Specifies the delay after which to redeliver the messages that failed to be processed. (Optional)
    #- name: redeliveryDelay
    #  value: "30s"
    # The topic tenant within the instance. Tenants are essential to multi-tenancy in Pulsar, and spread across clusters. (Optional)
    #- name: tenant
    #  value: "public"

Authentication profiles​

Available authentication profiles:

  • Shared token

  • OAuth2

Shared token​

Authenticate using a shared token

token (string)​

Shared JWT token for authentication

JWT Token Authentication: https://pulsar.apache.org/docs/3.0.x/security-jwt/#generate-tokens

OAuth2​

Authenticate using OAuth2 or OpenID Connect

oauth2Audiences (string)​

The OAuth 2.0 "resource server" identifier for a Pulsar cluster.

oauth2ClientID (string)​

The OAuth Client ID.

oauth2ClientSecret (string)​

The OAuth Client Secret.

oauth2Scopes (string)​

The scope of an access request. For more information, see Access Token Scope.

Access Token Scope: https://datatracker.ietf.org/doc/html/rfc6749#section-3.3

oauth2TokenCAPEM (string)​

The OAuth Token Certificate Authority PEM.

oauth2TokenURL (string)​

The OAuth Client URL.

Metadata​

host (string)​

Required - Address of the Pulsar broker.

Example value: mycompany.com:6650, http://pulsar-pj54qwwdpz4b-pulsar.ap-sg.public.pulsar.com:8080

<topic-name>.avroschema (string)​

Enforces Avro schema validation for the configured topic. The value is a string containing a JSON object.

Example value: `{
  "type": "record",
  "name": "Example",
  "namespace": "test",
  "fields": [
    {"name": "ID","type": "int"},
    {"name": "Name","type": "string"}
  ]
}
`

<topic-name>.jsonschema (string)​

Enforces JSON schema validation for the configured topic. The value is a string containing a JSON object.

Example value: `{
  "type": "record",
  "name": "Example",
  "namespace": "test",
  "fields": [
    {"name": "ID","type": "int"},
    {"name": "Name","type": "string"}
  ]
}
`

batchingMaxMessages (number)​

Sets the maximum number of messages permitted in a batch. If set to a value greater than 1, messages will be queued until this threshold is reached, batchingMaxSize has been reached, or the batch interval has elapsed.

Default value: 1000

Example value: 1000

batchingMaxPublishDelay (duration)​

If batching is enabled, this sets the time window within messages are batched. If set to a non-zero value, messages will be queued until this time interval has passed, or the batchingMaxMessages or batchingMaxSize conditions have been met.

Default value: 10ms

Example value: 10ms

batchingMaxSize (number)​

Sets the maximum number of bytes permitted in a batch If set to a value greater than 1, messages will be queued until this threshold is reached, batchingMaxMessages has been reached, or the batch interval has elapsed.

Default value: 131072" (128 KB)

Example value: 131072

consumerID (string)​

Used to set the subscription name or consumer ID.

Example value: channel1

disableBatching (bool)​

When enabled, the producer will send messages in a batch.

Default value: false

Example value: true, false

enableTLS (bool)​

Enable TLS.

Default value: false

Example value: true, false

keys (string)​

A comma delimited string containing names of Pulsar session keys. Used in conjunction with publicKey for publisher encryption.

maxConcurrentHandlers (number)​

Sets the maximum number of concurrent messages sent to the application. Default is 100.

Default value: 100

Example value: 100

namespace (string)​

The administrative unit of the topic, which acts as a grouping mechanism for related topics.

Default value: default

Example value: default

persistent (bool)​

Pulsar supports two kinds of topics: persistent and non-persistent. With persistent topics, all messages are durably persisted on disks (if the broker is not standalone, messages are durably persisted on multiple disks), whereas data for non-persistent topics is not persisted to storage disks.

Default value: true

Example value: true, false

Pulsar Persistent Storage: https://pulsar.apache.org/docs/3.0.x/concepts-architecture-overview/#persistent-storage

privateKey (string)​

A private key to be used for consumer encryption. Value can be one of two options: file path for a local PEM cert, or the cert data string value.

Example value:

-----BEGIN PRIVATE KEY-----\n<base64-encoded DER>\n-----END PRIVATE KEY-----

publicKey (string)​

A public key to be used for publisher and consumer encryption. Value can be one of two options: file path for a local PEM cert, or the cert data string value.

Example value:

-----BEGIN PUBLIC KEY-----\n<base64-encoded DER>\n-----END PUBLIC KEY-----

redeliveryDelay (duration)​

Specifies the delay after which to redeliver the messages that failed to be processed.

Default value: 30s

Example value: 30s

tenant (string)​

The topic tenant within the instance. Tenants are essential to multi-tenancy in Pulsar, and spread across clusters.

Default value: public

Example value: public