Pulsar configuration

Pulsar 配置可以通过 Pulsar 安装 目录下的 conf 文件夹里的一系列配置文件进行管理。

• BookKeeper
• Broker
• Client
• 服务发现
• Log4j
• Log4j shell
• Standalone
• WebSocket
• Pulsar proxy
• ZooKeeper

BookKeeper

BookKeeper 是一个冗余的日志存储系统，Pulsar 用它来持久化存储所有消息。

Broker

Pulsar broker 负责处理从生产者发出消息、向消费者派发消息、在集群间复制数据等。

<th>
  Description
</th>
<th>
  默认值
</th>

<td>
  Specify multiple advertised listeners for the broker.<br><br>The format is <code>&lt;listener_name&gt;:pulsar://&lt;host&gt;:&lt;port&gt;</code>.<br><br>If there are multiple listeners, separate them with commas.<br><br><strong x-id="1">Note</strong>: do not use this configuration with <code>advertisedAddress</code> and <code>brokerServicePort</code>. If the value of this configuration is empty, the broker uses <code>advertisedAddress</code> and <code>brokerServicePort</code>
</td>
<td>
  /
</td>

<td>
  Specify the internal listener name for the broker.<br><br><strong x-id="1">Note</strong>: the listener name must be contained in <code>advertisedListeners</code>.<br><br> If the value of this configuration is empty, the broker uses the first listener as the internal listener.
</td>
<td>
  /
</td>

<td>
  Whether persistent topics are enabled on the broker
</td>
<td>
  true
</td>

<td>
  Whether non-persistent topics are enabled on the broker
</td>
<td>
  true
</td>

<td>
  Whether the Pulsar Functions worker service is enabled in the broker
</td>
<td>
  false
</td>

<td>
  Zookeeper quorum connection string
</td>
<td>
</td>

<td>
  ZooKeeper 缓存过期时间（秒）
</td>
<td>
  300
</td>

<td>
  配置存储连接字符串（以逗号分隔的列表）
</td>
<td>
</td>

<td>
  Broker data port
</td>
<td>
  6650
</td>

<td>
  Broker data port for TLS
</td>
<td>
  6651
</td>

<td>
  Port to use to server HTTP request
</td>
<td>
  8080
</td>

<td>
  Port to use to server HTTPS request
</td>
<td>
  8443
</td>

<td>
  Enable the WebSocket API service in broker
</td>
<td>
  false
</td>

<td>
  Hostname or IP address the service binds on, default is 0.0.0.0.
</td>
<td>
  0.0.0.0
</td>

<td>
  Hostname or IP address the service advertises to the outside world. If not set, the value of <code>InetAddress.getLocalHost().getHostName()</code> is used.
</td>
<td>
</td>

<td>
  Name of the cluster to which this broker belongs to
</td>
<td>
</td>

<td>
  Sets the default behavior for message deduplication in the broker. If enabled, the broker will reject messages that were already stored in the topic. This setting can be overridden on a per-namespace basis.
</td>
<td>
  false
</td>

<td>
  The maximum number of producers for which information will be stored for deduplication purposes.
</td>
<td>
  10000
</td>

<td>
  去重快照生成后，允许存储的去重信息数量。 A larger interval will lead to fewer snapshots being taken, though this would also lengthen the topic recovery time (the time required for entries published after the snapshot to be replayed).
</td>
<td>
  1000
</td>

<td>
  The time of inactivity (in minutes) after which the broker will discard deduplication information related to a disconnected producer.
</td>
<td>
  360
</td>

<td>
  The default messages per second dispatch throttling-limit for every replicator in replication. The value of <code>0</code> means disabling replication message dispatch-throttling
</td>
<td>
  0
</td>

<td>
  The default bytes per second dispatch throttling-limit for every replicator in replication. The value of <code>0</code> means disabling replication message-byte dispatch-throttling
</td>
<td>
  0
</td>

<td>
  Zookeeper session timeout in milliseconds
</td>
<td>
  30000
</td>

<td>
  Time to wait for broker graceful shutdown. After this time elapses, the process will be killed
</td>
<td>
  60000
</td>

<td>
  Flag to skip broker shutdown when broker handles Out of memory error.
