# 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: 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: .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: .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: .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----- # # -----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----- # # -----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` ### `.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"} ] } ``` ### `.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"} ] } ``` ### `.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\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\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`