diff --git a/site2/docs/administration-zk-bk.md b/site2/docs/administration-zk-bk.md index 3965c3296aeee..11189424193f0 100644 --- a/site2/docs/administration-zk-bk.md +++ b/site2/docs/administration-zk-bk.md @@ -133,27 +133,19 @@ $ bin/pulsar-daemon start configuration-store ### ZooKeeper configuration -In Pulsar, ZooKeeper configuration is handled by two separate configuration files in the `conf` directory of your Pulsar installation: `conf/zookeeper.conf` for [local ZooKeeper](#local-zookeeper) and `conf/global-zookeeper.conf` for [configuration store](#configuration-store). +In Pulsar, ZooKeeper configuration is handled by two separate configuration files in the `conf` directory of your Pulsar installation: +* The `conf/zookeeper.conf` file handles the configuration for local ZooKeeper. +* The `conf/global-zookeeper.conf` file handles the configuration for configuration store. +See [parameters](reference-configuration.md#zookeeper) for more details. -#### Local ZooKeeper +#### Configure batching operations +Using the batching operations reduces the remote procedure call (RPC) traffic between ZooKeeper client and servers. It also reduces the number of write transactions, because each batching operation corresponds to a single ZooKeeper transaction, containing multiple read and write operations. -The [`conf/zookeeper.conf`](reference-configuration.md#zookeeper) file handles the configuration for local ZooKeeper. The table below shows the available parameters: +The following figure demonstrates a basic benchmark of batching read/write operations that can be requested to ZooKeeper in one second: -|Name|Description|Default| -|---|---|---| -|tickTime| The tick is the basic unit of time in ZooKeeper, measured in milliseconds and used to regulate things like heartbeats and timeouts. tickTime is the length of a single tick. |2000| -|initLimit| The maximum time, in ticks, that the leader ZooKeeper server allows follower ZooKeeper servers to successfully connect and sync. The tick time is set in milliseconds using the tickTime parameter. |10| -|syncLimit| The maximum time, in ticks, that a follower ZooKeeper server is allowed to sync with other ZooKeeper servers. The tick time is set in milliseconds using the tickTime parameter. |5| -|dataDir| The location where ZooKeeper stores in-memory database snapshots as well as the transaction log of updates to the database. |data/zookeeper| -|clientPort| The port on which the ZooKeeper server listens for connections. |2181| -|autopurge.snapRetainCount| In ZooKeeper, auto purge determines how many recent snapshots of the database stored in dataDir to retain within the time interval specified by autopurge.purgeInterval (while deleting the rest). |3| -|autopurge.purgeInterval| The time interval, in hours, which triggers the ZooKeeper database purge task. Setting to a non-zero number enables auto purge; setting to 0 disables. Read this guide before enabling auto purge. |1| -|maxClientCnxns| The maximum number of client connections. Increase this if you need to handle more ZooKeeper clients. |60| +![Zookeeper batching benchmark](assets/zookeeper-batching.png) - -#### Configuration Store - -The [`conf/global-zookeeper.conf`](reference-configuration.md#configuration-store) file handles the configuration for configuration store. The table below shows the available parameters: +To enable batching operations, set the [`metadataStoreBatchingEnabled`](reference-configuration.md#broker) parameter to `true` on the broker side. ## BookKeeper diff --git a/site2/docs/assets/zookeeper-batching.png b/site2/docs/assets/zookeeper-batching.png new file mode 100644 index 0000000000000..4bd461ef7f08c Binary files /dev/null and b/site2/docs/assets/zookeeper-batching.png differ diff --git a/site2/docs/concepts-messaging.md b/site2/docs/concepts-messaging.md index bd9d23c892fd6..e87f9b3815b0f 100644 --- a/site2/docs/concepts-messaging.md +++ b/site2/docs/concepts-messaging.md @@ -96,29 +96,44 @@ To avoid redelivering acknowledged messages in a batch to the consumer, Pulsar i By default, batch index acknowledgement is disabled (`acknowledgmentAtBatchIndexLevelEnabled=false`). You can enable batch index acknowledgement by setting the `acknowledgmentAtBatchIndexLevelEnabled` parameter to `true` at the broker side. Enabling batch index acknowledgement results in more memory overheads. ### Chunking -Before you enable chunking, read the following instructions. -- Batching and chunking cannot be enabled simultaneously. To enable chunking, you must disable batching in advance. -- Chunking is only supported for persisted topics. -- Chunking is only supported for Exclusive and Failover subscription types. +Message chunking enables Pulsar to process large payload messages by splitting the message into chunks at the producer side and aggregating chunked messages at the consumer side. -When chunking is enabled (`chunkingEnabled=true`), if the message size is greater than the allowed maximum publish-payload size, the producer splits the original message into chunked messages and publishes them with chunked metadata to the broker separately and in order. At the broker side, the chunked messages are stored in the managed-ledger in the same way as that of ordinary messages. The only difference is that the consumer needs to buffer the chunked messages and combines them into the real message when all chunked messages have been collected. The chunked messages in the managed-ledger can be interwoven with ordinary messages. If producer fails to publish all the chunks of a message, the consumer can expire incomplete chunks if consumer fail to receive all chunks in expire time. By default, the expire time is set to one minute. +With message chunking enabled, when the size of a message exceeds the allowed maximum payload size (the `maxMessageSize` parameter of broker), the workflow of messaging is as follows: +1. The producer splits the original message into chunked messages and publishes them with chunked metadata to the broker separately and in order. +2. The broker stores the chunked messages in one managed-ledger in the same way as that of ordinary messages, and it uses the `chunkedMessageRate` parameter to record chunked message rate on the topic. +3. The consumer buffers the chunked messages and aggregates them into the receiver queue when it receives all the chunks of a message. +4. The client consumes the aggregated message from the receiver queue. -The consumer consumes the chunked messages and buffers them until the consumer receives all the chunks of a message. And then the consumer stitches chunked messages together and places them into the receiver-queue. Clients consume messages from the receiver-queue. Once the consumer consumes the entire large message and acknowledges it, the consumer internally sends acknowledgement of all the chunk messages associated to that large message. You can set the `maxPendingChunkedMessage` parameter on the consumer. When the threshold is reached, the consumer drops the unchunked messages by silently acknowledging them or asking the broker to redeliver them later by marking them unacknowledged. +**Limitations:** +- Chunking is only available for persisted topics. +- Chunking is only available for the exclusive and failover subscription types. +- Chunking cannot be enabled simultaneously with batching. -The broker does not require any changes to support chunking for non-shared subscription. The broker only uses `chunkedMessageRate` to record chunked message rate on the topic. +#### Handle consecutive chunked messages with one ordered consumer -#### Handle chunked messages with one producer and one ordered consumer - -As shown in the following figure, when a topic has one producer which publishes large message payload in chunked messages along with regular non-chunked messages. The producer publishes message M1 in three chunks M1-C1, M1-C2 and M1-C3. The broker stores all the three chunked messages in the managed-ledger and dispatches to the ordered (exclusive/failover) consumer in the same order. The consumer buffers all the chunked messages in memory until it receives all the chunked messages, combines them into one message and then hands over the original message M1 to the client. +The following figure shows a topic with one producer which publishes a large message payload in chunked messages along with regular non-chunked messages. The producer publishes message M1 in three chunks labeled M1-C1, M1-C2 and M1-C3. The broker stores all the three chunked messages in the managed-ledger and dispatches them to the ordered (exclusive/failover) consumer in the same order. The consumer buffers all the chunked messages in memory until it receives all the chunked messages, aggregates them into one message and then hands over the original message M1 to the client. ![](assets/chunking-01.png) -#### Handle chunked messages with multiple producers and one ordered consumer +#### Handle interwoven chunked messages with one ordered consumer -When multiple publishers publish chunked messages into a single topic, the broker stores all the chunked messages coming from different publishers in the same managed-ledger. As shown below, Producer 1 publishes message M1 in three chunks M1-C1, M1-C2 and M1-C3. Producer 2 publishes message M2 in three chunks M2-C1, M2-C2 and M2-C3. All chunked messages of the specific message are still in order but might not be consecutive in the managed-ledger. This brings some memory pressure to the consumer because the consumer keeps separate buffer for each large message to aggregate all chunks of the large message and combine them into one message. +When multiple producers publish chunked messages into a single topic, the broker stores all the chunked messages coming from different producers in the same managed-ledger. The chunked messages in the managed-ledger can be interwoven with each other. As shown below, Producer 1 publishes message M1 in three chunks M1-C1, M1-C2 and M1-C3. Producer 2 publishes message M2 in three chunks M2-C1, M2-C2 and M2-C3. All chunked messages of the specific message are still in order but might not be consecutive in the managed-ledger. ![](assets/chunking-02.png) +> **Note** +> In this case, interwoven chunked messages may bring some memory pressure to the consumer because the consumer keeps a separate buffer for each large message to aggregate all its chunks in one message. You can limit the maximum number of chunked messages a consumer maintains concurrently by configuring the `maxPendingChunkedMessage` parameter. When the threshold is reached, the consumer drops pending messages by silently acknowledging them or asking the broker to redeliver them later, optimizing memory utilization. + +#### Enable Message Chunking + +**Prerequisite:** Disable batching by setting the `enableBatching` parameter to `false`. + +The message chunking feature is OFF by default. +To enable message chunking, set the `chunkingEnabled` parameter to `true` when creating a producer. + +> **Note** +> If the consumer fails to receive all chunks of a message within a specified time period, it expires incomplete chunks. The default value is 1 minute. For more information about the `expireTimeOfIncompleteChunkedMessage` parameter, refer to [org.apache.pulsar.client.api](https://pulsar.apache.org/api/client/). + ## Consumers A consumer is a process that attaches to a topic via a subscription and then receives messages. diff --git a/site2/docs/reference-configuration.md b/site2/docs/reference-configuration.md index 28e767fbec6a5..d3eb7251766db 100644 --- a/site2/docs/reference-configuration.md +++ b/site2/docs/reference-configuration.md @@ -109,7 +109,7 @@ BookKeeper is a replicated log storage system that Pulsar uses for durable stora |readBufferSizeBytes|The number of bytes we should use as capacity for BufferedReadChannel.|4096| |writeBufferSizeBytes|The number of bytes used as capacity for the write buffer|65536| |useHostNameAsBookieID|Whether the bookie should use its hostname to register with the coordination service (e.g.: zookeeper service). When false, bookie will use its ip address for the registration.|false| -|bookieId | If you want to custom a bookie ID or use a dynamic network address for the bookie, you can set the `bookieId`.