</td>
<td>
  false
</td>

<td>
  Enable backlog quota check. Enforces action on topic when the quota is reached
</td>
<td>
  true
</td>

<td>
  How often to check for topics that have reached the quota
</td>
<td>
  60
</td>

<td>
  The default per-topic backlog quota limit
</td>
<td>
  -1
</td>

<td>
  Enable topic auto creation if a new producer or consumer connected
</td>
<td>
  true
</td>

<td>
  The topic type (partitioned or non-partitioned) that is allowed to be automatically created.
</td>
<td>
  Partitioned
</td>

<td>
  Enable subscription auto creation if a new consumer connected
</td>
<td>
  true
</td>

<td>
  The number of partitioned topics that is allowed to be automatically created if <code>allowAutoTopicCreationType</code> is partitioned
</td>
<td>
  1
</td>

<td>
  Enable the deletion of inactive topics
</td>
<td>
  true
</td>

<td>
  How often to check for inactive topics
</td>
<td>
  60
</td>

<td>
  Set the mode to delete inactive topics. <li> <code>delete_when_no_subscriptions</code>: 删除没有订阅或活动生产者的主题。 <li> <code>delete_when_subscriptions_caught_up</code>: delete the topic whose subscriptions have no backlogs and which has no active producers or consumers.
</td>
<td>
  <code>delete_when_no_subscriptions</code>
</td>

<td>
  Set the maximum duration for inactive topics. If it is not specified, the <code>brokerDeleteInactiveTopicsFrequencySeconds</code> parameter is adopted.
</td>
<td>
  N/A
</td>

<td>
  How frequently to proactively check and purge expired messages
</td>
<td>
  5
</td>

<td>
  Interval between checks to see if topics with compaction policies need to be compacted
</td>
<td>
  60
</td>

<td>
  How long to delay rewinding cursor and dispatching messages when active consumer is changed.
</td>
<td>
  1000
</td>

<td>
  Enable check for minimum allowed client library version
</td>
<td>
  false
</td>

<td>
  Allow client libraries with no version information
</td>
<td>
  true
</td>

<td>
  Path for the file used to determine the rotation status for the broker when responding to service discovery health checks
</td>
<td>
</td>

<td>
  If true, (and ModularLoadManagerImpl is being used), the load manager will attempt to use only brokers running the latest software version (to minimize impact to bundles)
</td>
<td>
  false
</td>

<td>
  Max number of partitions per partitioned topic. Use 0 or negative number to disable the check
</td>
<td>
  0
</td>

<td>
  Enable TLS
</td>
<td>
  false
</td>

<td>
  TLS证书文件的路径
</td>
<td>
</td>

<td>
  TLS私钥文件的路径
</td>
<td>
</td>

<td>
  Path for the trusted TLS certificate file
</td>
<td>
</td>

<td>
  Accept untrusted TLS certificate from client
</td>
<td>
  false
</td>

<td>
  Specify the tls protocols the broker will use to negotiate during TLS Handshake. Multiple values can be specified, separated by commas. Example:- <code>TLSv1.2</code>, <code>TLSv1.1</code>, <code>TLSv1</code>
</td>
<td>
</td>

<td>
  Specify the tls cipher the broker will use to negotiate during TLS Handshake. Multiple values can be specified, separated by commas. Example:- <code>TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256</code>
</td>
<td>
</td>

<td>
  Enable TLS with KeyStore type configuration in broker
</td>
<td>
  false
</td>

<td>
  TLS Provider for KeyStore type
</td>
<td>
</td>

<td>
  LS KeyStore type configuration in broker: JKS, PKCS12
</td>
<td>
  JKS
</td>

<td>
  TLS KeyStore path in broker
</td>
<td>
</td>

<td>
  TLS KeyStore password for broker
</td>
<td>
</td>

<td>
  Whether internal client use KeyStore type to authenticate with Pulsar brokers
</td>
<td>
  false
</td>

<td>
  The TLS Provider used by internal client to authenticate with other Pulsar brokers
</td>
<td>
</td>

<td>
  TLS TrustStore type configuration for internal client: JKS, PKCS12, used by the internal client to authenticate with Pulsar brokers
</td>
<td>
  JKS
