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"}
    #      ]
    #    }
    #    
    # When set to "true", registers the user-provided schema as-is with the Pulsar Schema Registry instead of wrapping it in a CloudEvents envelope schema. Applies to both Avro (.avroschema) and JSON (.jsonschema) schema topics. Callers must also publish messages with rawPayload=true to send events without a CloudEvents envelope. Use this for topics that are intended to exclusively receive raw domain payloads. (Optional)
    #- name: <topic-name>.rawschema
    #  value: "false"
    # 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)"
    # Sets the compression level when compressionType is enabled. Higher compression levels provide better compression ratios but require more CPU resources. (Optional)
    #- name: compressionLevel
    #  value: "default"
    # Sets the compression type for messages sent by the producer. Compression can help reduce message size and improve throughput for large messages. (Optional)
    #- name: compressionType
    #  value: "none"
    # 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: ""
    # Configures the listener name the Pulsar client sends to the broker for connection redirects. Brokers with `advertisedListeners` resolve the redirect target using the matching named listener, enabling cross-network topologies (multi-cluster, internal/external endpoints) without a Pulsar proxy. (Optional)
    #- name: listenerName
    #  value: "external"
    # 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-----
    # Controls whether messages are processed synchronously or asynchronously. In sync mode, messages are handled one at a time in order. In async mode, up to maxConcurrentHandlers messages are processed concurrently. Can be overridden per subscription via subscription metadata. (Optional)
    #- name: processMode
    #  value: "async"
    # 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-----
    # Sets the size of the consumer receive queue. Controls how many messages can be accumulated by the consumer before it is explicitly called to read messages by Dapr. (Optional)
    #- name: receiverQueueSize
    #  value: "1000"
    # Specifies the delay after which to redeliver the messages that failed to be processed. (Optional)
    #- name: redeliveryDelay
    #  value: "30s"
    # Enable replication of subscription state across geo-replicated Pulsar clusters. When enabled, subscription state (such as cursor positions and acknowledgments) will be replicated to other clusters in a geo-replicated setup. This is useful for maintaining subscription consistency during cluster failovers. (Optional)
    #- name: replicateSubscriptionState
    #  value: "false"
    # Subscription position is the initial position which the cursor is set when start consuming: "latest", "earliest". (Optional)
    #- name: subscribeInitialPosition
    #  value: "latest"
    # Subscription mode indicates the cursor belongs to "durable" type or "non_durable" type, durable subscription retains messages and persists the current position. (Optional)
    #- name: subscribeMode
    #  value: "durable"
    # Pulsar supports four subscription types:"shared", "exclusive", "failover", "key_shared". (Optional)
    #- name: subscribeType
    #  value: "shared"
    # 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.

oauth2ClientSecretPath (string)​

The path to a plain text file containing the OAuth Client Secret.

Example value: /path/to/oauth2/client_secret.txt

oauth2CredentialsFile (string)​

The path to a JSON file containing both client_id and client_secret.

Example value: /path/to/oauth2/credentials.json

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"}
  ]
}

<topic-name>.rawschema (bool)​

When set to "true", registers the user-provided schema as-is with the Pulsar Schema Registry instead of wrapping it in a CloudEvents envelope schema. Applies to both Avro (.avroschema) and JSON (.jsonschema) schema topics. Callers must also publish messages with rawPayload=true to send events without a CloudEvents envelope. Use this for topics that are intended to exclusively receive raw domain payloads.

Default value: false

Example value: true, false

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

compressionLevel (string)​

Sets the compression level when compressionType is enabled. Higher compression levels provide better compression ratios but require more CPU resources.

Default value: default

Example value: faster

Allowed values:

  • default

  • faster

  • better

compressionType (string)​

Sets the compression type for messages sent by the producer. Compression can help reduce message size and improve throughput for large messages.

Default value: none

Example value: lz4

Allowed values:

  • none

  • lz4

  • zlib

  • zstd

Pulsar Message Compression: https://pulsar.apache.org/docs/3.0.x/concepts-messaging/#compression

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.

listenerName (string)​

Configures the listener name the Pulsar client sends to the broker for connection redirects. Brokers with advertisedListeners resolve the redirect target using the matching named listener, enabling cross-network topologies (multi-cluster, internal/external endpoints) without a Pulsar proxy.

Example value: external

Pulsar Multiple Advertised Listeners: https://pulsar.apache.org/docs/3.0.x/concepts-multiple-advertised-listeners/

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

processMode (string)​

Controls whether messages are processed synchronously or asynchronously. In sync mode, messages are handled one at a time in order. In async mode, up to maxConcurrentHandlers messages are processed concurrently. Can be overridden per subscription via subscription metadata.

Default value: async

Example value: sync

Allowed values:

  • sync

  • async

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

receiverQueueSize (number)​

Sets the size of the consumer receive queue. Controls how many messages can be accumulated by the consumer before it is explicitly called to read messages by Dapr.

Default value: 1000

Example value: 1000

redeliveryDelay (duration)​

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

Default value: 30s

Example value: 30s

replicateSubscriptionState (bool)​

Enable replication of subscription state across geo-replicated Pulsar clusters. When enabled, subscription state (such as cursor positions and acknowledgments) will be replicated to other clusters in a geo-replicated setup. This is useful for maintaining subscription consistency during cluster failovers.

Default value: false

Example value: true, false

Pulsar Geo-Replication: https://pulsar.apache.org/docs/administration-geo/

subscribeInitialPosition (string)​

Subscription position is the initial position which the cursor is set when start consuming: "latest", "earliest".

Default value: latest

Example value: earliest

Pulsar SubscriptionInitialPosition: https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionInitialPosition

subscribeMode (string)​

Subscription mode indicates the cursor belongs to "durable" type or "non_durable" type, durable subscription retains messages and persists the current position.

Default value: durable

Example value: durable

Pulsar SubscriptionMode: https://pkg.go.dev/github.com/apache/pulsar-client-go/pulsar#SubscriptionMode

subscribeType (string)​

Pulsar supports four subscription types:"shared", "exclusive", "failover", "key_shared".

Default value: shared

Example value: exclusive

Pulsar Subscription Types: https://pulsar.apache.org/docs/3.0.x/concepts-messaging/#subscription-types

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

Last updated on