Bookie advertises itself using the `bookieId` rather than the `BookieSocketAddress` (`hostname:port` or `IP:port`). If you set the `bookieId`, then the `useHostNameAsBookieID` does not take effect.

The `bookieId` is a non-empty string that can contain ASCII digits and letters ([a-zA-Z9-0]), colons, dashes, and dots.

For more information about `bookieId`, see [here](http://bookkeeper.apache.org/bps/BP-41-bookieid/).|N/A| +|bookieId | If you want to custom a bookie ID or use a dynamic network address for the bookie, you can set the `bookieId`.

Bookie advertises itself using the `bookieId` rather than the `BookieSocketAddress` (`hostname:port` or `IP:port`). If you set the `bookieId`, then the `useHostNameAsBookieID` does not take effect.

The `bookieId` is a non-empty string that can contain ASCII digits and letters ([a-zA-Z9-0]), colons, dashes, and dots.

For more information about `bookieId`, see [here](http://bookkeeper.apache.org/bps/BP-41-bookieid/).|N/A| |allowEphemeralPorts|Whether the bookie is allowed to use an ephemeral port (port 0) as its server port. By default, an ephemeral port is not allowed. Using an ephemeral port as the service port usually indicates a configuration error. However, in unit tests, using an ephemeral port will address port conflict problems and allow running tests in parallel.|false| |enableLocalTransport|Whether the bookie is allowed to listen for the BookKeeper clients executed on the local JVM.|false| |disableServerSocketBind|Whether the bookie is allowed to disable bind on network interfaces. This bookie will be available only to BookKeeper clients executed on the local JVM.|false| @@ -136,8 +136,8 @@ Pulsar brokers are responsible for handling incoming messages from producers, di |Name|Description|Default| |---|---|---| -|advertisedListeners|Specify multiple advertised listeners for the broker.

The format is `:pulsar://:`.

If there are multiple listeners, separate them with commas.

**Note**: do not use this configuration with `advertisedAddress` and `brokerServicePort`. If the value of this configuration is empty, the broker uses `advertisedAddress` and `brokerServicePort`|/| -|internalListenerName|Specify the internal listener name for the broker.

**Note**: the listener name must be contained in `advertisedListeners`.

If the value of this configuration is empty, the broker uses the first listener as the internal listener.|/| +|advertisedListeners|Specify multiple advertised listeners for the broker.

The format is `:pulsar://:`.

If there are multiple listeners, separate them with commas.

**Note**: do not use this configuration with `advertisedAddress` and `brokerServicePort`. If the value of this configuration is empty, the broker uses `advertisedAddress` and `brokerServicePort`|/| +|internalListenerName|Specify the internal listener name for the broker.

**Note**: the listener name must be contained in `advertisedListeners`.

If the value of this configuration is empty, the broker uses the first listener as the internal listener.|/| |authenticateOriginalAuthData| If this flag is set to `true`, the broker authenticates the original Auth data; else it just accepts the originalPrincipal and authorizes it (if required). |false| |enablePersistentTopics| Whether persistent topics are enabled on the broker |true| |enableNonPersistentTopics| Whether non-persistent topics are enabled on the broker |true| @@ -198,7 +198,7 @@ Pulsar brokers are responsible for handling incoming messages from producers, di |forceDeleteNamespaceAllowed| Enable you to delete a namespace forcefully. |false| |messageExpiryCheckIntervalInMinutes| The frequency of proactively checking and purging expired messages. |5| |brokerServiceCompactionMonitorIntervalInSeconds| Interval between checks to determine whether topics with compaction policies need compaction. |60| -brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater than this threshold, compression is triggered.

Set this threshold to 0 means disabling the compression check.|N/A +brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater than this threshold, compression is triggered.

Set this threshold to 0 means disabling the compression check.|N/A |delayedDeliveryEnabled| Whether to enable the delayed delivery for messages. If disabled, messages will be immediately delivered and there will be no tracking overhead.|true| |delayedDeliveryTickTimeMillis|Control the tick time for retrying on delayed delivery, which affects the accuracy of the delivery time compared to the scheduled time. By default, it is 1 second.|1000| |activeConsumerFailoverDelayTimeMillis| How long to delay rewinding cursor and dispatching messages when active consumer is changed. |1000| @@ -226,6 +226,10 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater |brokerClientTlsTrustStorePassword| TLS TrustStore password for internal client, used by the internal client to authenticate with Pulsar brokers || |brokerClientTlsCiphers| 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]|| |brokerClientTlsProtocols|Specify the tls protocols the broker will use to negotiate during TLS handshake. (a comma-separated list of protocol names). e.g. `TLSv1.3`, `TLSv1.2` || +| metadataStoreBatchingEnabled | Enable metadata operations batching. | true | +| metadataStoreBatchingMaxDelayMillis | Maximum delay to impose on batching grouping. | 5 | +| metadataStoreBatchingMaxOperations | Maximum number of operations to include in a singular batch. | 1000 | +| metadataStoreBatchingMaxSizeKb | Maximum size of a batch. | 128 | |ttlDurationDefaultInSeconds|The default Time to Live (TTL) for namespaces if the TTL is not configured at namespace policies. When the value is set to `0`, TTL is disabled. By default, TTL is disabled. |0| |tokenSettingPrefix| Configure the prefix of the token-related settings, such as `tokenSecretKey`, `tokenPublicKey`, `tokenAuthClaim`, `tokenPublicAlg`, `tokenAudienceClaim`, and `tokenAudience`. || |tokenSecretKey| Configure the secret key to be used to validate auth tokens. The key can be specified like: `tokenSecretKey=data:;base64,xxxxxxxxx` or `tokenSecretKey=file:///my/secret.key`. Note: key file must be DER-encoded.|| @@ -295,7 +299,7 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater |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| +|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 than once within this timeframe |30| @@ -360,8 +364,8 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater |haProxyProtocolEnabled | Enable or disable the [HAProxy](http://www.haproxy.org/) protocol. |false| | maxTopicsPerNamespace | The maximum number of persistent topics that can be created in the namespace. When the number of topics reaches this threshold, the broker rejects the request of creating a new topic, including the auto-created topics by the producer or consumer, until the number of connected consumers decreases. The default value 0 disables the check. | 0 | |subscriptionTypesEnabled| Enable all subscription types, which are exclusive, shared, failover, and key_shared. | Exclusive, Shared, Failover, Key_Shared | -| managedLedgerInfoCompressionType | Compression type of managed ledger information.

Available options are `NONE`, `LZ4`, `ZLIB`, `ZSTD`, and `SNAPPY`).

If this value is `NONE` or invalid, the `managedLedgerInfo` is not compressed.

**Note** that after enabling this configuration, if you want to degrade a broker, you need to change the value to `NONE` and make sure all ledger metadata is saved without compression. | None | -| additionalServlets | Additional servlet name.

If you have multiple additional servlets, separate them by commas.

For example, additionalServlet_1, additionalServlet_2 | N/A | +| managedLedgerInfoCompressionType | Compression type of managed ledger information.

Available options are `NONE`, `LZ4`, `ZLIB`, `ZSTD`, and `SNAPPY`).