</td>

<td>
  TLS TrustStore path for internal client, used by the internal client to authenticate with Pulsar brokers
</td>
<td>
</td>

<td>
  TLS TrustStore password for internal client, used by the internal client to authenticate with Pulsar brokers
</td>
<td>
</td>

<td>
  Specify the tls cipher the internal client will use to negotiate during TLS Handshake. (a comma-separated list of ciphers) e.g.  [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]
</td>
<td>
</td>

<td>
  Specify the tls protocols the broker will use to negotiate during TLS handshake. (逗号分隔的协议名称列表)。 e.g.  [TLSv1.2, TLSv1.1, TLSv1]
</td>
<td>
</td>

<td>
  The default ttl for namespaces if ttl is not configured at namespace policies.
</td>
<td>
  0
</td>

<td>
  Configure the secret key to be used to validate auth tokens. The key can be specified like: <code>tokenSecretKey=data:;base64,xxxxxxxxx</code> or <code>tokenSecretKey=file:///my/secret.key</code>
</td>
<td>
</td>

<td>
  Configure the public key to be used to validate auth tokens. The key can be specified like: <code>tokenPublicKey=data:;base64,xxxxxxxxx</code> or <code>tokenPublicKey=file:///my/secret.key</code>
</td>
<td>
</td>

<td>
  Configure the algorithm to be used to validate auth tokens. This can be any of the asymettric algorithms supported by Java JWT (https://github.com/jwtk/jjwt#signature-algorithms-keys)
</td>
<td>
  RS256
</td>

<td>
  Specify which of the token's claims will be used as the authentication "principal" or "role". The default "sub" claim will be used if this is left blank
</td>
<td>
</td>

<td>
  The token audience "claim" name, e.g. "aud", that will be used to get the audience from token. If not set, audience will not be verified.
</td>
<td>
</td>

<td>
  The token audience stands for this broker. The field <code>tokenAudienceClaim</code> of a valid token, need contains this.
</td>
<td>
</td>

<td>
  Max number of unacknowledged messages allowed to receive messages by a consumer on a shared subscription. Broker will stop sending messages to consumer once, this limit reaches until consumer starts acknowledging messages back. Using a value of 0, is disabling unackeMessage limit check and consumer can receive messages without any restriction
</td>
<td>
  50000
</td>

<td>
  Max number of unacknowledged messages allowed per shared subscription. Broker will stop dispatching messages to all consumers of the subscription once this limit reaches until consumer starts acknowledging messages back and unack count reaches to limit/2. Using a value of 0, is disabling unackedMessage-limit check and dispatcher can dispatch messages without any restriction
</td>
<td>
  200000
</td>

<td>
  Enable subscription message redelivery tracker
</td>
<td>
  true
</td>

subscriptionExpirationTimeMinutes | How long to delete inactive subscriptions from last consuming.

Setting this configuration to a value

greater than 0 deletes inactive subscriptions automatically.
Setting this configuration to 0 does not delete inactive subscriptions automatically.

Since this configuration takes effect on all topics, if there is even one topic whose subscriptions should not be deleted automatically, you need to set it to 0.
Instead, you can set a subscription expiration time for each namespace using the pulsar-admin namespaces set-subscription-expiration-time options command. | 0 | |maxConcurrentLookupRequest| Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic |50000| |maxConcurrentTopicLoadRequest| Max number of concurrent topic loading request broker allows to control number of zk-operations |5000| |authenticationEnabled| Enable authentication |false| |authenticationProviders| Autentication provider name list, which is comma separated list of class names || |authorizationEnabled| Enforce authorization |false| |superUserRoles| Role names that are treated as “super-user”, meaning they will be able to do all admin operations and publish/consume from all topics || |brokerClientAuthenticationPlugin| Authentication settings of the broker itself. 当 broker 连接到其它 broker 时使用，不管是相同集群还是其它集群中的 || |brokerClientAuthenticationParameters||| |athenzDomainNames| 用于认证的受支持的 Athenz 提供者域名(逗号分隔) || |exposePreciseBacklogInPrometheus| 暴露出精确的积压统计信息，设置为 false 来使用 published counter 和 consumed counter 来进行计算，这样更高效但可能会有误差。 |false| |bookkeeperMetadataServiceUri| bookkeeper 用来载入相应元数据驱动及解析其元数据服务位置的元数据服务 uri。 This value can be fetched using bookkeeper shell whatisinstanceid command in BookKeeper cluster. 例如： zk+hierarchical://localhost:2181/ledgers。 元数据服务 uri 列表也可以是一个分号间隔的值，如下所示： zk+hierarchical://zk1:2181;zk2:2181;zk3:2181/ledgers || |bookkeeperClientAuthenticationPlugin| 连接到 bookies 时使用的认证插件 || |bookkeeperClientAuthenticationParametersName| BookKeeper 认证插件的实现指定的名字与值 || |bookkeeperClientAuthenticationParameters||| |bookkeeperClientTimeoutInSeconds| BookKeeper 添加与读取操作的超时时间 |30| |bookkeeperClientSpeculativeReadTimeoutInMillis| 当一个读请求无法在一定时间内完成时启用 speculative 读取。采用 0 值时，禁用 speculative 读取。 |0| |bookkeeperClientHealthCheckEnabled| 启用 bookie 健康检查。 Bookies that have more than the configured number of failure within the interval will be quarantined for some time. 在此期间，这些 bookies 上不会创建新的 ledger |true| |bookkeeperClientHealthCheckIntervalSeconds||60| |bookkeeperClientHealthCheckErrorThresholdPerInterval||5| |bookkeeperClientHealthCheckQuarantineTimeInSeconds ||1800| |bookkeeperClientRackawarePolicyEnabled| 启用 rack-aware bookie 选择政策。 当生成新的 bookie ensemble 时 BK 会从不同的 rack 中选取 bookie |true| |bookkeeperClientRegionawarePolicyEnabled| 启用 region-aware bookie 选择政策。 BK will chose bookies from different regions and racks when forming a new bookie ensemble. If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored |false| |bookkeeperClientMinNumRacksPerWriteQuorum| Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to get bookies from at least ‘bookkeeperClientMinNumRacksPerWriteQuorum’ racks for a write quorum. |2| |bookkeeperClientEnforceMinNumRacksPerWriteQuorum| Enforces rack-aware bookie selection policy to pick bookies from ‘bookkeeperClientMinNumRacksPerWriteQuorum’ racks for a writeQuorum. If BK can’t find bookie then it would throw BKNotEnoughBookiesException instead of picking random one. |false| |bookkeeperClientReorderReadSequenceEnabled| Enable/disable reordering read sequence on reading entries. |false| |bookkeeperClientIsolationGroups| 通过指定一个要选择的 bookie 组列表来启用 bookie 隔离。 指定组外的任何 bookie 都不会被 broker 使用 || |bookkeeperClientSecondaryIsolationGroups| 当 bookkeeperClientIsolationGroups 没有足够多的可用 bookie 时启用 bookie 的 secondary-isolation 组。 || |bookkeeperClientMinAvailableBookiesInIsolationGroups| bookkeeperClientIsolationGroups 中应可用的最少 bookie 数，否则 broker 会将 bookkeeperClientSecondaryIsolationGroups 的 bookie 囊括在隔离列表中。 || |bookkeeperClientGetBookieInfoIntervalSeconds| 设置定期查询 bookie 信息的间隔 |86400| |bookkeeperClientGetBookieInfoRetryIntervalSeconds| 设置当查询 bookie 信息失败后的重试间隔 |60| |bookkeeperEnableStickyReads | 启用/禁用 使对一个 ledger 的读请求变得与单个 broker 粘滞（sticky）。 If this flag is enabled, the client will use one single bookie (by preference) to read all entries for a ledger. | true | |managedLedgerDefaultEnsembleSize| 当创建一个 ledger 时使用的 bookie 数量 |2| |managedLedgerDefaultWriteQuorum| 每个消息存储的拷贝数量 |2| |managedLedgerDefaultAckQuorum| 得到保证的拷贝数量 (写操作完成前需要等待的 ack) |2| |managedLedgerCacheSizeMB| 在托管的 ledger 中用于缓存数据负载的内存大小。 This memory is allocated from JVM direct memory and it’s shared across all the topics running in the same broker. 默认地，使用可用的直接内存的 1/5 || |managedLedgerCacheCopyEntries| 当插入缓存时是否应该创建条目负载的拷贝 | false| |managedLedgerCacheEvictionWatermark| 当驱逐触发时降低缓存等级的阈值 |0.9| |managedLedgerCacheEvictionFrequency| 为托管的 ledger 缓存配置缓存驱逐频率 (驱逐/秒) | 100.0 | |managedLedgerCacheEvictionTimeThresholdMillis| 所有在缓存中停留超过配置时长的条目会被驱逐 | 1000 | |managedLedgerCursorBackloggedThreshold| 设置当游标应被当作“积压的” 且应被设置为非活动的阈值 (以条目数为单位)。 | 1000| |managedLedgerDefaultMarkDeleteRateLimit| Rate limit the amount of writes per second generated by consumer acking the messages |1.0| |managedLedgerMaxEntriesPerLedger| The max number of entries to append to a ledger before triggering a rollover. A ledger rollover is triggered on these conditions:

• Either the max rollover time has been reached
• or the max entries have been written to the ledger and at least min-time has passed

|50000| |managedLedgerMinLedgerRolloverTimeMinutes| Minimum time between ledger rollover for a topic |10| |managedLedgerMaxLedgerRolloverTimeMinutes| Maximum time before forcing a ledger rollover for a topic |240| |managedLedgerCursorMaxEntriesPerLedger| Max number of entries to append to a cursor ledger |50000| |managedLedgerCursorRolloverTimeInSeconds| Max time before triggering a rollover on a cursor ledger |14400| |managedLedgerMaxUnackedRangesToPersist| Max number of “acknowledgment holes” that are going to be persistently stored. When acknowledging out of order, a consumer will leave holes that are supposed to be quickly filled by acking all the messages. The information of which messages are acknowledged is persisted by compressing in “ranges” of messages that were acknowledged. After the max number of ranges is reached, the information will only be tracked in memory and messages will be redelivered in case of crashes. |1000| |autoSkipNonRecoverableData| Skip reading non-recoverable/unreadable data-ledger under managed-ledger’s list.It helps when data-ledgers gets corrupted at bookkeeper and managed-cursor is stuck at that ledger. |false| |loadBalancerEnabled| Enable load balancer |true| |loadBalancerPlacementStrategy| Strategy to assign a new bundle weightedRandomSelection || |loadBalancerReportUpdateThresholdPercentage| Percentage of change to trigger load report update |10| |loadBalancerReportUpdateMaxIntervalMinutes| maximum interval to update load report |15| |loadBalancerHostUsageCheckIntervalMinutes| Frequency of report to collect |1| |loadBalancerSheddingIntervalMinutes| Load shedding interval. Broker periodically checks whether some traffic should be offload from some over-loaded broker to other under-loaded brokers |30| |loadBalancerSheddingGracePeriodMinutes| Prevent the same topics to be shed and moved to other broker more that once within this timeframe |30| |loadBalancerBrokerMaxTopics| Usage threshold to allocate max number of topics to broker |50000| |loadBalancerBrokerUnderloadedThresholdPercentage| Usage threshold to determine a broker as under-loaded |1| |loadBalancerBrokerOverloadedThresholdPercentage| Usage threshold to determine a broker as over-loaded |85| |loadBalancerResourceQuotaUpdateIntervalMinutes| Interval to update namespace bundle resource quotat |15| |loadBalancerBrokerComfortLoadLevelPercentage| Usage threshold to determine a broker is having just right level of load |65| |loadBalancerAutoBundleSplitEnabled| enable/disable namespace bundle auto split |false| |loadBalancerNamespaceBundleMaxTopics| maximum topics in a bundle, otherwise bundle split will be triggered |1000| |loadBalancerNamespaceBundleMaxSessions| maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered |1000| |loadBalancerNamespaceBundleMaxMsgRate| maximum msgRate (in + out) in a bundle, otherwise bundle split will be triggered |1000| |loadBalancerNamespaceBundleMaxBandwidthMbytes| maximum bandwidth (in + out) in a bundle, otherwise bundle split will be triggered |100| |loadBalancerNamespaceMaximumBundles| maximum number of bundles in a namespace |128| |replicationMetricsEnabled| Enable replication metrics |true| |replicationConnectionsPerBroker| Max number of connections to open for each broker in a remote cluster More connections host-to-host lead to better throughput over high-latency links. |16| |replicationProducerQueueSize| 复制器生产者队列大小 |1000| |replicatorPrefix| 复制器生产者名与游标名使用的复制器前缀 pulsar.repl|| |replicationTlsEnabled| 当与其它集群进行沟通以进行消息复制时启用 TLS |false| |defaultRetentionTimeInMinutes| 默认消息保留时间 || |defaultRetentionSizeInMB| 默认保留大小 |0| |keepAliveIntervalSeconds| 检查连接是否仍然存活的频率 |30| |loadManagerClassName| 要使用的载入管理器的名字 |org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl| |supportedNamespaceBundleSplitAlgorithms| 切分命名空间 bundle 支持的算法名 |[range_equally_divide,topic_count_equally_divide]| |defaultNamespaceBundleSplitAlgorithm| 默认的命名空间 bundle 切分算法名 |range_equally_divide| |managedLedgerOffloadDriver| 将旧数据卸载至长期存储时使用的驱动 (可能的值: S3) || |managedLedgerOffloadMaxThreads| ledger 卸载时使用的最大线程池线程数 |2| |managedLedgerUnackedRangesOpenCacheSetEnabled| 使用 Open Range-Set 来缓存未确认的消息 |true| |managedLedgerOffloadDeletionLagMs| 从 ledger 成功卸载至长期存储到 ledger 从 bookeeper 中删除的时间差 | 14400000| |managedLedgerOffloadAutoTriggerSizeThresholdBytes| 触发自动卸载至长期存储前的字节数 |-1 (禁用)| |s3ManagedLedgerOffloadRegion| 对于Amazon S3 ledger 卸载，AWS region|| |s3ManagedLedgerOffloadBucket| 对于Amazon S3 ledger 卸载，将卸载的 ledger 放入的 bucket || |s3ManagedLedgerOffloadServiceEndpoint| 对于Amazon S3 ledger 卸载，替代的连接端点 (对测试很有用) || |s3ManagedLedgerOffloadMaxBlockSizeInBytes| 对于 Amazon S3 ledger 卸载，最大 block 大小（以字节计） (默认为 64MB，最小 5MB) |67108864| |s3ManagedLedgerOffloadReadBufferSizeInBytes| 对于 Amazon S3 ledger 卸载，以字节计的读缓存大小 (默认为 1MB) |1048576| |s3ManagedLedgerOffloadRole| 对于 Amazon S3 ledger 卸载，在写入 s3 前提供一个假定角色 || |s3ManagedLedgerOffloadRoleSessionName| 对于 Amazon S3 ledger 卸载，当使用一个角色时提供一个角色会话名 |pulsar-s3-offload| | acknowledgmentAtBatchIndexLevelEnabled | 启用或禁用批量下标确认。 | false | | maxMessageSize | 设置消息的最大大小。 | 5 MB |

Client

The pulsar-client CLI tool can be used to publish messages to Pulsar and consume messages from Pulsar topics. This tool can be used in lieu of a client library.

服务发现

Log4j

Log4j shell

Standalone

WebSocket

Pulsar proxy

The Pulsar proxy can be configured in the conf/proxy.conf file.

ZooKeeper

ZooKeeper handles a broad range of essential configuration- and coordination-related tasks for Pulsar. The default configuration file for ZooKeeper is in the conf/zookeeper.conf file in your Pulsar installation. The following parameters are available:

In addition to the parameters in the table above, configuring ZooKeeper for Pulsar involves adding a server.N line to the conf/zookeeper.conf file for each node in the ZooKeeper cluster, where N is the number of the ZooKeeper node. Here’s an example for a three-node ZooKeeper cluster:

server.1=zk1.us-west.example.com:2888:3888
server.2=zk2.us-west.example.com:2888:3888
server.3=zk3.us-west.example.com:2888:3888

We strongly recommend consulting the ZooKeeper Administrator’s Guide for a more thorough and comprehensive introduction to ZooKeeper configuration