If this value is `NONE` or invalid, the `managedLedgerInfo` is not compressed.

**Note** that after enabling this configuration, if you want to degrade a broker, you need to change the value to `NONE` and make sure all ledger metadata is saved without compression. | None | +| additionalServlets | Additional servlet name.

If you have multiple additional servlets, separate them by commas.

For example, additionalServlet_1, additionalServlet_2 | N/A | | additionalServletDirectory | Location of broker additional servlet NAR directory | ./brokerAdditionalServlet | | brokerEntryMetadataInterceptors | Set broker entry metadata interceptors.

Multiple interceptors should be separated by commas.

Available values:
  • org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor
  • org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor


    • Example
    brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendBrokerTimestampMetadataInterceptor, org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor|N/A | | enableExposingBrokerEntryMetadataToClient|Whether to expose broker entry metadata to client or not.

    Available values:
  • true
  • false

    • Example
    enableExposingBrokerEntryMetadataToClient=true | false | @@ -519,7 +523,10 @@ You can set the log level and configuration in the [log4j2.yaml](https://github | maxProducersPerTopic | Maximum number of producers allowed to connect to topic. Once this limit reaches, the broker rejects new producers until the number of connected producers decreases. When the value is set to 0, maxProducersPerTopic-limit check is disabled. | 0 | | maxConsumersPerTopic | Maximum number of consumers allowed to connect to topic. Once this limit reaches, the broker rejects new consumers until the number of connected consumers decreases. When the value is set to 0, maxConsumersPerTopic-limit check is disabled. | 0 | | maxConsumersPerSubscription | Maximum number of consumers allowed to connect to subscription. Once this limit reaches, the broker rejects new consumers until the number of connected consumers decreases. When the value is set to 0, maxConsumersPerSubscription-limit check is disabled. | 0 | -| maxNumPartitionsPerPartitionedTopic | Maximum number of partitions per partitioned topic. When the value is set to a negative number or is set to 0, the check is disabled. | 0 | +| metadataStoreBatchingEnabled | Enable metadata operations batching. | true | +| metadataStoreBatchingMaxDelayMillis | Maximum delay to impose on batching grouping. | 5 | +| metadataStoreBatchingMaxOperations | Maximum number of operations to include in a singular batch. | 1000 | +| metadataStoreBatchingMaxSizeKb | Maximum size of a batch. | 128 | | tlsCertRefreshCheckDurationSec | TLS certificate refresh duration in seconds. When the value is set to 0, check the TLS certificate on every new connection. | 300 | | tlsCertificateFilePath | Path for the TLS certificate file. | | | tlsKeyFilePath | Path for the TLS private key file. | | @@ -647,7 +654,7 @@ You can set the log level and configuration in the [log4j2.yaml](https://github |loadBalancerAutoBundleSplitEnabled| |false| | loadBalancerAutoUnloadSplitBundlesEnabled | Enable/Disable automatic unloading of split bundles. | true | |loadBalancerNamespaceBundleMaxTopics| |1000| -|loadBalancerNamespaceBundleMaxSessions| |1000| +|loadBalancerNamespaceBundleMaxSessions| Maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered.
    To disable the threshold check, set the value to -1. |1000| |loadBalancerNamespaceBundleMaxMsgRate| |1000| |loadBalancerNamespaceBundleMaxBandwidthMbytes| |100| |loadBalancerNamespaceMaximumBundles| |128| @@ -670,7 +677,7 @@ You can set the log level and configuration in the [log4j2.yaml](https://github |defaultRetentionSizeInMB| |0| |keepAliveIntervalSeconds| |30| |haProxyProtocolEnabled | Enable or disable the [HAProxy](http://www.haproxy.org/) protocol. |false| -|bookieId | If you want to custom a bookie ID or use a dynamic network address for a bookie, you can set the `bookieId`.

    Bookie advertises itself using the `bookieId` rather than the `BookieSocketAddress` (`hostname:port` or `IP:port`).

    The `bookieId` is a non-empty string that can contain ASCII digits and letters ([a-zA-Z9-0]), colons, dashes, and dots.

    For more information about `bookieId`, see [here](http://bookkeeper.apache.org/bps/BP-41-bookieid/).|/| +|bookieId | If you want to custom a bookie ID or use a dynamic network address for a bookie, you can set the `bookieId`.

    Bookie advertises itself using the `bookieId` rather than the `BookieSocketAddress` (`hostname:port` or `IP:port`).

    The `bookieId` is a non-empty string that can contain ASCII digits and letters ([a-zA-Z9-0]), colons, dashes, and dots.

    For more information about `bookieId`, see [here](http://bookkeeper.apache.org/bps/BP-41-bookieid/).|/| | maxTopicsPerNamespace | The maximum number of persistent topics that can be created in the namespace. When the number of topics reaches this threshold, the broker rejects the request of creating a new topic, including the auto-created topics by the producer or consumer, until the number of connected consumers decreases. The default value 0 disables the check. | 0 | ## WebSocket