You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rx...@apache.org on 2020/08/18 08:39:21 UTC
[pulsar] branch master updated: Update 2.6.1 docs (#7832)
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 8de22bd Update 2.6.1 docs (#7832)
8de22bd is described below
commit 8de22bdff0227fd6008afe293e50c0deb4f0331f
Author: 冉小龙 <rx...@apache.org>
AuthorDate: Tue Aug 18 16:39:03 2020 +0800
Update 2.6.1 docs (#7832)
Signed-off-by: xiaolong.ran <rx...@apache.org>
### Motivation
Update 2.6.1 docs
### Modifications
- update 2.6.1 docs
- add 2.6.1 sidebars.json
- update releases.json
- update versions.json
---
site2/website/releases.json | 1 +
.../version-2.6.1/admin-api-namespaces.md | 11 +-
.../version-2.6.1/admin-api-schemas.md | 7 -
.../version-2.6.1/administration-zk-bk.md | 92 ++--
.../version-2.6.1/client-libraries-go.md | 19 +
.../version-2.6.1/client-libraries-java.md | 49 +-
.../version-2.6.1/client-libraries-python.md | 27 --
.../version-2.6.1/concepts-clients.md | 2 +-
.../version-2.6.1/concepts-messaging.md | 158 +++---
.../version-2.6.1/concepts-proxy-sni-routing.md | 121 +++++
.../version-2.6.1/concepts-schema-registry.md | 6 +
.../version-2.6.1/cookbooks-deduplication.md | 10 +-
.../version-2.6.1/cookbooks-encryption.md | 2 +-
.../version-2.6.1/cookbooks-tiered-storage.md | 2 -
.../version-2.6.1/deploy-bare-metal.md | 2 +-
.../versioned_docs/version-2.6.1/deploy-dcos.md | 4 +-
.../versioned_docs/version-2.6.1/functions-cli.md | 2 +-
.../version-2.6.1/functions-debug.md | 2 +-
.../version-2.6.1/functions-deploy.md | 2 +-
.../version-2.6.1/functions-develop.md | 2 +-
.../version-2.6.1/functions-metrics.md | 7 -
.../version-2.6.1/functions-runtime.md | 5 +-
.../version-2.6.1/functions-worker.md | 46 +-
.../getting-started-concepts-and-architecture.md | 16 -
.../version-2.6.1/getting-started-helm.md | 33 +-
.../versioned_docs/version-2.6.1/helm-install.md | 2 +-
.../versioned_docs/version-2.6.1/io-cdc-canal.md | 6 +
.../website/versioned_docs/version-2.6.1/io-cli.md | 12 +-
.../versioned_docs/version-2.6.1/io-hdfs2-sink.md | 6 +-
.../versioned_docs/version-2.6.1/io-quickstart.md | 4 -
.../versioned_docs/version-2.6.1/io-twitter.md | 7 -
.../version-2.6.1/reference-cli-tools.md | 8 +-
.../version-2.6.1/reference-configuration.md | 5 +-
.../version-2.6.1/reference-pulsar-admin.md | 2 +-
.../version-2.6.1/schema-get-started.md | 4 -
.../version-2.6.1/security-bouncy-castle.md | 7 +-
.../version-2.6.1/security-oauth2.md | 135 ++++++
.../version-2.6.1/security-overview.md | 4 +
.../version-2.6.1/sql-deployment-configurations.md | 6 +-
.../version-2.6.1/window-functions-context.md | 529 +++++++++++++++++++++
.../versioned_sidebars/version-2.6.1-sidebars.json | 155 ++++++
site2/website/versions.json | 1 +
42 files changed, 1237 insertions(+), 284 deletions(-)
diff --git a/site2/website/releases.json b/site2/website/releases.json
index b82be5f..bad6d43 100644
--- a/site2/website/releases.json
+++ b/site2/website/releases.json
@@ -1,4 +1,5 @@
[
+ "2.6.1",
"2.6.0",
"2.5.2",
"2.5.1",
diff --git a/site2/website/versioned_docs/version-2.6.1/admin-api-namespaces.md b/site2/website/versioned_docs/version-2.6.1/admin-api-namespaces.md
index c1cdd44..d437cf5 100644
--- a/site2/website/versioned_docs/version-2.6.1/admin-api-namespaces.md
+++ b/site2/website/versioned_docs/version-2.6.1/admin-api-namespaces.md
@@ -555,11 +555,18 @@ admin.namespaces().getRetention(namespace)
#### set dispatch throttling
-It sets message dispatch rate for all the topics under a given namespace.
+It sets message dispatch rate for all the topics under a given namespace.
Dispatch rate can be restricted by number of message per X seconds (`msg-dispatch-rate`) or by number of message-bytes per X second (`byte-dispatch-rate`).
dispatch rate is in second and it can be configured with `dispatch-rate-period`. Default value of `msg-dispatch-rate` and `byte-dispatch-rate` is -1 which
disables the throttling.
+#### Note
+> - If neither `clusterDispatchRate` nor `topicDispatchRate` is configured, dispatch throttling is disabled.
+> >
+> - If `topicDispatchRate` is not configured, `clusterDispatchRate` takes effect.
+> >
+> - If `topicDispatchRate` is configured, `topicDispatchRate` takes effect.
+
###### CLI
```
@@ -670,7 +677,7 @@ $ pulsar-admin namespaces get-subscription-dispatch-rate test-tenant/ns1
admin.namespaces().getSubscriptionDispatchRate(namespace)
```
-#### set dispatch throttling for subscription
+#### set dispatch throttling for replicator
It sets message dispatch rate for all the replicator between replication clusters under a given namespace.
Dispatch rate can be restricted by number of message per X seconds (`msg-dispatch-rate`) or by number of message-bytes per X second (`byte-dispatch-rate`).
diff --git a/site2/website/versioned_docs/version-2.6.1/admin-api-schemas.md b/site2/website/versioned_docs/version-2.6.1/admin-api-schemas.md
deleted file mode 100644
index eb36f87..0000000
--- a/site2/website/versioned_docs/version-2.6.1/admin-api-schemas.md
+++ /dev/null
@@ -1,7 +0,0 @@
----
-id: version-2.6.1-admin-api-schemas
-title: Managing Schemas
-sidebar_label: Schemas
-original_id: admin-api-schemas
----
-
diff --git a/site2/website/versioned_docs/version-2.6.1/administration-zk-bk.md b/site2/website/versioned_docs/version-2.6.1/administration-zk-bk.md
index ff85296..c6d0309 100644
--- a/site2/website/versioned_docs/version-2.6.1/administration-zk-bk.md
+++ b/site2/website/versioned_docs/version-2.6.1/administration-zk-bk.md
@@ -163,49 +163,17 @@ BookKeeper is responsible for all durable message storage in Pulsar. BookKeeper
> For a guide to managing message persistence, retention, and expiry in Pulsar, see [this cookbook](cookbooks-retention-expiry.md).
-### Hardware considerations
-
-Bookie hosts are responsible for storing message data on disk. In order for bookies to provide optimal performance, ensuring that the bookies have a suitable hardware configuration is essential. You can choose two key dimensions to bookie hardware capacity:
-
-* Disk I/O capacity read/write
-* Storage capacity
-
-Message entries written to bookies are always synced to disk before returning an acknowledgement to the Pulsar broker. To ensure low write latency, BookKeeper is designed to use multiple devices:
-
-* A **journal** to ensure durability. For sequential writes, having fast [fsync](https://linux.die.net/man/2/fsync) operations on bookie hosts is critical. Typically, small and fast [solid-state drives](https://en.wikipedia.org/wiki/Solid-state_drive) (SSDs) should suffice, or [hard disk drives](https://en.wikipedia.org/wiki/Hard_disk_drive) (HDDs) with a [RAID](https://en.wikipedia.org/wiki/RAID)s controller and a battery-backed write cache. Both solutions can reach fsync latency of ~0.4 ms.
-* A **ledger storage device** is where data is stored until all consumers have acknowledged the message. Writes happen in the background, so write I/O is not a big concern. Reads happen sequentially most of the time and the backlog is drained only in case of consumer drain. To store large amounts of data, a typical configuration involves multiple HDDs with a RAID controller.
-
-### Configure BookKeeper
-
-You can configure BookKeeper bookies using the [`conf/bookkeeper.conf`](reference-configuration.md#bookkeeper) configuration file. The most important aspect of configuring each bookie is ensuring that the [`zkServers`](reference-configuration.md#bookkeeper-zkServers) parameter is set to the connection string for local ZooKeeper of the Pulsar cluster.
-
-Minimum configuration changes required in `conf/bookkeeper.conf` are:
-
-```properties
-# Change to point to journal disk mount point
-journalDirectory=data/bookkeeper/journal
-
-# Point to ledger storage disk mount point
-ledgerDirectories=data/bookkeeper/ledgers
-
-# Point to local ZK quorum
-zkServers=zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181
-
-# Change the ledger manager type
-ledgerManagerType=hierarchical
-```
-
-To change the zookeeper root path that Bookkeeper uses, use zkLedgersRootPath=/MY-PREFIX/ledgers instead of zkServers=localhost:2181/MY-PREFIX
-
-> Consult the official [BookKeeper docs](http://bookkeeper.apache.org) for more information about BookKeeper.
-
### Deploy BookKeeper
BookKeeper provides [persistent message storage](concepts-architecture-overview.md#persistent-storage) for Pulsar.
Each Pulsar broker needs to have its own cluster of bookies. The BookKeeper cluster shares a local ZooKeeper quorum with the Pulsar cluster.
-### Starting bookies manually
+### Configure bookies
+
+You can configure BookKeeper bookies using the [`conf/bookkeeper.conf`](reference-configuration.md#bookkeeper) configuration file. The most important aspect of configuring each bookie is ensuring that the [`zkServers`](reference-configuration.md#bookkeeper-zkServers) parameter is set to the connection string for local ZooKeeper of the Pulsar cluster.
+
+### Start up bookies
You can start up a bookie in two ways: in the foreground or as a background daemon.
@@ -229,46 +197,44 @@ $ bin/bookkeeper shell bookiesanity
This command creates a new ledger on the local bookie, writes a few entries, reads them back and finally deletes the ledger.
-### Decommissioning bookies cleanly
+### Hardware considerations
+Bookie hosts are responsible for storing message data on disk. In order for bookies to provide optimal performance, ensuring that the bookies have a suitable hardware configuration is essential. You can choose two key dimensions to bookie hardware capacity:
-In case the user wants to decommission a bookie, the following process is useful to follow in order to verify if the
-decommissioning was safely done.
+* Disk I/O capacity read/write
+* Storage capacity
-#### Before we decommission
-1. Ensure state of your cluster can support the decommissioning of the target bookie.
-Check if `EnsembleSize >= Write Quorum >= Ack Quorum` stays true with one less bookie
+Message entries written to bookies are always synced to disk before returning an acknowledgement to the Pulsar broker. To ensure low write latency, BookKeeper is designed to use multiple devices:
-2. Ensure target bookie shows up in the listbookies command.
+* A **journal** to ensure durability. For sequential writes, having fast [fsync](https://linux.die.net/man/2/fsync) operations on bookie hosts is critical. Typically, small and fast [solid-state drives](https://en.wikipedia.org/wiki/Solid-state_drive) (SSDs) should suffice, or [hard disk drives](https://en.wikipedia.org/wiki/Hard_disk_drive) (HDDs) with a [RAID](https://en.wikipedia.org/wiki/RAID)s controller and a battery-backed write cache. Both solutions can reach fsync latency of ~0.4 ms.
+* A **ledger storage device** is where data is stored until all consumers have acknowledged the message. Writes happen in the background, so write I/O is not a big concern. Reads happen sequentially most of the time and the backlog is drained only in case of consumer drain. To store large amounts of data, a typical configuration involves multiple HDDs with a RAID controller.
-3. Ensure that there is no other process ongoing (upgrade etc).
-#### Process of Decommissioning
-1. Log on to the bookie node, check if there are underreplicated ledgers.
-If there are, the decommission command will force them to be replicated.
-`$ bin/bookkeeper shell listunderreplicated`
+### Configure BookKeeper
-2. Stop the bookie by killing the bookie process. Make sure there are no liveness / readiness probes setup for the bookies to spin them back up if you are deployed in a kubernetes environment.
+you can find configurable parameters for BookKeeper bookies in the [`conf/bookkeeper.conf`](reference-configuration.md#bookkeeper) file.
-3. Run the decommission command.
-If you have logged onto the node you wish to decommission, you don't need to provide `-bookieid`
-If you are running the decommission command for target bookie node from another bookie node you should mention
-the target bookie id in the arguments for `-bookieid`
-`$ bin/bookkeeper shell decommissionbookie`
-or
-`$ bin/bookkeeper shell decommissionbookie -bookieid <target bookieid>`
+Minimum configuration changes required in `conf/bookkeeper.conf` are:
-4. Validate that there are no ledgers on decommissioned bookie
-`$ bin/bookkeeper shell listledgers -bookieid <target bookieid>`
+```properties
+# Change to point to journal disk mount point
+journalDirectory=data/bookkeeper/journal
-Last step to verify is you could run this command to check if the bookie you decommissioned doesn’t show up in list bookies:
+# Point to ledger storage disk mount point
+ledgerDirectories=data/bookkeeper/ledgers
-```bash
-./bookkeeper shell listbookies -rw -h
-./bookkeeper shell listbookies -ro -h
+# Point to local ZK quorum
+zkServers=zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181
+
+# Change the ledger manager type
+ledgerManagerType=hierarchical
```
+To change the zookeeper root path that Bookkeeper uses, use zkLedgersRootPath=/MY-PREFIX/ledgers instead of zkServers=localhost:2181/MY-PREFIX
+
+> Consult the official [BookKeeper docs](http://bookkeeper.apache.org) for more information about BookKeeper.
+
## BookKeeper persistence policies
In Pulsar, you can set *persistence policies*, at the namespace level, that determine how BookKeeper handles persistent storage of messages. Policies determine four things:
diff --git a/site2/website/versioned_docs/version-2.6.1/client-libraries-go.md b/site2/website/versioned_docs/version-2.6.1/client-libraries-go.md
index 7e1ec4c..b56c0d4 100644
--- a/site2/website/versioned_docs/version-2.6.1/client-libraries-go.md
+++ b/site2/website/versioned_docs/version-2.6.1/client-libraries-go.md
@@ -659,3 +659,22 @@ opts := pulsar.ClientOptions{
Authentication: NewAuthenticationTLS("my-cert.pem", "my-key.pem"),
}
```
+
+## OAuth2 authentication
+
+To use [OAuth2 authentication](security-oauth2.md), you'll need to configure your client to perform the following operations.
+This example shows how to configure OAuth2 authentication.
+
+```go
+oauth := pulsar.NewAuthenticationOAuth2(map[string]string{
+ "type": "client_credentials",
+ "issuerUrl": "https://dev-kt-aa9ne.us.auth0.com/oauth/token",
+ "audience": "https://dev-kt-aa9ne.us.auth0.com/api/v2/",
+ "privateKey": "/path/to/privateKey",
+ "clientId": "0Xx...Yyxeny",
+ })
+client, err := pulsar.NewClient(pulsar.ClientOptions{
+ URL: "puslar://my-cluster:6650",
+ Authentication: oauth,
+})
+```
diff --git a/site2/website/versioned_docs/version-2.6.1/client-libraries-java.md b/site2/website/versioned_docs/version-2.6.1/client-libraries-java.md
index 34dae18..8283b2b 100644
--- a/site2/website/versioned_docs/version-2.6.1/client-libraries-java.md
+++ b/site2/website/versioned_docs/version-2.6.1/client-libraries-java.md
@@ -380,33 +380,18 @@ ConsumerBuilder consumerBuilder = pulsarClient.newConsumer()
.subscriptionName(subscription);
// Subscribe to all topics in a namespace
-Pattern allTopicsInNamespace = Pattern.compile("public/default/.*");
+Pattern allTopicsInNamespace = Pattern.compile("persistent://public/default/.*");
Consumer allTopicsConsumer = consumerBuilder
.topicsPattern(allTopicsInNamespace)
.subscribe();
// Subscribe to a subsets of topics in a namespace, based on regex
-Pattern someTopicsInNamespace = Pattern.compile("public/default/foo.*");
+Pattern someTopicsInNamespace = Pattern.compile("persistent://public/default/foo.*");
Consumer allTopicsConsumer = consumerBuilder
.topicsPattern(someTopicsInNamespace)
.subscribe();
```
-In the above example, the consumer subscribes to the `persistent` topics that can match the topic name pattern. If you want the consumer subscribes to all `persistent` and `non-persistent` topics that can match the topic name pattern, set `subscriptionTopicsMode` to `RegexSubscriptionMode.AllTopics`.
-
-```java
-Pattern pattern = Pattern.compile("public/default/.*");
-pulsarClient.newConsumer()
- .subscriptionName("my-sub")
- .topicsPattern(pattern)
- .subscriptionTopicsMode(RegexSubscriptionMode.AllTopics)
- .subscribe();
-```
-
-> #### Note
->
-> By default, the `subscriptionTopicsMode` of the consumer is `PersistentOnly`. Available options of `subscriptionTopicsMode` are `PersistentOnly`, `NonPersistentOnly`, and `AllTopics`.
-
You can also subscribe to an explicit list of topics (across namespaces if you wish):
```java
@@ -646,11 +631,12 @@ Producer producer = client.newProducer()
## Reader
-With the [reader interface](concepts-clients.md#reader-interface), Pulsar clients can "manually position" themselves within a topic and reading all messages from a specified message onward. The Pulsar API for Java enables you to create {@inject: javadoc:Reader:/client/org/apache/pulsar/client/api/Reader} objects by specifying a topic and a {@inject: javadoc:MessageId:/client/org/apache/pulsar/client/api/MessageId}.
+With the [reader interface](concepts-clients.md#reader-interface), Pulsar clients can "manually position" themselves within a topic and reading all messages from a specified message onward. The Pulsar API for Java enables you to create {@inject: javadoc:Reader:/client/org/apache/pulsar/client/api/Reader} objects by specifying a topic, a {@inject: javadoc:MessageId:/client/org/apache/pulsar/client/api/MessageId}, and {@inject: javadoc:ReaderConfiguration:/client/org/apache/pulsar/client/a [...]
The following is an example.
```java
+ReaderConfiguration conf = new ReaderConfiguration();
byte[] msgIdBytes = // Some message ID byte array
MessageId id = MessageId.fromByteArray(msgIdBytes);
Reader reader = pulsarClient.newReader()
@@ -796,7 +782,7 @@ The following schema formats are currently available for Java:
## Authentication
-Pulsar currently supports two authentication schemes: [TLS](security-tls-authentication.md) and [Athenz](security-athenz.md). You can use the Pulsar Java client with both.
+Pulsar currently supports three authentication schemes: [TLS](security-tls-authentication.md), [Athenz](security-athenz.md), and [Oauth2](security-oauth2.md). You can use the Pulsar Java client with all of them.
### TLS Authentication
@@ -855,3 +841,28 @@ PulsarClient client = PulsarClient.builder()
> * `file:///path/to/file`
> * `file:/path/to/file`
> * `data:application/x-pem-file;base64,<base64-encoded value>`
+
+### Oauth2
+
+The following example shows how to use [Oauth2](security-oauth2.md) as an authentication provider for the Pulsar Java client.
+
+You can use the factory method to configure authentication for Pulsar Java client.
+
+```java
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://broker.example.com:6650/")
+ .authentication(
+ AuthenticationFactoryOAuth2.clientCredentials(this.issuerUrl, this.credentialsUrl, this.audience))
+ .build();
+```
+
+In addition, you can also use the encoded parameters to configure authentication for Pulsar Java client.
+
+```java
+Authentication auth = AuthenticationFactory
+ .create(AuthenticationOAuth2.class.getName(), "{"type":"client_credentials","privateKey":"...","issuerUrl":"...","audience":"..."}");
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://broker.example.com:6650/")
+ .authentication(auth)
+ .build();
+```
\ No newline at end of file
diff --git a/site2/website/versioned_docs/version-2.6.1/client-libraries-python.md b/site2/website/versioned_docs/version-2.6.1/client-libraries-python.md
index cf5e424..d41f013 100644
--- a/site2/website/versioned_docs/version-2.6.1/client-libraries-python.md
+++ b/site2/website/versioned_docs/version-2.6.1/client-libraries-python.md
@@ -87,33 +87,6 @@ while True:
client.close()
```
-This example shows how to configure negative acknowledgement.
-
-```python
-from pulsar import Client, schema
-client = Client('pulsar://localhost:6650')
-consumer = client.subscribe('negative_acks','test',schema=schema.StringSchema())
-producer = client.create_producer('negative_acks',schema=schema.StringSchema())
-for i in range(10):
- print('send msg "hello-%d"' % i)
- producer.send_async('hello-%d' % i, callback=None)
-producer.flush()
-for i in range(10):
- msg = consumer.receive()
- consumer.negative_acknowledge(msg)
- print('receive and nack msg "%s"' % msg.data())
-for i in range(10):
- msg = consumer.receive()
- consumer.acknowledge(msg)
- print('receive and ack msg "%s"' % msg.data())
-try:
- # No more messages expected
- msg = consumer.receive(100)
-except:
- print("no more msg")
- pass
-```
-
### Reader interface example
You can use the Pulsar Python API to use the Pulsar [reader interface](concepts-clients.md#reader-interface). Here's an example:
diff --git a/site2/website/versioned_docs/version-2.6.1/concepts-clients.md b/site2/website/versioned_docs/version-2.6.1/concepts-clients.md
index a92ddc8..f5c2ca2 100644
--- a/site2/website/versioned_docs/version-2.6.1/concepts-clients.md
+++ b/site2/website/versioned_docs/version-2.6.1/concepts-clients.md
@@ -32,7 +32,7 @@ The **reader interface** for Pulsar enables applications to manually manage curs
* The **latest** available message in the topic
* Some other message between the earliest and the latest. If you select this option, you'll need to explicitly provide a message ID. Your application will be responsible for "knowing" this message ID in advance, perhaps fetching it from a persistent data store or cache.
-The reader interface is helpful for use cases like using Pulsar to provide [effectively-once](https://streaml.io/blog/exactly-once/) processing semantics for a stream processing system. For this use case, it's essential that the stream processing system be able to "rewind" topics to a specific message and begin reading there. The reader interface provides Pulsar clients with the low-level abstraction necessary to "manually position" themselves within a topic.
+The reader interface is helpful for use cases like using Pulsar to provide effectively-once processing semantics for a stream processing system. For this use case, it's essential that the stream processing system be able to "rewind" topics to a specific message and begin reading there. The reader interface provides Pulsar clients with the low-level abstraction necessary to "manually position" themselves within a topic.
Internally, the reader interface is implemented as a consumer using an exclusive, non-durable subscription to the topic with a randomly-allocated name.
diff --git a/site2/website/versioned_docs/version-2.6.1/concepts-messaging.md b/site2/website/versioned_docs/version-2.6.1/concepts-messaging.md
index b0b1a6e..afa019c 100644
--- a/site2/website/versioned_docs/version-2.6.1/concepts-messaging.md
+++ b/site2/website/versioned_docs/version-2.6.1/concepts-messaging.md
@@ -1,47 +1,47 @@
---
id: version-2.6.1-concepts-messaging
-title: Messaging Concepts
+title: Messaging
sidebar_label: Messaging
original_id: concepts-messaging
---
Pulsar is built on the [publish-subscribe](https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern) pattern (short for pub-sub). In this pattern, [producers](#producers) publish messages to [topics](#topics). [Consumers](#consumers) [subscribe](#subscription-modes) to those topics, process incoming messages, and send an acknowledgement when processing is complete.
-Once a subscription has been created, all messages are [retained](concepts-architecture-overview.md#persistent-storage) by Pulsar, even if the consumer gets disconnected. Retained messages are discarded only when a consumer acknowledges that those messages are processed successfully.
+When a subscription is created, Pulsar [retains](concepts-architecture-overview.md#persistent-storage) all messages, even if the consumer is disconnected. Retained messages are discarded only when a consumer acknowledges that those messages are processed successfully.
## Messages
-Messages are the basic "unit" of Pulsar. Messages are what producers publish to topics and what consumers then consume from topics (and acknowledge when the message has been processed). Messages are the analogue of letters in a postal service system.
+Messages are the basic "unit" of Pulsar. The following table lists the components of messages.
-Component | Purpose
+Component | Description
:---------|:-------
-Value / data payload | The data carried by the message. All Pulsar messages carry raw bytes, although message data can also conform to data [schemas](schema-get-started.md).
-Key | Messages can optionally be tagged with keys, which can be useful for things like [topic compaction](concepts-topic-compaction.md).
+Value / data payload | The data carried by the message. All Pulsar messages contain raw bytes, although message data can also conform to data [schemas](schema-get-started.md).
+Key | Messages are optionally tagged with keys, which is useful for things like [topic compaction](concepts-topic-compaction.md).
Properties | An optional key/value map of user-defined properties.
-Producer name | The name of the producer that produced the message (producers are automatically given default names, but you can apply your own explicitly as well).
-Sequence ID | Each Pulsar message belongs to an ordered sequence on its topic. A message's sequence ID is its ordering in that sequence.
-Publish time | The timestamp of when the message was published (automatically applied by the producer).
-Event time | An optional timestamp that applications can attach to the message representing when something happened, for example, when the message was processed. The event time of a message is 0 if none is explicitly set.
-TypedMessageBuilder | `TypedMessageBuilder` is used to construct a message. You can set message properties like the message key, message value with `TypedMessageBuilder`. </br> When you set `TypedMessageBuilder`, the best practice is to set the key as a string. If you set the key as other types, for example, an AVRO object, the key is sent as bytes, and it is difficult to get the AVRO object back on the consumer.
+Producer name | The name of the producer who produces the message. If you do not specify a producer name, the default name is used.
+Sequence ID | Each Pulsar message belongs to an ordered sequence on its topic. The sequence ID of the message is its order in that sequence.
+Publish time | The timestamp of when the message is published. The timestamp is automatically applied by the producer.
+Event time | An optional timestamp attached to a message by applications. For example, applications attach a timestamp on when the message is processed. If nothing is set to event time, the value is `0`.
+TypedMessageBuilder | It is used to construct a message. You can set message properties such as the message key, message value with `TypedMessageBuilder`. </br> When you set `TypedMessageBuilder`, set the key as a string. If you set the key as other types, for example, an AVRO object, the key is sent as bytes, and it is difficult to get the AVRO object back on the consumer.
-> For a more in-depth breakdown of Pulsar message contents, see Pulsar [binary protocol](developing-binary-protocol.md).
+> For more information on Pulsar message contents, see Pulsar [binary protocol](developing-binary-protocol.md).
## Producers
-A producer is a process that attaches to a topic and publishes messages to a Pulsar [broker](reference-terminology.md#broker) for processing.
+A producer is a process that attaches to a topic and publishes messages to a Pulsar [broker](reference-terminology.md#broker). The Pulsar broker process the messages.
### Send modes
-Producers can send messages to brokers either synchronously (sync) or asynchronously (async).
+Producers send messages to brokers synchronously (sync) or asynchronously (async).
-| Mode | Description |
-|:-----------|:-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| Sync send | The producer will wait for acknowledgement from the broker after sending each message. If acknowledgment isn't received then the producer will consider the send operation a failure. |
-| Async send | The producer will put the message in a blocking queue and return immediately. The client library will then send the message to the broker in the background. If the queue is full (max size [configurable](reference-configuration.md#broker)), the producer could be blocked or fail immediately when calling the API, depending on arguments passed to the producer. |
+| Mode | Description |
+|:-----------|-----------|
+| Sync send | The producer waits for an acknowledgement from the broker after sending every message. If the acknowledgment is not received, the producer treats the sending operation as a failure. |
+| Async send | The producer puts a message in a blocking queue and returns immediately. The client library sends the message to the broker in the background. If the queue is full (you can [configure](reference-configuration.md#broker) the maximum size), the producer is blocked or fails immediately when calling the API, depending on arguments passed to the producer. |
### Compression
-Messages published by producers can be compressed during transportation in order to save bandwidth. Pulsar currently supports the following types of compression:
+You can compress messages published by producers during transportation. Pulsar currently supports the following types of compression:
* [LZ4](https://github.com/lz4/lz4)
* [ZLIB](https://zlib.net/)
@@ -52,28 +52,52 @@ Messages published by producers can be compressed during transportation in order
When batching is enabled, the producer accumulates and sends a batch of messages in a single request. The batch size is defined by the maximum number of messages and the maximum publish latency. Therefore, the backlog size represents the total number of batches instead of the total number of messages.
-In Pulsar, batches are tracked and stored as single units rather than as individual messages. Under the hood, the consumer unbundles a batch into individual messages. However, scheduled messages (configured through the `deliverAt` or the `deliverAfter` parameter) are always sent as individual messages even batching is enabled.
+In Pulsar, batches are tracked and stored as single units rather than as individual messages. Consumer unbundles a batch into individual messages. However, scheduled messages (configured through the `deliverAt` or the `deliverAfter` parameter) are always sent as individual messages even batching is enabled.
-In general, a batch is acknowledged when all its messages are acknowledged by the consumer. This means unexpected failures, negative acknowledgements, or acknowledgement timeouts can result in redelivery of all messages in a batch, even if some of the messages have already been acknowledged.
+In general, a batch is acknowledged when all of its messages are acknowledged by a consumer. It means unexpected failures, negative acknowledgements, or acknowledgement timeouts can result in redelivery of all messages in a batch, even if some of the messages are acknowledged.
To avoid redelivering acknowledged messages in a batch to the consumer, Pulsar introduces batch index acknowledgement since Pulsar 2.6.0. When batch index acknowledgement is enabled, the consumer filters out the batch index that has been acknowledged and sends the batch index acknowledgement request to the broker. The broker maintains the batch index acknowledgement status and tracks the acknowledgement status of each batch index to avoid dispatching acknowledged messages to the consumer [...]
-By default, batch index acknowledgement is disabled (`batchIndexAcknowledgeEnable=false`). You can enable batch index acknowledgement by setting the `batchIndexAcknowledgeEnable` parameter to `true` at the broker side. Enabling batch index acknowledgement may bring more memory overheads. So, perform this operation with caution.
+By default, batch index acknowledgement is disabled (`batchIndexAcknowledgeEnable=false`). You can enable batch index acknowledgement by setting the `batchIndexAcknowledgeEnable` parameter to `true` at the broker side. Enabling batch index acknowledgement results in more memory overheads.
+
+### Chunking
+When 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 the exclusive and failover subscription modes.
+
+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 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 `maxPendingChuckedMessage` param [...]
+
+ The broker does not require any changes to support chunking for non-shared subscription. The broker only uses `chuckedMessageRate` to record chunked message rate on the topic.
+
+#### 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, combi [...]
+
+
+
+#### Handle chunked messages with multiple producers and 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 c [...]
+
+
## Consumers
A consumer is a process that attaches to a topic via a subscription and then receives messages.
-A consumer sends a [flow permit request](developing-binary-protocol.md#flow-control) to a broker to get messages. There is a queue at the consumer side to receive messages pushed from the broker. The queue size is configurable by [`receiverQueueSize`](client-libraries-java.md#configure-consumer) (default: 1000). Each time `consumer.receive()` is called, a message is dequeued from the buffer.
+A consumer sends a [flow permit request](developing-binary-protocol.md#flow-control) to a broker to get messages. There is a queue at the consumer side to receive messages pushed from the broker. You can configure the queue size with the [`receiverQueueSize`](client-libraries-java.md#configure-consumer) parameter. The default size is `1000`). Each time `consumer.receive()` is called, a message is dequeued from the buffer.
### Receive modes
-Messages can be received from [brokers](reference-terminology.md#broker) either synchronously (sync) or asynchronously (async).
+Messages are received from [brokers](reference-terminology.md#broker) either synchronously (sync) or asynchronously (async).
| Mode | Description |
|:--------------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| Sync receive | A sync receive will be blocked until a message is available. |
-| Async receive | An async receive will return immediately with a future value---a [`CompletableFuture`](http://www.baeldung.com/java-completablefuture) in Java, for example---that completes once a new message is available. |
+| Sync receive | A sync receive is blocked until a message is available. |
+| Async receive | An async receive returns immediately with a future value—for example, a [`CompletableFuture`](http://www.baeldung.com/java-completablefuture) in Java—that completes once a new message is available. |
### Listeners
@@ -81,40 +105,40 @@ Client libraries provide listener implementation for consumers. For example, the
### Acknowledgement
-When a consumer has consumed a message successfully, the consumer sends an acknowledgement request to the broker. This message is permanently [stored](concepts-architecture-overview.md#persistent-storage) and then it is deleted only after all the subscriptions have acknowledged it. If you want to store the message that has been acknowledged by a consumer, you need to configure the [message retention policy](concepts-messaging.md#message-retention-and-expiry).
+When a consumer consumes a message successfully, the consumer sends an acknowledgement request to the broker. This message is permanently stored, and then deleted only after all the subscriptions have acknowledged it. If you want to store the message that has been acknowledged by a consumer, you need to configure the [message retention policy](concepts-messaging.md#message-retention-and-expiry).
For a batch message, if batch index acknowledgement is enabled, the broker maintains the batch index acknowledgement status and tracks the acknowledgement status of each batch index to avoid dispatching acknowledged messages to the consumer. When all indexes of the batch message are acknowledged, the batch message is deleted. For details about the batch index acknowledgement, see [batching](#batching).
-Messages can be acknowledged either one by one or cumulatively. With cumulative acknowledgement, the consumer only needs to acknowledge the last message it received. All messages in the stream up to (and including) the provided message will not be re-delivered to that consumer.
+Messages is acknowledged either one by one or cumulatively. With cumulative acknowledgement, the consumer only needs to acknowledge the last message it received. All messages in the stream up to (and including) the provided message will not be re-delivered to that consumer.
Messages can be acknowledged in the following two ways:
- Messages are acknowledged individually. With individual acknowledgement, the consumer needs to acknowledge each message and sends an acknowledgement request to the broker.
- Messages are acknowledged cumulatively. With cumulative acknowledgement, the consumer only needs to acknowledge the last message it received. All messages in the stream up to (and including) the provided message are not re-delivered to that consumer.
-> #### Note
+> Note
>
-> Cumulative acknowledgement cannot be used in the [shared subscription mode](#subscription-modes), because the shared subscription mode involves multiple consumers having access to the same subscription. In the shared subscription mode, messages can be acknowledged individually.
+> Cumulative acknowledgement cannot be used in the [shared subscription mode](#subscription-modes), because the shared subscription mode involves multiple consumers who have access to the same subscription. In the shared subscription mode, messages are acknowledged individually.
### Negative acknowledgement
-When a consumer does not consume a message successfully at a time, and wants to consume the message again, the consumer can send a negative acknowledgement to the broker, and then the broker will redeliver the message.
+When a consumer does not consume a message successfully at a time, and wants to consume the message again, the consumer sends a negative acknowledgement to the broker, and then the broker redelivers the message.
-Messages can be negatively acknowledged one by one or cumulatively, which depends on the consumption subscription mode.
+Messages are negatively acknowledged one by one or cumulatively, which depends on the consumption subscription mode.
-In the exclusive and failover subscription modes, consumers only negatively acknowledge the last message they have received.
+In the exclusive and failover subscription modes, consumers only negatively acknowledge the last message they receive.
In the shared and Key_Shared subscription modes, you can negatively acknowledge messages individually.
> Note
-> If batching is enabled, other messages in the same batch may be redelivered to the consumer as well as the negatively acknowledged messages.
+> If batching is enabled, other messages and the negatively acknowledged messages in the same batch are redelivered to the consumer.
### Acknowledgement timeout
-When a message is not consumed successfully, and you want to trigger the broker to redeliver the message automatically, you can adopt the unacknowledged message automatic re-delivery mechanism. Client will track the unacknowledged messages within the entire `acktimeout` time range, and send a `redeliver unacknowledged messages` request to the broker automatically when the acknowledgement timeout is specified.
+If a message is not consumed successfully, and you want to trigger the broker to redeliver the message automatically, you can adopt the unacknowledged message automatic re-delivery mechanism. Client tracks the unacknowledged messages within the entire `acktimeout` time range, and sends a `redeliver unacknowledged messages` request to the broker automatically when the acknowledgement timeout is specified.
> Note
-> If batching is enabled, other messages in the same batch may be redelivered to the consumer as well as the unacknowledged messages.
+> If batching is enabled, other messages and the unacknowledged messages in the same batch are redelivered to the consumer.
> Note
> Prefer negative acknowledgements over acknowledgement timeout. Negative acknowledgement controls the re-delivery of individual messages with more precision, and avoids invalid redeliveries when the message processing time exceeds the acknowledgement timeout.
@@ -163,11 +187,11 @@ Dead letter topic depends on message re-delivery. Messages are redelivered eithe
### Retry letter topic
-For many online business systems, a message needs to be re-consumed because any exception occurs in the business logic processing. Generally, users hope that they can flexibly configure the delay time for re-consuming the failed messages. In this case, you can configure the producer to send messages to both the business topic and the retry letter topic and you can enable automatic retry on the consumer. When automatic retry is enabled on the consumer, a message is stored in the retry let [...]
+For many online business systems, a message is re-consumed due to exception occurs in the business logic processing. To configure the delay time for re-consuming the failed messages, you can configure the producer to send messages to both the business topic and the retry letter topic, and enable automatic retry on the consumer. When automatic retry is enabled on the consumer, a message is stored in the retry letter topic if the messages are not consumed, and therefore the consumer automa [...]
By default, automatic retry is disabled. You can set `enableRetry` to `true` to enable automatic retry on the consumer.
-This example shows how to consumer messages from a retry letter topic.
+This example shows how to consume messages from a retry letter topic.
```java
Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
@@ -186,7 +210,7 @@ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
## Topics
-As in other pub-sub systems, topics in Pulsar are named channels for transmitting messages from [producers](reference-terminology.md#producer) to [consumers](reference-terminology.md#consumer). Topic names are URLs that have a well-defined structure:
+As in other pub-sub systems, topics in Pulsar are named channels for transmitting messages from producers to consumers. Topic names are URLs that have a well-defined structure:
```http
{persistent|non-persistent}://tenant/namespace/topic
@@ -194,34 +218,34 @@ As in other pub-sub systems, topics in Pulsar are named channels for transmittin
Topic name component | Description
:--------------------|:-----------
-`persistent` / `non-persistent` | This identifies the type of topic. Pulsar supports two kind of topics: [persistent](concepts-architecture-overview.md#persistent-storage) and [non-persistent](#non-persistent-topics) (persistent is the default, so if you don't specify a type the topic will be persistent). With persistent topics, all messages are durably [persisted](concepts-architecture-overview.md#persistent-storage) on disk (that means on multiple disks unless the broker is standalone) [...]
-`tenant` | The topic's tenant within the instance. Tenants are essential to multi-tenancy in Pulsar and can be spread across clusters.
-`namespace` | The administrative unit of the topic, which acts as a grouping mechanism for related topics. Most topic configuration is performed at the [namespace](#namespaces) level. Each tenant can have multiple namespaces.
-`topic` | The final part of the name. Topic names are freeform and have no special meaning in a Pulsar instance.
-
+`persistent` / `non-persistent` | This identifies the type of topic. Pulsar supports two kind of topics: [persistent](concepts-architecture-overview.md#persistent-storage) and [non-persistent](#non-persistent-topics). The default is persistent, so if you do not specify a type, the topic is 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 topi [...]
+`tenant` | The topic tenant within the instance. Tenants are essential to multi-tenancy in Pulsar, and spread across clusters.
+`namespace` | The administrative unit of the topic, which acts as a grouping mechanism for related topics. Most topic configuration is performed at the [namespace](#namespaces) level. Each tenant has one or multiple namespaces.
+`topic` | The final part of the name. Topic names have no special meaning in a Pulsar instance.
> #### No need to explicitly create new topics
-> You don't need to explicitly create topics in Pulsar. If a client attempts to write or receive messages to/from a topic that does not yet exist, Pulsar will automatically create that topic under the [namespace](#namespaces) provided in the [topic name](#topics).
+> You do not need to explicitly create topics in Pulsar. If a client attempts to write or receive messages to/from a topic that does not yet exist, Pulsar creates that topic under the namespace provided in the [topic name](#topics) automatically.
+> If no tenant or namespace is specified when a client creates a topic, the topic is created in the default tenant and namespace. You can also create a topic in a specified tenant and namespace, such as `persistent://my-tenant/my-namespace/my-topic`. `persistent://my-tenant/my-namespace/my-topic` means the `my-topic` topic is created in the `my-namespace` namespace of the `my-tenant` tenant.
## Namespaces
-A namespace is a logical nomenclature within a tenant. A tenant can create multiple namespaces via the [admin API](admin-api-namespaces.md#create). For instance, a tenant with different applications can create a separate namespace for each application. A namespace allows the application to create and manage a hierarchy of topics. The topic `my-tenant/app1` is a namespace for the application `app1` for `my-tenant`. You can create any number of [topics](#topics) under the namespace.
+A namespace is a logical nomenclature within a tenant. A tenant creates multiple namespaces via the [admin API](admin-api-namespaces.md#create). For instance, a tenant with different applications can create a separate namespace for each application. A namespace allows the application to create and manage a hierarchy of topics. The topic `my-tenant/app1` is a namespace for the application `app1` for `my-tenant`. You can create any number of [topics](#topics) under the namespace.
## Subscriptions
-A subscription is a named configuration rule that determines how messages are delivered to consumers. There are four available subscription modes in Pulsar: [exclusive](#exclusive), [shared](#shared), [failover](#failover), and [key_shared](#key_shared). These modes are illustrated in the figure below.
+A subscription is a named configuration rule that determines how messages are delivered to consumers. Four subscription modes are available in Pulsar: [exclusive](#exclusive), [shared](#shared), [failover](#failover), and [key_shared](#key_shared). These modes are illustrated in the figure below.
-> #### Pub-Sub, Queuing, or Both
-> There is a lot of flexibility in how to combine subscriptions:
-> * If you want to achieve traditional "fan-out pub-sub messaging" among consumers, you can make each consumer have a unique subscription name (exclusive)
-> * If you want to achieve "message queuing" among consumers, you can make multiple consumers have the same subscription name (shared, failover, key_shared)
-> * If you want to do both simultaneously, you can have some consumers with exclusive subscriptions while others do not
+> ### Pub-Sub or Queuing
+> In Pulsar, you can use different subscriptions flexibly.
+> * If you want to achieve traditional "fan-out pub-sub messaging" among consumers, specify a unique subscription name for each consumer. It is exclusive subscription mode.
+> * If you want to achieve "message queuing" among consumers, share the same subscription name among multiple consumers(shared, failover, key_shared).
+> * If you want to achieve both effects simultaneously, combine exclusive subscription mode with other subscription modes for consumers.
### Exclusive
-In *exclusive* mode, only a single consumer is allowed to attach to the subscription. If more than one consumer attempts to subscribe to a topic using the same subscription, the consumer receives an error.
+In *exclusive* mode, only a single consumer is allowed to attach to the subscription. If multiple consumers subscribe to a topic using the same subscription, an error occurs.
In the diagram below, only **Consumer A-0** is allowed to consume messages.
@@ -257,7 +281,7 @@ In the diagram below, **Consumer-C-1** and **Consumer-C-2** are able to subscrib
In *Key_Shared* mode, multiple consumers can attach to the same subscription. Messages are delivered in a distribution across consumers and message with same key or same ordering key are delivered to only one consumer. No matter how many times the message is re-delivered, it is delivered to the same consumer. When a consumer connected or disconnected will cause served consumer change for some key of message.
> #### Limitations of Key_Shared mode
-> When using Key_Shared mode, be aware that:
+> When you use Key_Shared mode, be aware that:
> * You need to specify a key or orderingKey for messages
> * You cannot use cumulative acknowledgment with Key_Shared mode.
@@ -272,14 +296,14 @@ When a consumer subscribes to a Pulsar topic, by default it subscribes to one sp
* On the basis of a [**reg**ular **ex**pression](https://en.wikipedia.org/wiki/Regular_expression) (regex), for example `persistent://public/default/finance-.*`
* By explicitly defining a list of topics
-> When subscribing to multiple topics by regex, all topics must be in the same [namespace](#namespaces)
+> When subscribing to multiple topics by regex, all topics must be in the same [namespace](#namespaces).
-When subscribing to multiple topics, the Pulsar client will automatically make a call to the Pulsar API to discover the topics that match the regex pattern/list and then subscribe to all of them. If any of the topics don't currently exist, the consumer will auto-subscribe to them once the topics are created.
+When subscribing to multiple topics, the Pulsar client automatically makes a call to the Pulsar API to discover the topics that match the regex pattern/list, and then subscribe to all of them. If any of the topics do not exist, the consumer auto-subscribes to them once the topics are created.
> #### No ordering guarantees across multiple topics
> When a producer sends messages to a single topic, all messages are guaranteed to be read from that topic in the same order. However, these guarantees do not hold across multiple topics. So when a producer sends message to multiple topics, the order in which messages are read from those topics is not guaranteed to be the same.
-Here are some multi-topic subscription examples for Java:
+The following are multi-topic subscription examples for Java.
```java
import java.util.regex.Pattern;
@@ -304,21 +328,19 @@ Consumer<byte[]> someTopicsConsumer = pulsarClient.newConsumer()
.subscribe();
```
-For code examples, see:
-
-* [Java](client-libraries-java.md#multi-topic-subscriptions)
+For code examples, see [Java](client-libraries-java.md#multi-topic-subscriptions).
## Partitioned topics
-Normal topics can be served only by a single broker, which limits the topic's maximum throughput. *Partitioned topics* are a special type of topic that can be handled by multiple brokers, which allows for much higher throughput.
+Normal topics are served only by a single broker, which limits the maximum throughput of the topic. *Partitioned topics* are a special type of topic that are handled by multiple brokers, thus allowing for higher throughput.
-Behind the scenes, a partitioned topic is actually implemented as N internal topics, where N is the number of partitions. When publishing messages to a partitioned topic, each message is routed to one of several brokers. The distribution of partitions across brokers is handled automatically by Pulsar.
+A partitioned topic is actually implemented as N internal topics, where N is the number of partitions. When publishing messages to a partitioned topic, each message is routed to one of several brokers. The distribution of partitions across brokers is handled automatically by Pulsar.
The diagram below illustrates this:
-Here, the topic **Topic1** has five partitions (**P0** through **P4**) split across three brokers. Because there are more partitions than brokers, two brokers handle two partitions a piece, while the third handles only one (again, Pulsar handles this distribution of partitions automatically).
+The **Topic1** topic has five partitions (**P0** through **P4**) split across three brokers. Because there are more partitions than brokers, two brokers handle two partitions a piece, while the third handles only one (again, Pulsar handles this distribution of partitions automatically).
Messages for this topic are broadcast to two consumers. The [routing mode](#routing-modes) determines each message should be published to which partition, while the [subscription mode](#subscription-modes) determines which messages go to which consumers.
@@ -437,7 +459,7 @@ With message expiry, shown at the bottom, some messages are <span style="color:
## Message deduplication
-Message **duplication** occurs when a message is [persisted](concepts-architecture-overview.md#persistent-storage) by Pulsar more than once. Message ***de*duplication** is an optional Pulsar feature that prevents unnecessary message duplication by processing each message only once, *even if the message is received more than once*.
+Message duplication occurs when a message is persisted](concepts-architecture-overview.md#persistent-storage) by Pulsar more than once. Message deduplication is an optional Pulsar feature that prevents unnecessary message duplication by processing each message only once, even if the message is received more than once.
The following diagram illustrates what happens when message deduplication is disabled vs. enabled:
@@ -453,18 +475,18 @@ In the second scenario at the bottom, the producer publishes message 1, which is
### Producer idempotency
-The other available approach to message deduplication is to ensure that each message is *only produced once*. This approach is typically called **producer idempotency**. The drawback of this approach is that it defers the work of message deduplication to the application. In Pulsar, this is handled at the [broker](reference-terminology.md#broker) level, which means that you don't need to modify your Pulsar client code. Instead, you only need to make administrative changes (see the [Managi [...]
+The other available approach to message deduplication is to ensure that each message is *only produced once*. This approach is typically called **producer idempotency**. The drawback of this approach is that it defers the work of message deduplication to the application. In Pulsar, this is handled at the [broker](reference-terminology.md#broker) level, so you do not need to modify your Pulsar client code. Instead, you only need to make administrative changes. For details, see [Managing m [...]
### Deduplication and effectively-once semantics
-Message deduplication makes Pulsar an ideal messaging system to be used in conjunction with stream processing engines (SPEs) and other systems seeking to provide [effectively-once](https://streaml.io/blog/exactly-once) processing semantics. Messaging systems that don't offer automatic message deduplication require the SPE or other system to guarantee deduplication, which means that strict message ordering comes at the cost of burdening the application with the responsibility of deduplica [...]
+Message deduplication makes Pulsar an ideal messaging system to be used in conjunction with stream processing engines (SPEs) and other systems seeking to provide effectively-once processing semantics. Messaging systems that do not offer automatic message deduplication require the SPE or other system to guarantee deduplication, which means that strict message ordering comes at the cost of burdening the application with the responsibility of deduplication. With Pulsar, strict ordering guar [...]
-> More in-depth information can be found in [this post](https://streaml.io/blog/pulsar-effectively-once/) on the [Streamlio blog](https://streaml.io/blog)
+> You can find more in-depth information in [this post](https://www.splunk.com/en_us/blog/it/exactly-once-is-not-exactly-the-same.html).
## Delayed message delivery
Delayed message delivery enables you to consume a message later rather than immediately. In this mechanism, a message is stored in BookKeeper, `DelayedDeliveryTracker` maintains the time index(time -> messageId) in memory after published to a broker, and it is delivered to a consumer once the specific delayed time is passed.
-Delayed message delivery only works well in Shared subscription mode. In Exclusive and Failover subscription mode, the delayed message is dispatched immediately.
+Delayed message delivery only works in Shared subscription mode. In Exclusive and Failover subscription modes, the delayed message is dispatched immediately.
The diagram below illustrates the concept of delayed message delivery:
diff --git a/site2/website/versioned_docs/version-2.6.1/concepts-proxy-sni-routing.md b/site2/website/versioned_docs/version-2.6.1/concepts-proxy-sni-routing.md
new file mode 100644
index 0000000..6f7ed80
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.6.1/concepts-proxy-sni-routing.md
@@ -0,0 +1,121 @@
+---
+id: version-2.6.1-concepts-proxy-sni-routing
+title: Proxy support with SNI routing
+sidebar_label: Proxy support with SNI routing
+original_id: concepts-proxy-sni-routing
+---
+
+## Pulsar Proxy with SNI routing
+A proxy server is an intermediary server that forwards requests from multiple clients to different servers across the Internet. The proxy server acts as a "traffic cop" in both forward and reverse proxy scenarios, and benefits your system such as load balancing, performance, security, auto-scaling, and so on.
+
+The proxy in Pulsar acts as a reverse proxy, and creates a gateway in front of brokers. Proxies such as Apache Traffic Server (ATS), HAProxy, Nginx, and Envoy are not supported in Pulsar. These proxy-servers support **SNI routing**. SNI routing is used to route traffic to a destination without terminating the SSL connection. Layer 4 routing provides greater transparency because the outbound connection is determined by examining the destination address in the client TCP packets.
+
+Pulsar clients support [SNI routing protocol](https://github.com/apache/pulsar/wiki/PIP-60:-Support-Proxy-server-with-SNI-routing), so you can connect to brokers through the proxy. This document walks you through how to set up the ATS proxy, enable SNI routing, and connect Pulsar client to the broker through the ATS proxy.
+
+### ATS-SNI Routing in Pulsar
+To support [layer-4 SNI routing](https://docs.trafficserver.apache.org/en/latest/admin-guide/layer-4-routing.en.html) with ATS, the inbound connection must be a TLS connection. Pulsar client supports SNI routing protocol on TLS connection, so when Pulsar clients connect to broker through ATS proxy, Pulsar uses ATS as a reverse proxy.
+
+Pulsar supports SNI routing for geo-replication, so brokers can connect to brokers in other clusters through the ATS proxy.
+
+This section explains how to set up and use ATS as a reverse proxy, so Pulsar clients can connect to brokers through the ATS proxy using the SNI routing protocol on TLS connection.
+
+#### Set up ATS Proxy for layer-4 SNI routing
+To support layer 4 SNI routing, you need to configure the `records.conf` and `ssl_server_name.conf` files.
+
+
+
+The [records.config](https://docs.trafficserver.apache.org/en/latest/admin-guide/files/records.config.en.html) file is located in the `/usr/local/etc/trafficserver/` directory by default. The file lists configurable variables used by the ATS.
+
+To configure the `records.config` files, complete the following steps.
+1. Update TLS port (`http.server_ports`) on which proxy listens, and update proxy certs (`ssl.client.cert.path` and `ssl.client.cert.filename`) to secure TLS tunneling.
+2. Configure server ports (`http.connect_ports`) used for tunneling to the broker. If Pulsar brokers are listening on `4443` and `6651` ports, add the brokers service port in the `http.connect_ports` configuration.
+
+The following is an example.
+
+```
+# PROXY TLS PORT
+CONFIG proxy.config.http.server_ports STRING 4443:ssl 4080
+# PROXY CERTS FILE PATH
+CONFIG proxy.config.ssl.client.cert.path STRING /proxy-cert.pem
+# PROXY KEY FILE PATH
+CONFIG proxy.config.ssl.client.cert.filename STRING /proxy-key.pem
+
+
+# The range of origin server ports that can be used for tunneling via CONNECT. # Traffic Server allows tunnels only to the specified ports. Supports both wildcards (*) and ranges (e.g. 0-1023).
+CONFIG proxy.config.http.connect_ports STRING 4443 6651
+```
+
+The [ssl_server_name](https://docs.trafficserver.apache.org/en/8.0.x/admin-guide/files/ssl_server_name.yaml.en.html) file is used to configure TLS connection handling for inbound and outbound connections. The configuration is determined by the SNI values provided by the inbound connection. The file consists of a set of configuration items, and each is identified by an SNI value (`fqdn`). When an inbound TLS connection is made, the SNI value from the TLS negotiation is matched with the it [...]
+
+The following example shows mapping of the inbound SNI hostname coming from the client, and the actual broker service URL where request should be redirected. For example, if the client sends the SNI header `pulsar-broker1`, the proxy creates a TLS tunnel by redirecting request to the `pulsar-broker1:6651` service URL.
+
+```
+server_config = {
+ {
+ fqdn = 'pulsar-broker-vip',
+ # Forward to Pulsar broker which is listening on 6651
+ tunnel_route = 'pulsar-broker-vip:6651'
+ },
+ {
+ fqdn = 'pulsar-broker1',
+ # Forward to Pulsar broker-1 which is listening on 6651
+ tunnel_route = 'pulsar-broker1:6651'
+ },
+ {
+ fqdn = 'pulsar-broker2',
+ # Forward to Pulsar broker-2 which is listening on 6651
+ tunnel_route = 'pulsar-broker2:6651'
+ },
+}
+```
+
+After you configure the `ssl_server_name.config` and `records.config` files, the ATS-proxy server handles SNI routing and creates TCP tunnel between the client and the broker.
+
+#### Configure Pulsar-client with SNI routing
+ATS SNI-routing works only with TLS. You need to enable TLS for the ATS proxy and brokers first, configure the SNI routing protocol, and then connect Pulsar clients to brokers through ATS proxy. Pulsar clients support SNI routing by connecting to the proxy, and sending the target broker URL to the SNI header. This process is processed internally. You only need to configure the following proxy configuration initially when you create a Pulsar client to use the SNI routing protocol.
+
+```
+String brokerServiceUrl = “pulsar+ssl://pulsar-broker-vip:6651/”;
+String proxyUrl = “pulsar+ssl://ats-proxy:443”;
+ClientBuilder clientBuilder = PulsarClient.builder()
+ .serviceUrl(brokerServiceUrl)
+ .tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH)
+ .enableTls(true)
+ .allowTlsInsecureConnection(false)
+ .proxyServiceUrl(proxyUrl, ProxyProtocol.SNI)
+ .operationTimeout(1000, TimeUnit.MILLISECONDS);
+
+Map<String, String> authParams = new HashMap<>();
+authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH);
+authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH);
+clientBuilder.authentication(AuthenticationTls.class.getName(), authParams);
+
+PulsarClient pulsarClient = clientBuilder.build();
+```
+
+#### Pulsar geo-replication with SNI routing
+You can use the ATS proxy for geo-replication. Pulsar brokers can connect to brokers in geo-replication by using SNI routing. To enable SNI routing for broker connection cross clusters, you need to configure SNI proxy URL to the cluster metadata. If you have configured SNI proxy URL in the cluster metadata, you can connect to broker cross clusters through the proxy over SNI routing.
+
+
+
+In this example, a Pulsar cluster is deployed into two separate regions, `us-west` and `us-east`. Both regions are configured with ATS proxy, and brokers in each region run behind the ATS proxy. We configure the cluster metadata for both clusters, so brokers in one cluster can use SNI routing and connect to brokers in other clusters through the ATS proxy.
+
+(a) Configure the cluster metadata for `us-east` with `us-east` broker service URL and `us-east` ATS proxy URL with SNI proxy-protocol.
+
+```
+./pulsar-admin clusters update \
+--broker-url-secure pulsar+ssl://east-broker-vip:6651 \
+--url http://east-broker-vip:8080 \
+--proxy-protocol SNI \
+--proxy-url pulsar+ssl://east-ats-proxy:443
+```
+
+(b) Configure the cluster metadata for `us-west` with `us-west` broker service URL and `us-west` ATS proxy URL with SNI proxy-protocol.
+
+```
+./pulsar-admin clusters update \
+--broker-url-secure pulsar+ssl://west-broker-vip:6651 \
+--url http://west-broker-vip:8080 \
+--proxy-protocol SNI \
+--proxy-url pulsar+ssl://west-ats-proxy:443
+```
diff --git a/site2/website/versioned_docs/version-2.6.1/concepts-schema-registry.md b/site2/website/versioned_docs/version-2.6.1/concepts-schema-registry.md
new file mode 100644
index 0000000..b41470f
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.6.1/concepts-schema-registry.md
@@ -0,0 +1,6 @@
+---
+id: version-2.6.1-concepts-schema-registry
+title: Schema Registry
+sidebar_label: Schema Registry
+original_id: concepts-schema-registry
+---
diff --git a/site2/website/versioned_docs/version-2.6.1/cookbooks-deduplication.md b/site2/website/versioned_docs/version-2.6.1/cookbooks-deduplication.md
index 688f792..7faf0a2 100644
--- a/site2/website/versioned_docs/version-2.6.1/cookbooks-deduplication.md
+++ b/site2/website/versioned_docs/version-2.6.1/cookbooks-deduplication.md
@@ -7,13 +7,11 @@ original_id: cookbooks-deduplication
When **Message deduplication** is enabled, it ensures that each message produced on Pulsar topics is persisted to disk *only once*, even if the message is produced more than once. Message deduplication is handled automatically on the server side.
-To use message deduplication in Pulsar, you have to [configure](#configure-message-deduplication) your Pulsar brokers and [clients](#pulsar-clients).
-
-> For more details on message deduplication, refer to [Concepts and Architecture](concepts-messaging.md#message-deduplication).
+To use message deduplication in Pulsar, you need to configure your Pulsar brokers and clients.
## How it works
-You can enable or disable message deduplication on a per-namespace basis. By default, it is *disabled* on all namespaces. You can enable it in the following ways:
+You can enable or disable message deduplication on a per-namespace basis. By default, it is disabled on all namespaces. You can enable it in the following ways:
* Enable for all namespaces at the broker-level
* Enable for specific namespaces with the `pulsar-admin namespaces` interface
@@ -24,7 +22,7 @@ You can configure message deduplication in Pulsar using the [`broker.conf`](refe
Parameter | Description | Default
:---------|:------------|:-------
-`brokerDeduplicationEnabled` | Sets the default behavior for message deduplication in the Pulsar [broker](reference-terminology.md#broker). If it is set to `true`, message deduplication is enabled by default on all namespaces; if it is set to `false` (the default), you have to enable or disable deduplication on a per-namespace basis. | `false`
+`brokerDeduplicationEnabled` | Sets the default behavior for message deduplication in the Pulsar broker. If it is set to `true`, message deduplication is enabled by default on all namespaces; if it is set to `false`, you have to enable or disable deduplication on a per-namespace basis. | `false`
`brokerDeduplicationMaxNumberOfProducers` | The maximum number of producers for which information is stored for deduplication purposes. | `10000`
`brokerDeduplicationEntriesInterval` | The number of entries after which a deduplication informational snapshot is taken. A larger interval leads to fewer snapshots being taken, though this lengthens the topic recovery time (the time required for entries published after the snapshot to be replayed). | `1000`
`brokerDeduplicationProducerInactivityTimeoutMinutes` | The time of inactivity (in minutes) after which the broker discards deduplication information related to a disconnected producer. | `360` (6 hours)
@@ -33,7 +31,7 @@ Parameter | Description | Default
By default, message deduplication is *disabled* on all Pulsar namespaces. To enable it by default on all namespaces, set the `brokerDeduplicationEnabled` parameter to `true` and re-start the broker.
-Even if you set the value for `brokerDeduplicationEnabled`, enabling or disabling via Pulsar admin CLI will override the default settings at the broker-level.
+Even if you set the value for `brokerDeduplicationEnabled`, enabling or disabling via Pulsar admin CLI overrides the default settings at the broker-level.
### Enable message deduplication
diff --git a/site2/website/versioned_docs/version-2.6.1/cookbooks-encryption.md b/site2/website/versioned_docs/version-2.6.1/cookbooks-encryption.md
index 3d4490a..7fbb591 100644
--- a/site2/website/versioned_docs/version-2.6.1/cookbooks-encryption.md
+++ b/site2/website/versioned_docs/version-2.6.1/cookbooks-encryption.md
@@ -1,7 +1,7 @@
---
id: version-2.6.1-cookbooks-encryption
title: Pulsar Encryption
-sidebar_label: Encryption
+sidebar_label: Encryption
original_id: cookbooks-encryption
---
diff --git a/site2/website/versioned_docs/version-2.6.1/cookbooks-tiered-storage.md b/site2/website/versioned_docs/version-2.6.1/cookbooks-tiered-storage.md
index fd16678..67331cd 100644
--- a/site2/website/versioned_docs/version-2.6.1/cookbooks-tiered-storage.md
+++ b/site2/website/versioned_docs/version-2.6.1/cookbooks-tiered-storage.md
@@ -34,8 +34,6 @@ Pulsar uses multi-part objects to upload the segment data. It is possible that a
We recommend you add a life cycle rule your bucket to expire incomplete multi-part upload after a day or two to avoid
getting charged for incomplete uploads.
-When ledgers are offloaded to long term storage, you can still query data in the offloaded ledgers with Pulsar SQL.
-
## Configuring the offload driver
Offloading is configured in ```broker.conf```.
diff --git a/site2/website/versioned_docs/version-2.6.1/deploy-bare-metal.md b/site2/website/versioned_docs/version-2.6.1/deploy-bare-metal.md
index ed293da..f5174f3 100644
--- a/site2/website/versioned_docs/version-2.6.1/deploy-bare-metal.md
+++ b/site2/website/versioned_docs/version-2.6.1/deploy-bare-metal.md
@@ -286,7 +286,7 @@ $ bin/pulsar-daemon start bookie
To start the bookie in the foreground:
```bash
-$ bin/bookkeeper bookie
+$ bin/pulsar bookie
```
You can verify that a bookie works properly by running the `bookiesanity` command on the [BookKeeper shell](reference-cli-tools.md#shell):
diff --git a/site2/website/versioned_docs/version-2.6.1/deploy-dcos.md b/site2/website/versioned_docs/version-2.6.1/deploy-dcos.md
index a1dbd92..7c4e5ee 100644
--- a/site2/website/versioned_docs/version-2.6.1/deploy-dcos.md
+++ b/site2/website/versioned_docs/version-2.6.1/deploy-dcos.md
@@ -1,8 +1,8 @@
---
-id: version-2.6.1-deploy-dcos
+id: version-2.6.1-deploy-docs
title: Deploy Pulsar on DC/OS
sidebar_label: DC/OS
-original_id: deploy-dcos
+original_id: deploy-docs
---
> ### Tips
diff --git a/site2/website/versioned_docs/version-2.6.1/functions-cli.md b/site2/website/versioned_docs/version-2.6.1/functions-cli.md
index b3aa60e..810ef0d 100644
--- a/site2/website/versioned_docs/version-2.6.1/functions-cli.md
+++ b/site2/website/versioned_docs/version-2.6.1/functions-cli.md
@@ -1,7 +1,7 @@
---
id: version-2.6.1-functions-cli
title: Pulsar Functions command line tool
-sidebar_label: Reference: CLI
+sidebar_label: "Reference: CLI"
original_id: functions-cli
---
diff --git a/site2/website/versioned_docs/version-2.6.1/functions-debug.md b/site2/website/versioned_docs/version-2.6.1/functions-debug.md
index 41d9e28..e3aa139 100644
--- a/site2/website/versioned_docs/version-2.6.1/functions-debug.md
+++ b/site2/website/versioned_docs/version-2.6.1/functions-debug.md
@@ -1,7 +1,7 @@
---
id: version-2.6.1-functions-debug
title: Debug Pulsar Functions
-sidebar_label: How-to: Debug
+sidebar_label: "How-to: Debug"
original_id: functions-debug
---
diff --git a/site2/website/versioned_docs/version-2.6.1/functions-deploy.md b/site2/website/versioned_docs/version-2.6.1/functions-deploy.md
index fd091ed..542bde6 100644
--- a/site2/website/versioned_docs/version-2.6.1/functions-deploy.md
+++ b/site2/website/versioned_docs/version-2.6.1/functions-deploy.md
@@ -1,7 +1,7 @@
---
id: version-2.6.1-functions-deploy
title: Deploy Pulsar Functions
-sidebar_label: How-to: Deploy
+sidebar_label: "How-to: Deploy"
original_id: functions-deploy
---
diff --git a/site2/website/versioned_docs/version-2.6.1/functions-develop.md b/site2/website/versioned_docs/version-2.6.1/functions-develop.md
index 3df452f..021e344 100644
--- a/site2/website/versioned_docs/version-2.6.1/functions-develop.md
+++ b/site2/website/versioned_docs/version-2.6.1/functions-develop.md
@@ -1,7 +1,7 @@
---
id: version-2.6.1-functions-develop
title: Develop Pulsar Functions
-sidebar_label: How-to: Develop
+sidebar_label: "How-to: Develop"
original_id: functions-develop
---
diff --git a/site2/website/versioned_docs/version-2.6.1/functions-metrics.md b/site2/website/versioned_docs/version-2.6.1/functions-metrics.md
deleted file mode 100644
index 5415f9c..0000000
--- a/site2/website/versioned_docs/version-2.6.1/functions-metrics.md
+++ /dev/null
@@ -1,7 +0,0 @@
----
-id: version-2.6.1-functions-metrics
-title: Metrics for Pulsar Functions
-sidebar_label: Metrics
-original_id: functions-metrics
----
-
diff --git a/site2/website/versioned_docs/version-2.6.1/functions-runtime.md b/site2/website/versioned_docs/version-2.6.1/functions-runtime.md
index e99976e..9262949 100644
--- a/site2/website/versioned_docs/version-2.6.1/functions-runtime.md
+++ b/site2/website/versioned_docs/version-2.6.1/functions-runtime.md
@@ -1,7 +1,7 @@
---
id: version-2.6.1-functions-runtime
title: Configure Functions runtime
-sidebar_label: Setup: Configure Functions runtime
+sidebar_label: "Setup: Configure Functions runtime"
original_id: functions-runtime
---
@@ -11,6 +11,9 @@ Pulsar Functions support the following methods to run functions.
- *Process*: Invoke functions in processes forked by Functions Worker.
- *Kubernetes*: Submit functions as Kubernetes StatefulSets by Functions Worker.
+#### Note
+> Pulsar supports adding labels to the Kubernetes StatefulSets and services while launching functions, which facilitates selecting the target Kubernetes objects.
+
The differences of the thread and process modes are:
- Thread mode: when a function runs in thread mode, it runs on the same Java virtual machine (JVM) with Functions worker.
- Process mode: when a function runs in process mode, it runs on the same machine that Functions worker runs.
diff --git a/site2/website/versioned_docs/version-2.6.1/functions-worker.md b/site2/website/versioned_docs/version-2.6.1/functions-worker.md
index 6e839c7..90bdf4e 100644
--- a/site2/website/versioned_docs/version-2.6.1/functions-worker.md
+++ b/site2/website/versioned_docs/version-2.6.1/functions-worker.md
@@ -1,12 +1,12 @@
---
id: version-2.6.1-functions-worker
title: Deploy and manage functions worker
-sidebar_label: Setup: Pulsar Functions Worker
+sidebar_label: "Setup: Pulsar Functions Worker"
original_id: functions-worker
---
Before using Pulsar Functions, you need to learn how to set up Pulsar Functions worker and how to [configure Functions runtime](functions-runtime.md).
-Pulsar `functions-worker` is a logic component to run Pulsar Functions in cluster mode. Two options are available, and you can select either of the two options based on your requirements.
+Pulsar `functions-worker` is a logic component to run Pulsar Functions in cluster mode. Two options are available, and you can select either based on your requirements.
- [run with brokers](#run-functions-worker-with-brokers)
- [run it separately](#run-functions-worker-separately) in a different broker
@@ -25,16 +25,16 @@ To enable functions-worker running as part of a broker, you need to set `functio
functionsWorkerEnabled=true
```
-When you set `functionsWorkerEnabled` to `true`, it means that you start functions-worker as part of a broker. You need to configure the `conf/functions_worker.yml` file to customize your functions_worker.
+If the `functionsWorkerEnabled` is set to `true`, the functions-worker is started as part of a broker. You need to configure the `conf/functions_worker.yml` file to customize your functions_worker.
Before you run Functions-worker with broker, you have to configure Functions-worker, and then start it with brokers.
### Configure Functions-Worker to run with brokers
-In this mode, since `functions-worker` is running as part of broker, most of the settings already inherit from your broker configuration (for example, configurationStore settings, authentication settings, and so on).
+In this mode, most of the settings are already inherited from your broker configuration (for example, configurationStore settings, authentication settings, and so on) since `functions-worker` is running as part of the broker.
Pay attention to the following required settings when configuring functions-worker in this mode.
-- `numFunctionPackageReplicas`: The number of replicas to store function packages. The default value is `1`, which is good for standalone deployment. For production deployment, to ensure high availability, set it to be more than `2` .
+- `numFunctionPackageReplicas`: The number of replicas to store function packages. The default value is `1`, which is good for standalone deployment. For production deployment, to ensure high availability, set it to be larger than `2`.
- `pulsarFunctionsCluster`: Set the value to your Pulsar cluster name (same as the `clusterName` setting in the broker configuration).
If authentication is enabled on the BookKeeper cluster, configure the following BookKeeper authentication settings.
@@ -53,7 +53,7 @@ And then you can use the following command to verify if `functions-worker` is ru
curl <broker-ip>:8080/admin/v2/worker/cluster
```
-After entering the command above, a list of active function workers in the cluster is returned. The output is something similar as follows.
+After entering the command above, a list of active function workers in the cluster is returned. The output is similar to the following.
```json
[{"workerId":"<worker-id>","workerHostname":"<worker-hostname>","port":8080}]
@@ -66,7 +66,7 @@ This section illustrates how to run `functions-worker` as a separate process in
> Note
-In this mode, make sure `functionsWorkerEnabled` is set to `false`, so you won't start `functions-worker` with brokers by mistake.
+> In this mode, make sure `functionsWorkerEnabled` is set to `false`, so you won't start `functions-worker` with brokers by mistake.
### Configure Functions-worker to run separately
@@ -74,7 +74,7 @@ To run function-worker separately, you have to configure the following parameter
#### Worker parameters
-- `workerId`: The type is string. It is unique across clusters, used to identify a worker machine.
+- `workerId`: The type is string. It is unique across clusters, which is used to identify a worker machine.
- `workerHostname`: The hostname of the worker machine.
- `workerPort`: The port that the worker server listens on. Keep it as default if you don't customize it.
- `workerPortTls`: The TLS port that the worker server listens on. Keep it as default if you don't customize it.
@@ -106,25 +106,44 @@ If you want to enable security on functions workers, you *should*:
To enable TLS transport encryption, configure the following settings.
```
+useTLS: true
+pulsarServiceUrl: pulsar+ssl://localhost:6651/
+pulsarWebServiceUrl: https://localhost:8443
+
tlsEnabled: true
tlsCertificateFilePath: /path/to/functions-worker.cert.pem
tlsKeyFilePath: /path/to/functions-worker.key-pk8.pem
tlsTrustCertsFilePath: /path/to/ca.cert.pem
+
+// The path to trusted certificates used by the Pulsar client to authenticate with Pulsar brokers
+brokerClientTrustCertsFilePath: /path/to/ca.cert.pem
```
For details on TLS encryption, refer to [Transport Encryption using TLS](security-tls-transport.md).
##### Enable Authentication Provider
-To enable authentication on Functions Worker, configure the following settings.
+To enable authentication on Functions Worker, you need to configure the following settings.
+
> Note
-Substitute the *providers list* with the providers you want to enable.
+> Substitute the *providers list* with the providers you want to enable.
```
authenticationEnabled: true
authenticationProviders: [ provider1, provider2 ]
```
+For *TLS Authentication* provider, follow the example below to add the necessary settings.
+See [TLS Authentication](security-tls-authentication.md) for more details.
+
+```
+brokerClientAuthenticationPlugin: org.apache.pulsar.client.impl.auth.AuthenticationTls
+brokerClientAuthenticationParameters: tlsCertFile:/path/to/admin.cert.pem,tlsKeyFile:/path/to/admin.key-pk8.pem
+
+authenticationEnabled: true
+authenticationProviders: ['org.apache.pulsar.broker.authentication.AuthenticationProviderTls']
+```
+
For *SASL Authentication* provider, add `saslJaasClientAllowedIds` and `saslJaasBrokerSectionName`
under `properties` if needed.
@@ -134,7 +153,7 @@ properties:
saslJaasBrokerSectionName: Broker
```
-For *Token Authentication* prodivder, add necessary settings under `properties` if needed.
+For *Token Authentication* provider, add necessary settings for `properties` if needed.
See [Token Authentication](security-jwt.md) for more details.
```
properties:
@@ -145,10 +164,11 @@ properties:
##### Enable Authorization Provider
-To enable authorization on Functions Worker, you need to configure `authorizationEnabled` and `configurationStoreServers`. The authentication provider connects to `configurationStoreServers` to receive namespace policies.
+To enable authorization on Functions Worker, you need to configure `authorizationEnabled`, `authorizationProvider` and `configurationStoreServers`. The authentication provider connects to `configurationStoreServers` to receive namespace policies.
```yaml
authorizationEnabled: true
+authorizationProvider: org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
configurationStoreServers: <configuration-store-servers>
```
@@ -163,7 +183,7 @@ superUserRoles:
#### BookKeeper Authentication
-If authentication is enabled on the BookKeeper cluster, you should configure the BookKeeper authentication settings as follows:
+If authentication is enabled on the BookKeeper cluster, you need configure the BookKeeper authentication settings as follows:
- `bookkeeperClientAuthenticationPlugin`: the plugin name of BookKeeper client authentication.
- `bookkeeperClientAuthenticationParametersName`: the plugin parameters name of BookKeeper client authentication.
diff --git a/site2/website/versioned_docs/version-2.6.1/getting-started-concepts-and-architecture.md b/site2/website/versioned_docs/version-2.6.1/getting-started-concepts-and-architecture.md
deleted file mode 100644
index 051cc64..0000000
--- a/site2/website/versioned_docs/version-2.6.1/getting-started-concepts-and-architecture.md
+++ /dev/null
@@ -1,16 +0,0 @@
----
-id: version-2.6.1-concepts-architecture
-title: Pulsar concepts and architecture
-sidebar_label: Concepts and architecture
-original_id: concepts-architecture
----
-
-
-
-
-
-
-
-
-
-
diff --git a/site2/website/versioned_docs/version-2.6.1/getting-started-helm.md b/site2/website/versioned_docs/version-2.6.1/getting-started-helm.md
index ba988ce..dff5bdf 100644
--- a/site2/website/versioned_docs/version-2.6.1/getting-started-helm.md
+++ b/site2/website/versioned_docs/version-2.6.1/getting-started-helm.md
@@ -130,7 +130,7 @@ We use [Minikube](https://kubernetes.io/docs/getting-started-guides/minikube/) i
1. Enter the `toolset` container.
```bash
- kubectl exec -it -n pulsar pulsar-mini-toolset-0 -- /bin/bash
+ kubectl exec -it -n pulsar pulsar-mini-toolset-0 /bin/bash
```
2. In the `toolset` container, create a tenant named `apache`.
@@ -193,7 +193,7 @@ We use [Minikube](https://kubernetes.io/docs/getting-started-guides/minikube/) i
You can use the Pulsar client to create producers and consumers to produce and consume messages.
-By default, the Pulsar Helm chart exposes the Pulsar cluster through a Kubernetes `LoadBalancer`. In Minikube, you can use the following command to get the IP address of the proxy service.
+By default, the Pulsar Helm chart exposes the Pulsar cluster through a Kubernetes `LoadBalancer`. In Minikube, you can use the following command to check the proxy service.
```bash
kubectl get services -n pulsar | grep pulsar-mini-proxy
@@ -205,19 +205,36 @@ You will see a similar output as below.
pulsar-mini-proxy LoadBalancer 10.97.240.109 <pending> 80:32305/TCP,6650:31816/TCP 28m
```
-This output tells what are the node ports that Pulsar cluster's binary port and HTTP port are exposed to. The port after `80:` is the HTTP port while the port after `6650:` is the binary port.
+This output tells what are the node ports that Pulsar cluster's binary port and HTTP port are mapped to. The port after `80:` is the HTTP port while the port after `6650:` is the binary port.
-Then you can find the IP address of your Minikube server by running the following command.
+Then you can find the IP address and exposed ports of your Minikube server by running the following command.
```bash
-minikube ip
+minikube service pulsar-mini-proxy -n pulsar
```
-At this point, you can get the service URLs to connect to your Pulsar client.
+**Output**
+```bash
+|-----------|-------------------|-------------|-------------------------|
+| NAMESPACE | NAME | TARGET PORT | URL |
+|-----------|-------------------|-------------|-------------------------|
+| pulsar | pulsar-mini-proxy | http/80 | http://172.17.0.4:32305 |
+| | | pulsar/6650 | http://172.17.0.4:31816 |
+|-----------|-------------------|-------------|-------------------------|
+🏃 Starting tunnel for service pulsar-mini-proxy.
+|-----------|-------------------|-------------|------------------------|
+| NAMESPACE | NAME | TARGET PORT | URL |
+|-----------|-------------------|-------------|------------------------|
+| pulsar | pulsar-mini-proxy | | http://127.0.0.1:61853 |
+| | | | http://127.0.0.1:61854 |
+|-----------|-------------------|-------------|------------------------|
+```
+
+At this point, you can get the service URLs to connect to your Pulsar client. Here are URL examples:
```
-webServiceUrl=http://$(minikube ip):<exposed-http-port>/
-brokerServiceUrl=pulsar://$(minikube ip):<exposed-binary-port>/
+webServiceUrl=http://127.0.0.1:61853/
+brokerServiceUrl=pulsar://127.0.0.1:61854/
```
Then you can proceed with the following steps:
diff --git a/site2/website/versioned_docs/version-2.6.1/helm-install.md b/site2/website/versioned_docs/version-2.6.1/helm-install.md
index 04eedd9..9784b91 100644
--- a/site2/website/versioned_docs/version-2.6.1/helm-install.md
+++ b/site2/website/versioned_docs/version-2.6.1/helm-install.md
@@ -1,7 +1,7 @@
---
id: version-2.6.1-helm-install
title: Install Apache Pulsar using Helm
-sidebar_label: Install
+sidebar_label: Install
original_id: helm-install
---
diff --git a/site2/website/versioned_docs/version-2.6.1/io-cdc-canal.md b/site2/website/versioned_docs/version-2.6.1/io-cdc-canal.md
new file mode 100644
index 0000000..a91f591
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.6.1/io-cdc-canal.md
@@ -0,0 +1,6 @@
+---
+id: version-2.6.1-io-cdc-canal
+title: CDC Canal Connector
+sidebar_label: CDC Canal Connector
+original_id: io-cdc-canal
+---
diff --git a/site2/website/versioned_docs/version-2.6.1/io-cli.md b/site2/website/versioned_docs/version-2.6.1/io-cli.md
index e20e0a5..b3e7a44 100644
--- a/site2/website/versioned_docs/version-2.6.1/io-cli.md
+++ b/site2/website/versioned_docs/version-2.6.1/io-cli.md
@@ -70,7 +70,7 @@ $ pulsar-admin sources create options
| `-st`, `--schema-type` | The schema type.<br> Either a builtin schema (for example, AVRO and JSON) or custom schema class name to be used to encode messages emitted from source.
| `--source-config` | Source config key/values.
| `--source-config-file` | The path to a YAML config file specifying the source's configuration.
-| `-t`, `--source-type` | The source's connector provider. The `source-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+| `-t`, `--source-type` | The source's connector provider.
| `--tenant` | The source's tenant.
### `update`
@@ -101,7 +101,7 @@ $ pulsar-admin sources update options
| `-st`, `--schema-type` | The schema type.<br> Either a builtin schema (for example, AVRO and JSON) or custom schema class name to be used to encode messages emitted from source.
| `--source-config` | Source config key/values.
| `--source-config-file` | The path to a YAML config file specifying the source's configuration.
-| `-t`, `--source-type` | The source's connector provider. The `source-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+| `-t`, `--source-type` | The source's connector provider.
| `--tenant` | The source's tenant.
| `--update-auth-data` | Whether or not to update the auth data.<br>**Default value: false.**
@@ -269,7 +269,7 @@ $ pulsar-admin sources localrun options
| `-st`, `--schema-type` | The schema type.<br> Either a builtin schema (for example, AVRO and JSON) or custom schema class name to be used to encode messages emitted from source.
|`--source-config`|Source config key/values.
|`--source-config-file`|The path to a YAML config file specifying the source’s configuration.
-|`--source-type`|The source's connector provider. The `source-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+|`--source-type`|The source's connector provider.
|`--tenant`|The source’s tenant.
|`--tls-allow-insecure`|Allow insecure tls connection.<br>**Default value: false**.
|`--tls-trust-cert-path`|The tls trust cert file path.
@@ -360,7 +360,7 @@ $ pulsar-admin sinks create options
| `--retain-ordering` | Sink consumes and sinks messages in order.
| `--sink-config` | sink config key/values.
| `--sink-config-file` | The path to a YAML config file specifying the sink's configuration.
-| `-t`, `--sink-type` | The sink's connector provider. The `sink-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+| `-t`, `--sink-type` | The sink's connector provider.
| `--subs-name` | Pulsar source subscription name if user wants a specific subscription-name for input-topic consumer.
| `--tenant` | The sink's tenant.
| `--timeout-ms` | The message timeout in milliseconds.
@@ -396,7 +396,7 @@ $ pulsar-admin sinks update options
| `--retain-ordering` | Sink consumes and sinks messages in order.
| `--sink-config` | sink config key/values.
| `--sink-config-file` | The path to a YAML config file specifying the sink's configuration.
-| `-t`, `--sink-type` | The sink's connector provider. The `sink-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+| `-t`, `--sink-type` | The sink's connector provider.
| `--subs-name` | Pulsar source subscription name if user wants a specific subscription-name for input-topic consumer.
| `--tenant` | The sink's tenant.
| `--timeout-ms` | The message timeout in milliseconds.
@@ -570,7 +570,7 @@ $ pulsar-admin sinks localrun options
|`--retain-ordering` | Sink consumes and sinks messages in order.
|`--sink-config`|sink config key/values.
|`--sink-config-file`|The path to a YAML config file specifying the sink’s configuration.
-|`--sink-type`|The sink's connector provider. The `sink-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+|`--sink-type`|The sink's connector provider.
|`--subs-name` | Pulsar source subscription name if user wants a specific subscription-name for input-topic consumer.
|`--tenant`|The sink’s tenant.
| `--timeout-ms` | The message timeout in milliseconds.
diff --git a/site2/website/versioned_docs/version-2.6.1/io-hdfs2-sink.md b/site2/website/versioned_docs/version-2.6.1/io-hdfs2-sink.md
index d953944..68c30ec 100644
--- a/site2/website/versioned_docs/version-2.6.1/io-hdfs2-sink.md
+++ b/site2/website/versioned_docs/version-2.6.1/io-hdfs2-sink.md
@@ -22,8 +22,8 @@ The configuration of the HDFS2 sink connector has the following properties.
| `compression` | Compression |false |None |The compression code used to compress or de-compress the files on HDFS. <br/><br/>Below are the available options:<br/><li>BZIP2<br/><li>DEFLATE<br/><li>GZIP<br/><li>LZ4<br/><li>SNAPPY|
| `kerberosUserPrincipal` |String| false| None|The principal account of Kerberos user used for authentication. |
| `keytab` | String|false|None| The full pathname of the Kerberos keytab file used for authentication. |
-| `filenamePrefix` |String| false |None |The prefix of the files created inside the HDFS directory.<br/><br/>**Example**<br/> The value of topicA result in files named topicA-. |
-| `fileExtension` | String| false | None| The extension added to the files written to HDFS.<br/><br/>**Example**<br/>'.txt'<br/> '.seq' |
+| `filenamePrefix` |String| true, if `compression` is set to `None`. | None |The prefix of the files created inside the HDFS directory.<br/><br/>**Example**<br/> The value of topicA result in files named topicA-. |
+| `fileExtension` | String| true | None | The extension added to the files written to HDFS.<br/><br/>**Example**<br/>'.txt'<br/> '.seq' |
| `separator` | char|false |None |The character used to separate records in a text file. <br/><br/>If no value is provided, the contents from all records are concatenated together in one continuous byte array. |
| `syncInterval` | long| false |0| The interval between calls to flush data to HDFS disk in milliseconds. |
| `maxPendingRecords` |int| false|Integer.MAX_VALUE | The maximum number of records that hold in memory before acking. <br/><br/>Setting this property to 1 makes every record send to disk before the record is acked.<br/><br/>Setting this property to a higher value allows buffering records before flushing them to disk.
@@ -39,6 +39,7 @@ Before using the HDFS2 sink connector, you need to create a configuration file t
"hdfsConfigResources": "core-site.xml",
"directory": "/foo/bar",
"filenamePrefix": "prefix",
+ "fileExtension": ".log",
"compression": "SNAPPY"
}
```
@@ -50,5 +51,6 @@ Before using the HDFS2 sink connector, you need to create a configuration file t
hdfsConfigResources: "core-site.xml"
directory: "/foo/bar"
filenamePrefix: "prefix"
+ fileExtension: ".log"
compression: "SNAPPY"
```
diff --git a/site2/website/versioned_docs/version-2.6.1/io-quickstart.md b/site2/website/versioned_docs/version-2.6.1/io-quickstart.md
index 83801dd..030b248 100644
--- a/site2/website/versioned_docs/version-2.6.1/io-quickstart.md
+++ b/site2/website/versioned_docs/version-2.6.1/io-quickstart.md
@@ -203,10 +203,6 @@ to create a sink connector and perform other operations on them.
Run the following command to create a Cassandra sink connector with sink type _cassandra_ and the config file _examples/cassandra-sink.yml_ created previously.
-#### Note
-
-> The `sink-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
-
```bash
bin/pulsar-admin sinks create \
--tenant public \
diff --git a/site2/website/versioned_docs/version-2.6.1/io-twitter.md b/site2/website/versioned_docs/version-2.6.1/io-twitter.md
deleted file mode 100644
index 02b11a2..0000000
--- a/site2/website/versioned_docs/version-2.6.1/io-twitter.md
+++ /dev/null
@@ -1,7 +0,0 @@
----
-id: version-2.6.1-io-twitter
-title: Twitter Firehose Connector
-sidebar_label: Twitter Firehose Connector
-original_id: io-twitter
----
-
diff --git a/site2/website/versioned_docs/version-2.6.1/reference-cli-tools.md b/site2/website/versioned_docs/version-2.6.1/reference-cli-tools.md
index 46b9dab..1c514ff 100644
--- a/site2/website/versioned_docs/version-2.6.1/reference-cli-tools.md
+++ b/site2/website/versioned_docs/version-2.6.1/reference-cli-tools.md
@@ -292,6 +292,7 @@ Options
|---|---|---|
|`--auth-params`|Authentication parameters, whose format is determined by the implementation of method `configure` in authentication plugin class, for example "key1:val1,key2:val2" or "{\"key1\":\"val1\",\"key2\":\"val2\"}"|{"saslJaasClientSectionName":"PulsarClient", "serverType":"broker"}|
|`--auth-plugin`|Authentication plugin class name|org.apache.pulsar.client.impl.auth.AuthenticationSasl|
+|`--listener-name`|Listener name for the broker||
|`--url`|Broker URL to which to connect|pulsar://localhost:6650/ </br> ws://localhost:8080 |
@@ -416,6 +417,7 @@ Options
|---|---|---|
|`--auth_params`|Authentication parameters, whose format is determined by the implementation of method `configure` in authentication plugin class, for example "key1:val1,key2:val2" or "{"key1":"val1","key2":"val2"}.||
|`--auth_plugin`|Authentication plugin class name||
+|`--listener-name`|Listener name for the broker||
|`--acks-delay-millis`|Acknowlegments grouping delay in millis|100|
|`-k`, `--encryption-key-name`|The private key name to decrypt payload||
|`-v`, `--encryption-key-value-file`|The file which contains the private key to decrypt payload||
@@ -432,6 +434,7 @@ Options
|`-st`, `--subscription-type`|Subscriber type. Possible values are Exclusive, Shared, Failover, Key_Shared.|Exclusive|
|`-sp`, `--subscription-position`|Subscriber position. Possible values are Latest, Earliest.|Latest|
|`--trust-cert-file`|Path for the trusted TLS certificate file||
+|`--tls-allow-insecure`|Allow insecure TLS connection||
### `produce`
@@ -448,6 +451,7 @@ Options
|---|---|---|
|`--auth_params`|Authentication parameters, whose format is determined by the implementation of method `configure` in authentication plugin class, for example "key1:val1,key2:val2" or "{"key1":"val1","key2":"val2"}.||
|`--auth_plugin`|Authentication plugin class name||
+|`--listener-name`|Listener name for the broker||
|`-b`, `--batch-time-window`|Batch messages in a window of the specified number of milliseconds|1|
|`-z`, `--compression`|Compress messages’ payload. Possible values are NONE, LZ4, ZLIB, ZSTD or SNAPPY.||
|`--conf-file`|Configuration file||
@@ -469,6 +473,7 @@ Options
|`-time`, `--test-duration`|Test duration in secs. If set to 0, it will keep publishing.|0|
|`--trust-cert-file`|Path for the trusted TLS certificate file||
|`--warmup-time`|Warm-up time in seconds|1|
+|`--tls-allow-insecure`|Allow insecure TLS connection||
### `read`
@@ -484,6 +489,7 @@ Options
|---|---|---|
|`--auth_params`|Authentication parameters, whose format is determined by the implementation of method `configure` in authentication plugin class, for example "key1:val1,key2:val2" or "{"key1":"val1","key2":"val2"}.||
|`--auth_plugin`|Authentication plugin class name||
+|`--listener-name`|Listener name for the broker||
|`--conf-file`|Configuration file||
|`-h`, `--help`|Help message|false|
|`-c`, `--max-connections`|Max number of TCP connections to a single broker|100|
@@ -495,7 +501,7 @@ Options
|`-i`, `--stats-interval-seconds`|Statistics interval seconds. If 0, statistics will be disabled.|0|
|`--trust-cert-file`|Path for the trusted TLS certificate file||
|`--use-tls`|Use TLS encryption on the connection|false|
-
+|`--tls-allow-insecure`|Allow insecure TLS connection||
### `websocket-producer`
Run a websocket producer
diff --git a/site2/website/versioned_docs/version-2.6.1/reference-configuration.md b/site2/website/versioned_docs/version-2.6.1/reference-configuration.md
index 22caa28..fd540a9 100644
--- a/site2/website/versioned_docs/version-2.6.1/reference-configuration.md
+++ b/site2/website/versioned_docs/version-2.6.1/reference-configuration.md
@@ -106,8 +106,6 @@ Pulsar brokers are responsible for handling incoming messages from producers, di
|Name|Description|Default|
|---|---|---|
-|advertisedListeners|Specify multiple advertised listeners for the broker.<br><br>The format is `<listener_name>:pulsar://<host>:<port>`.<br><br>If there are multiple listeners, separate them with commas.<br><br>**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.<br><br>**Note**: the listener name must be contained in `advertisedListeners`.<br><br> If the value of this configuration is empty, the broker uses the first listener as the internal listener.|/|
|enablePersistentTopics| Whether persistent topics are enabled on the broker |true|
|enableNonPersistentTopics| Whether non-persistent topics are enabled on the broker |true|
|functionsWorkerEnabled| Whether the Pulsar Functions worker service is enabled in the broker |false|
@@ -269,7 +267,6 @@ subscriptionExpirationTimeMinutes | How long to delete inactive subscriptions fr
|s3ManagedLedgerOffloadRoleSessionName| For Amazon S3 ledger offload, provide a role session name when using a role |pulsar-s3-offload|
| acknowledgmentAtBatchIndexLevelEnabled | Enable or disable the batch index acknowledgement. | false |
| maxMessageSize | Set the maximum size of a message. | 5 MB |
-| preciseTopicPublishRateLimiterEnable | Enable precise topic publish rate limiting. | false |
@@ -442,6 +439,7 @@ The [`pulsar-client`](reference-cli-tools.md#pulsar-client) CLI tool can be used
|replicationMetricsEnabled| |true|
|replicationConnectionsPerBroker| |16|
|replicationProducerQueueSize| |1000|
+| replicationPolicyCheckDurationSeconds | Duration to check replication policy to avoid replicator inconsistency due to missing ZooKeeper watch. When the value is set to 0, disable checking replication policy. | 600 |
|defaultRetentionTimeInMinutes| |0|
|defaultRetentionSizeInMB| |0|
|keepAliveIntervalSeconds| |30|
@@ -518,7 +516,6 @@ The [Pulsar proxy](concepts-architecture-overview.md#pulsar-proxy) can be config
|tokenPublicKey| Configure the public key to be used to validate auth tokens. The key can be specified like: `tokenPublicKey=data:;base64,xxxxxxxxx` or `tokenPublicKey=file:///my/secret.key`||
|tokenPublicAlg| 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) |RS256|
|tokenAuthClaim| Specify the token claim that will be used as the authentication "principal" or "role". The "subject" field will be used if this is left blank ||
-| proxyLogLevel | Set the Pulsar Proxy log level. <li> If the value is set to 0, no TCP channel information is logged. <li> If the value is set to 1, only the TCP channel information and command information (without message body) are parsed and logged. <li> If the value is set to 2, all TCP channel information, command information, and message body are parsed and logged. | 0 |
## ZooKeeper
diff --git a/site2/website/versioned_docs/version-2.6.1/reference-pulsar-admin.md b/site2/website/versioned_docs/version-2.6.1/reference-pulsar-admin.md
index ee16efa..536db4c 100644
--- a/site2/website/versioned_docs/version-2.6.1/reference-pulsar-admin.md
+++ b/site2/website/versioned_docs/version-2.6.1/reference-pulsar-admin.md
@@ -451,7 +451,7 @@ Options
|`--schema-type`|The builtin schema type or custom schema class name to be used for messages output by the function||
|`--sliding-interval-count`|The number of messages after which the window slides||
|`--sliding-interval-duration-ms`|The time duration after which the window slides||
-|`--state-storage-service-url`|The URL for the state storage service (by default Apache BookKeeper)||
+|`--state-storage-service-url`|The URL for the state storage service. By default, it it set to the service URL of the Apache BookKeeper. This service URL must be added manually when the Pulsar Function runs locally. ||
|`--tenant`|The function’s tenant||
|`--topics-pattern`|The topic pattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe class name for a pattern in --custom-serde-inputs (supported for java fun only)||
|`--user-config`|User-defined config key/values||
diff --git a/site2/website/versioned_docs/version-2.6.1/schema-get-started.md b/site2/website/versioned_docs/version-2.6.1/schema-get-started.md
index 15ead9e..86cab40 100644
--- a/site2/website/versioned_docs/version-2.6.1/schema-get-started.md
+++ b/site2/website/versioned_docs/version-2.6.1/schema-get-started.md
@@ -15,10 +15,6 @@ Producers and consumers need some kind of mechanism for coordinating types at th
Applications typically adopt one of the following approaches to guarantee type safety in messaging. Both approaches are available in Pulsar, and you're free to adopt one or the other or to mix and match on a per-topic basis.
-#### Note
->
-> Currently, the Pulsar schema registry is only available for the [Java client](client-libraries-java.md), [CGo client](client-libraries-cgo.md), [Python client](client-libraries-python.md), and [C++ client](client-libraries-cpp.md).
-
### Client-side approach
Producers and consumers are responsible for not only serializing and deserializing messages (which consist of raw bytes) but also "knowing" which types are being transmitted via which topics.
diff --git a/site2/website/versioned_docs/version-2.6.1/security-bouncy-castle.md b/site2/website/versioned_docs/version-2.6.1/security-bouncy-castle.md
index e017705..15cdb6b 100644
--- a/site2/website/versioned_docs/version-2.6.1/security-bouncy-castle.md
+++ b/site2/website/versioned_docs/version-2.6.1/security-bouncy-castle.md
@@ -38,7 +38,8 @@ Pulsar module `bouncy-castle-bc`, which defined by `bouncy-castle/bc/pom.xml` co
</dependency>
```
-By using this `bouncy-castle-bc` module, user can easily include and exclude BouncyCastle non-FIPS jars.
+And based on Pulsar module `bouncy-castle-bc`, Pulsar shades a fat jar in module `bouncy-castle-bc-shaded` that contains needed classes of BouncyCastle non-FIPS jars.
+By using this `bouncy-castle-bc-shaded` module, user can easily include and exclude BouncyCastle non-FIPS jars.
### Pulsar Client and Broker dependencies on BC-non-FIPS
@@ -47,7 +48,7 @@ Pulsar Client(`pulsar-client-original`) module include BouncyCastle non-FIPS jar
```xml
<dependency>
<groupId>org.apache.pulsar</groupId>
- <artifactId>bouncy-castle-bc</artifactId>
+ <artifactId>bouncy-castle-bc-shaded</artifactId>
<version>${project.parent.version}</version>
</dependency>
```
@@ -95,7 +96,7 @@ For example:
<exclusions>
<exclusion>
<groupId>${project.groupId}</groupId>
- <artifactId>bouncy-castle-bc</artifactId>
+ <artifactId>bouncy-castle-bc-shaded</artifactId>
</exclusion>
</exclusions>
</dependency>
diff --git a/site2/website/versioned_docs/version-2.6.1/security-oauth2.md b/site2/website/versioned_docs/version-2.6.1/security-oauth2.md
new file mode 100644
index 0000000..9f5d298
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.6.1/security-oauth2.md
@@ -0,0 +1,135 @@
+---
+id: version-2.6.1-security-oauth2
+title: Client authentication using OAuth 2.0 access tokens
+sidebar_label: Authentication using OAuth 2.0 access tokens
+original_id: security-oauth2
+---
+
+Pulsar supports authenticating clients using OAuth 2.0 access tokens. You can use OAuth 2.0 access tokens to identify a Pulsar client and associate the Pulsar client with some "principal" (or "role"), which is permitted to do some actions, such as publishing messages to a topic or consume messages from a topic.
+
+This module is used to support the Pulsar client authentication plugin for OAuth 2.0. After communicating with the Oauth 2.0 server, the Pulsar client gets an `access token` from the Oauth 2.0 server, and passes this `access token` to the Pulsar broker to do the authentication. The broker can use the `org.apache.pulsar.broker.authentication.AuthenticationProviderToken`. Or, you can add your own `AuthenticationProvider` to make it with this module.
+
+## Authentication provider configuration
+
+This library allows you to authenticate the Pulsar client by using an access token that is obtained from an OAuth 2.0 authorization service, which acts as a _token issuer_.
+
+### Authentication types
+
+The authentication type determines how to obtain an access token through an OAuth 2.0 authorization flow.
+
+#### Note
+> Currently, the Pulsar Java client only supports the `client_credentials` authentication type .
+
+#### Client credentials
+
+The following table lists parameters supported for the `client credentials` authentication type.
+
+| Parameter | Description | Example | Required or not |
+| --- | --- | --- | --- |
+| `type` | Oauth 2.0 authentication type. | `client_credentials` (default) | Optional |
+| `issuerUrl` | URL of the authentication provider which allows the Pulsar client to obtain an access token | `https://accounts.google.com` | Required |
+| `privateKey` | URL to a JSON credentials file | Support the following pattern formats: <br> <li> `file:///path/to/file` <li>`file:/path/to/file` <li> `data:application/json;base64,<base64-encoded value>` | Required |
+| `audience` | An OAuth 2.0 "resource server" identifier for the Pulsar cluster | `https://broker.example.com` | Required |
+
+The credentials file contains service account credentials used with the client authentication type. The following shows an example of a credentials file `credentials_file.json`.
+
+```json
+{
+ "type": "client_credentials",
+ "client_id": "d9ZyX97q1ef8Cr81WHVC4hFQ64vSlDK3",
+ "client_secret": "on1uJ...k6F6R",
+ "client_email": "1234567890-abcdefghijklmnopqrstuvwxyz@developer.gserviceaccount.com",
+ "issuer_url": "https://accounts.google.com"
+}
+```
+
+In the above example, the authentication type is set to `client_credentials` by default. And the fields "client_id" and "client_secret" are required.
+
+### Typical original Oauth2 request mapping
+
+The following shows a typical original Oauth2 request, which is used to obtain the access token from the Oauth2 server.
+
+```bash
+curl --request POST \
+ --url https://dev-kt-aa9ne.us.auth0.com/oauth/token \
+ --header 'content-type: application/json' \
+ --data '{
+ "client_id":"Xd23RHsUnvUlP7wchjNYOaIfazgeHd9x",
+ "client_secret":"rT7ps7WY8uhdVuBTKWZkttwLdQotmdEliaM5rLfmgNibvqziZ-g07ZH52N_poGAb",
+ "audience":"https://dev-kt-aa9ne.us.auth0.com/api/v2/",
+ "grant_type":"client_credentials"}'
+```
+
+In the above example, the mapping relationship is shown as below.
+
+- The `issuerUrl` parameter in this plugin is mapped to `--url https://dev-kt-aa9ne.us.auth0.com/oauth/token`.
+- The `privateKey` file parameter in this plugin should at least contains the `client_id` and `client_secret` fields.
+- The `audience` parameter in this plugin is mapped to `"audience":"https://dev-kt-aa9ne.us.auth0.com/api/v2/"`.
+
+## Client Configuration
+
+You can use the Oauth2 authentication provider with the following Pulsar clients.
+
+### Java
+
+You can use the factory method to configure authentication for Pulsar Java client.
+
+```java
+String issuerUrl = "https://dev-kt-aa9ne.us.auth0.com/oauth/token";
+String credentialsUrl = "file:///path/to/KeyFile.json";
+String audience = "https://dev-kt-aa9ne.us.auth0.com/api/v2/";
+
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://broker.example.com:6650/")
+ .authentication(
+ AuthenticationFactoryOAuth2.clientCredentials(issuerUrl, credentialsUrl, audience))
+ .build();
+```
+
+In addition, you can also use the encoded parameters to configure authentication for Pulsar Java client.
+
+```java
+Authentication auth = AuthenticationFactory
+ .create(AuthenticationOAuth2.class.getName(), "{"type":"client_credentials","privateKey":"./key/path/..","issuerUrl":"...","audience":"..."}");
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://broker.example.com:6650/")
+ .authentication(auth)
+ .build();
+```
+
+### C++ client
+
+The C++ client is similar to the Java client. You need to provide parameters of `issuerUrl`, `private_key` (the credentials file path), and the audience.
+
+```c++
+#include <pulsar/Client.h>
+
+pulsar::ClientConfiguration config;
+std::string params = R"({
+ "issuer_url": "https://dev-kt-aa9ne.us.auth0.com/oauth/token",
+ "private_key": "../../pulsar-broker/src/test/resources/authentication/token/cpp_credentials_file.json",
+ "audience": "https://dev-kt-aa9ne.us.auth0.com/api/v2/"})";
+
+config.setAuth(pulsar::AuthOauth2::create(params));
+
+pulsar::Client client("pulsar://broker.example.com:6650/", config);
+```
+
+### Go client
+
+To enable OAuth2 authentication in Go client, you need to configure OAuth2 authentication.
+This example shows how to configure OAuth2 authentication in Go client.
+
+```go
+oauth := pulsar.NewAuthenticationOAuth2(map[string]string{
+ "type": "client_credentials",
+ "issuerUrl": "https://dev-kt-aa9ne.us.auth0.com/oauth/token",
+ "audience": "https://dev-kt-aa9ne.us.auth0.com/api/v2/",
+ "privateKey": "/path/to/privateKey",
+ "clientId": "0Xx...Yyxeny",
+ })
+client, err := pulsar.NewClient(pulsar.ClientOptions{
+ URL: "puslar://my-cluster:6650",
+ Authentication: oauth,
+})
+```
diff --git a/site2/website/versioned_docs/version-2.6.1/security-overview.md b/site2/website/versioned_docs/version-2.6.1/security-overview.md
index c9b70d3..86ef398 100644
--- a/site2/website/versioned_docs/version-2.6.1/security-overview.md
+++ b/site2/website/versioned_docs/version-2.6.1/security-overview.md
@@ -11,6 +11,10 @@ By default, Pulsar configures no encryption, authentication, or authorization. A
Pulsar supports a pluggable authentication mechanism. And Pulsar clients use this mechanism to authenticate with brokers and proxies. You can also configure Pulsar to support multiple authentication sources.
+The Pulsar broker validates the authentication credentials when a connection is established. After the initial connection is authenticated, the "principal" token is stored for authorization though the connection is not re-authenticated. The broker periodically checks the expiration status of every `ServerCnx` object. You can set the `authenticationRefreshCheckSeconds` on the broker to control the frequency to check the expiration status. By default, the `authenticationRefreshCheckSeconds [...]
+
+The broker supports learning whether a particular client supports authentication refreshing. If a client supports authentication refreshing and the credential is expired, the authentication provider calls the `refreshAuthentication` method to initiate the refreshing process. If a client does not support authentication refreshing and the credential is expired, the broker disconnects the client.
+
You had better secure the service components in your Apache Pulsar deployment.
## Role tokens
diff --git a/site2/website/versioned_docs/version-2.6.1/sql-deployment-configurations.md b/site2/website/versioned_docs/version-2.6.1/sql-deployment-configurations.md
index 5c0c555..4a368c9 100644
--- a/site2/website/versioned_docs/version-2.6.1/sql-deployment-configurations.md
+++ b/site2/website/versioned_docs/version-2.6.1/sql-deployment-configurations.md
@@ -44,7 +44,7 @@ $ wget pulsar:binary_release_url
## Deploy a new cluster
-Since Pulsar SQL is powered by [Presto](https://prestosql.io), the configuration for deployment is the same for the Pulsar SQL worker.
+Since Pulsar SQL is powered by [Presto](https://prestodb.io), the configuration for deployment is the same for the Pulsar SQL worker.
> Note
> For how to set up a standalone single node environment, refer to [Query data](sql-getting-started.md).
@@ -153,7 +153,7 @@ presto> SELECT * FROM system.runtime.nodes;
2 | http://192.168.2.3:8081 | testversion | false | active
```
-For more information about deployment in Presto, refer to [Presto deployment](https://prestosql.io/docs/current/installation/deployment.html).
+For more information about deployment in Presto, refer to [Presto deployment](https://prestodb.io/docs/current/installation/deployment.html).
> Note
-> The broker does not advance LAC, so when Pulsar SQL bypass broker to query data, it can only read entries up to the LAC that all the bookies learned. You can enable periodically write LAC on the broker by setting "bookkeeperExplicitLacIntervalInMills" in the broker.conf.
+> The broker does not advance LAC, so when Pulsar SQL bypass broker to query data, it can only read entries up to the LAC that all the bookies learned. You can enable periodically write LAC on the broker by setting "bookkeeperExplicitLacIntervalInMills" in the broker.conf.
\ No newline at end of file
diff --git a/site2/website/versioned_docs/version-2.6.1/window-functions-context.md b/site2/website/versioned_docs/version-2.6.1/window-functions-context.md
new file mode 100644
index 0000000..6164ef4
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.6.1/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.6.1-window-functions-context
+title: Window Functions Context
+sidebar_label: "Window Functions: Context"
+original_id: window-functions-context
+---
+
+Java SDK provides access to a **window context object** that can be used by a window function. This context object provides a wide variety of information and functionality for Pulsar window functions as below.
+
+- [Spec](#spec)
+
+ * Names of all input topics and the output topic associated with the function.
+ * Tenant and namespace associated with the function.
+ * Pulsar window function name, ID, and version.
+ * ID of the Pulsar function instance running the window function.
+ * Number of instances that invoke the window function.
+ * Built-in type or custom class name of the output schema.
+
+- [Logger](#logger)
+
+ * Logger object used by the window function, which can be used to create window function log messages.
+
+- [User config](#user-config)
+
+ * Access to arbitrary user configuration values.
+
+- [Routing](#routing)
+
+ * Routing is supported in Pulsar window functions. Pulsar window functions send messages to arbitrary topics as per the `publish` interface.
+
+- [Metrics](#metrics)
+
+ * Interface for recording metrics.
+
+- [State storage](#state-storage)
+
+ * Interface for storing and retrieving state in [state storage](#state-storage).
+
+## Spec
+
+Spec contains the basic information of a function.
+
+### Get input topics
+
+The `getInputTopics` method gets the **name list** of all input topics.
+
+This example demonstrates how to get the name list of all input topics in a Java window function.
+
+```java
+public class GetInputTopicsWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ Collection<String> inputTopics = context.getInputTopics();
+ System.out.println(inputTopics);
+
+ return null;
+ }
+
+}
+```
+
+### Get output topic
+
+The `getOutputTopic` method gets the **name of a topic** to which the message is sent.
+
+This example demonstrates how to get the name of an output topic in a Java window function.
+
+```java
+public class GetOutputTopicWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String outputTopic = context.getOutputTopic();
+ System.out.println(outputTopic);
+
+ return null;
+ }
+}
+```
+
+### Get tenant
+
+The `getTenant` method gets the tenant name associated with the window function.
+
+This example demonstrates how to get the tenant name in a Java window function.
+
+```java
+public class GetTenantWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String tenant = context.getTenant();
+ System.out.println(tenant);
+
+ return null;
+ }
+
+}
+```
+
+### Get namespace
+
+The `getNamespace` method gets the namespace associated with the window function.
+
+This example demonstrates how to get the namespace in a Java window function.
+
+```java
+public class GetNamespaceWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String ns = context.getNamespace();
+ System.out.println(ns);
+
+ return null;
+ }
+
+}
+```
+
+### Get function name
+
+The `getFunctionName` method gets the window function name.
+
+This example demonstrates how to get the function name in a Java window function.
+
+```java
+public class GetNameOfWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String functionName = context.getFunctionName();
+ System.out.println(functionName);
+
+ return null;
+ }
+
+}
+```
+
+### Get function ID
+
+The `getFunctionId` method gets the window function ID.
+
+This example demonstrates how to get the function ID in a Java window function.
+
+```java
+public class GetFunctionIDWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String functionID = context.getFunctionId();
+ System.out.println(functionID);
+
+ return null;
+ }
+
+}
+```
+
+### Get function version
+
+The `getFunctionVersion` method gets the window function version.
+
+This example demonstrates how to get the function version of a Java window function.
+
+```java
+public class GetVersionOfWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String functionVersion = context.getFunctionVersion();
+ System.out.println(functionVersion);
+
+ return null;
+ }
+
+}
+```
+
+### Get instance ID
+
+The `getInstanceId` method gets the instance ID of a window function.
+
+This example demonstrates how to get the instance ID in a Java window function.
+
+```java
+public class GetInstanceIDWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ int instanceId = context.getInstanceId();
+ System.out.println(instanceId);
+
+ return null;
+ }
+
+}
+```
+
+### Get num instances
+
+The `getNumInstances` method gets the number of instances that invoke the window function.
+
+This example demonstrates how to get the number of instances in a Java window function.
+
+```java
+public class GetNumInstancesWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ int numInstances = context.getNumInstances();
+ System.out.println(numInstances);
+
+ return null;
+ }
+
+}
+```
+
+### Get output schema type
+
+The `getOutputSchemaType` method gets the built-in type or custom class name of the output schema.
+
+This example demonstrates how to get the output schema type of a Java window function.
+
+```java
+public class GetOutputSchemaTypeWindowFunction implements WindowFunction<String, Void> {
+
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String schemaType = context.getOutputSchemaType();
+ System.out.println(schemaType);
+
+ return null;
+ }
+}
+```
+
+## Logger
+
+Pulsar window functions using Java SDK has access to an [SLF4j](https://www.slf4j.org/) [`Logger`](https://www.slf4j.org/api/org/apache/log4j/Logger.html) object that can be used to produce logs at the chosen log level.
+
+This example logs either a `WARNING`-level or `INFO`-level log based on whether the incoming string contains the word `danger` or not in a Java function.
+
+```java
+import java.util.Collection;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.functions.api.WindowContext;
+import org.apache.pulsar.functions.api.WindowFunction;
+import org.slf4j.Logger;
+
+public class LoggingWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ Logger log = context.getLogger();
+ for (Record<String> record : inputs) {
+ log.info(record + "-window-log");
+ }
+ return null;
+ }
+
+}
+```
+
+If you need your function to produce logs, specify a log topic when creating or running the function.
+
+```bash
+bin/pulsar-admin functions create \
+ --jar my-functions.jar \
+ --classname my.package.LoggingFunction \
+ --log-topic persistent://public/default/logging-function-logs \
+ # Other function configs
+```
+
+You can access all logs produced by `LoggingFunction` via the `persistent://public/default/logging-function-logs` topic.
+
+## Metrics
+
+Pulsar window functions can publish arbitrary metrics to the metrics interface which can be queried.
+
+> **Note**
+>
+> If a Pulsar window function uses the language-native interface for Java, that function is not able to publish metrics and stats to Pulsar.
+
+You can record metrics using the context object on a per-key basis.
+
+This example sets a metric for the `process-count` key and a different metric for the `elevens-count` key every time the function processes a message in a Java function.
+
+```java
+import java.util.Collection;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.functions.api.WindowContext;
+import org.apache.pulsar.functions.api.WindowFunction;
+
+
+/**
+ * Example function that wants to keep track of
+ * the event time of each message sent.
+ */
+public class UserMetricWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+
+ for (Record<String> record : inputs) {
+ if (record.getEventTime().isPresent()) {
+ context.recordMetric("MessageEventTime", record.getEventTime().get().doubleValue());
+ }
+ }
+
+ return null;
+ }
+}
+```
+
+## User config
+
+When you run or update Pulsar Functions that are created using SDK, you can pass arbitrary key/value pairs to them with the `--user-config` flag. Key/value pairs **must** be specified as JSON.
+
+This example passes a user configured key/value to a function.
+
+```bash
+bin/pulsar-admin functions create \
+ --name word-filter \
+ --user-config '{"forbidden-word":"rosebud"}' \
+ # Other function configs
+```
+
+### API
+You can use the following APIs to get user-defined information for window functions.
+#### getUserConfigMap
+
+`getUserConfigMap` API gets a map of all user-defined key/value configurations for the window function.
+
+
+```java
+/**
+ * Get a map of all user-defined key/value configs for the function.
+ *
+ * @return The full map of user-defined config values
+ */
+ Map<String, Object> getUserConfigMap();
+```
+
+
+#### getUserConfigValue
+
+The `getUserConfigValue` API gets a user-defined key/value.
+
+```java
+/**
+ * Get any user-defined key/value.
+ *
+ * @param key The key
+ * @return The Optional value specified by the user for that key.
+ */
+ Optional<Object> getUserConfigValue(String key);
+```
+
+#### getUserConfigValueOrDefault
+
+The `getUserConfigValueOrDefault` API gets a user-defined key/value or a default value if none is present.
+
+```java
+/**
+ * Get any user-defined key/value or a default value if none is present.
+ *
+ * @param key
+ * @param defaultValue
+ * @return Either the user config value associated with a given key or a supplied default value
+ */
+ Object getUserConfigValueOrDefault(String key, Object defaultValue);
+```
+
+This example demonstrates how to access key/value pairs provided to Pulsar window functions.
+
+Java SDK context object enables you to access key/value pairs provided to Pulsar window functions via the command line (as JSON).
+
+>**Tip**
+>
+> For all key/value pairs passed to Java window functions, both the `key` and the `value` are `String`. To set the value to be a different type, you need to deserialize it from the `String` type.
+
+This example passes a key/value pair in a Java window function.
+
+```bash
+bin/pulsar-admin functions create \
+ --user-config '{"word-of-the-day":"verdure"}' \
+ # Other function configs
+ ```
+
+This example accesses values in a Java window function.
+
+The `UserConfigFunction` function logs the string `"The word of the day is verdure"` every time the function is invoked (which means every time a message arrives). The user config of `word-of-the-day` is changed **only** when the function is updated with a new config value via
+multiple ways, such as the command line tool or REST API.
+
+```java
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+import org.slf4j.Logger;
+
+import java.util.Optional;
+
+public class UserConfigWindowFunction implements WindowFunction<String, String> {
+ @Override
+ public String process(Collection<Record<String>> input, WindowContext context) throws Exception {
+ Optional<Object> whatToWrite = context.getUserConfigValue("WhatToWrite");
+ if (whatToWrite.get() != null) {
+ return (String)whatToWrite.get();
+ } else {
+ return "Not a nice way";
+ }
+ }
+
+}
+```
+
+If no value is provided, you can access the entire user config map or set a default value.
+
+```java
+// Get the whole config map
+Map<String, String> allConfigs = context.getUserConfigMap();
+
+// Get value or resort to default
+String wotd = context.getUserConfigValueOrDefault("word-of-the-day", "perspicacious");
+```
+
+## Routing
+
+You can use the `context.publish()` interface to publish as many results as you want.
+
+This example shows that the `PublishFunction` class uses the built-in function in the context to publish messages to the `publishTopic` in a Java function.
+
+```java
+public class PublishWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> input, WindowContext context) throws Exception {
+ String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic");
+ String output = String.format("%s!", input);
+ context.publish(publishTopic, output);
+
+ return null;
+ }
+
+}
+```
+
+## State storage
+
+Pulsar window functions use [Apache BookKeeper](https://bookkeeper.apache.org) as a state storage interface. Apache Pulsar installation (including the standalone installation) includes the deployment of BookKeeper bookies.
+
+Apache Pulsar integrates with Apache BookKeeper `table service` to store the `state` for functions. For example, the `WordCount` function can store its `counters` state into BookKeeper table service via Pulsar Functions state APIs.
+
+States are key-value pairs, where the key is a string and the value is arbitrary binary data—counters are stored as 64-bit big-endian binary values. Keys are scoped to an individual Pulsar Function and shared between instances of that function.
+
+Currently, Pulsar window functions expose Java API to access, update, and manage states. These APIs are available in the context object when you use Java SDK functions.
+
+| Java API| Description
+|---|---
+|`incrCounter`|Increases a built-in distributed counter referred by key.
+|`getCounter`|Gets the counter value for the key.
+|`putState`|Updates the state value for the key.
+
+You can use the following APIs to access, update, and manage states in Java window functions.
+
+#### incrCounter
+
+The `incrCounter` API increases a built-in distributed counter referred by key.
+
+Applications use the `incrCounter` API to change the counter of a given `key` by the given `amount`. If the `key` does not exist, a new key is created.
+
+```java
+ /**
+ * Increment the builtin distributed counter referred by key
+ * @param key The name of the key
+ * @param amount The amount to be incremented
+ */
+ void incrCounter(String key, long amount);
+```
+
+#### getCounter
+
+The `getCounter` API gets the counter value for the key.
+
+Applications uses the `getCounter` API to retrieve the counter of a given `key` changed by the `incrCounter` API.
+
+```java
+ /**
+ * Retrieve the counter value for the key.
+ *
+ * @param key name of the key
+ * @return the amount of the counter value for this key
+ */
+ long getCounter(String key);
+```
+
+Except the `getCounter` API, Pulsar also exposes a general key/value API (`putState`) for functions to store general key/value state.
+
+#### putState
+
+The `putState` API updates the state value for the key.
+
+```java
+ /**
+ * Update the state value for the key.
+ *
+ * @param key name of the key
+ * @param value state value of the key
+ */
+ void putState(String key, ByteBuffer value);
+```
+
+This example demonstrates how applications store states in Pulsar window functions.
+
+The logic of the `WordCountWindowFunction` is simple and straightforward.
+
+1. The function first splits the received string into multiple words using regex `\\.`.
+
+2. For each `word`, the function increments the corresponding `counter` by 1 via `incrCounter(key, amount)`.
+
+```java
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+
+import java.util.Arrays;
+
+public class WordCountWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ for (Record<String> input : inputs) {
+ Arrays.asList(input.getValue().split("\\.")).forEach(word -> context.incrCounter(word, 1));
+ }
+ return null;
+
+ }
+}
+```
+
diff --git a/site2/website/versioned_sidebars/version-2.6.1-sidebars.json b/site2/website/versioned_sidebars/version-2.6.1-sidebars.json
new file mode 100644
index 0000000..57f5651
--- /dev/null
+++ b/site2/website/versioned_sidebars/version-2.6.1-sidebars.json
@@ -0,0 +1,155 @@
+{
+ "docs": {
+ "Get started": [
+ "version-2.6.1-pulsar-2.0",
+ "version-2.6.1-standalone",
+ "version-2.6.1-standalone-docker",
+ "version-2.6.1-kubernetes-helm",
+ "version-2.6.1-client-libraries"
+ ],
+ "Concepts and Architecture": [
+ "version-2.6.1-concepts-overview",
+ "version-2.6.1-concepts-messaging",
+ "version-2.6.1-concepts-architecture-overview",
+ "version-2.6.1-concepts-clients",
+ "version-2.6.1-concepts-replication",
+ "version-2.6.1-concepts-multi-tenancy",
+ "version-2.6.1-concepts-authentication",
+ "version-2.6.1-concepts-topic-compaction",
+ "version-2.6.1-concepts-tiered-storage",
+ "version-2.6.1-concepts-proxy-sni-routing"
+ ],
+ "Pulsar Schema": [
+ "version-2.6.1-schema-get-started",
+ "version-2.6.1-schema-understand",
+ "version-2.6.1-schema-evolution-compatibility",
+ "version-2.6.1-schema-manage"
+ ],
+ "Pulsar Functions": [
+ "version-2.6.1-functions-overview",
+ "version-2.6.1-functions-worker",
+ "version-2.6.1-functions-runtime",
+ "version-2.6.1-functions-develop",
+ "version-2.6.1-functions-debug",
+ "version-2.6.1-functions-deploy",
+ "version-2.6.1-functions-cli",
+ "version-2.6.1-window-functions-context"
+ ],
+ "Pulsar IO": [
+ "version-2.6.1-io-overview",
+ "version-2.6.1-io-quickstart",
+ "version-2.6.1-io-use",
+ "version-2.6.1-io-debug",
+ "version-2.6.1-io-connectors",
+ "version-2.6.1-io-cdc",
+ "version-2.6.1-io-develop",
+ "version-2.6.1-io-cli"
+ ],
+ "Pulsar SQL": [
+ "version-2.6.1-sql-overview",
+ "version-2.6.1-sql-getting-started",
+ "version-2.6.1-sql-deployment-configurations",
+ "version-2.6.1-sql-rest-api"
+ ],
+ "Kubernetes (Helm)": [
+ "version-2.6.1-helm-overview",
+ "version-2.6.1-helm-prepare",
+ "version-2.6.1-helm-install",
+ "version-2.6.1-helm-deploy",
+ "version-2.6.1-helm-upgrade",
+ "version-2.6.1-helm-tools"
+ ],
+ "Deployment": [
+ "version-2.6.1-deploy-aws",
+ "version-2.6.1-deploy-kubernetes",
+ "version-2.6.1-deploy-bare-metal",
+ "version-2.6.1-deploy-bare-metal-multi-cluster",
+ "version-2.6.1-deploy-dcos",
+ "version-2.6.1-deploy-monitoring"
+ ],
+ "Administration": [
+ "version-2.6.1-administration-zk-bk",
+ "version-2.6.1-administration-geo",
+ "version-2.6.1-administration-pulsar-manager",
+ "version-2.6.1-administration-stats",
+ "version-2.6.1-administration-load-balance",
+ "version-2.6.1-administration-proxy",
+ "version-2.6.1-administration-upgrade"
+ ],
+ "Security": [
+ "version-2.6.1-security-overview",
+ "version-2.6.1-security-tls-transport",
+ "version-2.6.1-security-tls-authentication",
+ "version-2.6.1-security-tls-keystore",
+ "version-2.6.1-security-jwt",
+ "version-2.6.1-security-athenz",
+ "version-2.6.1-security-kerberos",
+ "version-2.6.1-security-oauth2",
+ "version-2.6.1-security-authorization",
+ "version-2.6.1-security-encryption",
+ "version-2.6.1-security-extending",
+ "version-2.6.1-security-bouncy-castle"
+ ],
+ "Performance": [
+ "version-2.6.1-performance-pulsar-perf"
+ ],
+ "Client libraries": [
+ "version-2.6.1-client-libraries-java",
+ "version-2.6.1-client-libraries-go",
+ "version-2.6.1-client-libraries-python",
+ "version-2.6.1-client-libraries-cpp",
+ "version-2.6.1-client-libraries-node",
+ "version-2.6.1-client-libraries-websocket",
+ "version-2.6.1-client-libraries-dotnet"
+ ],
+ "Admin API": [
+ "version-2.6.1-admin-api-overview",
+ "version-2.6.1-admin-api-clusters",
+ "version-2.6.1-admin-api-tenants",
+ "version-2.6.1-admin-api-brokers",
+ "version-2.6.1-admin-api-namespaces",
+ "version-2.6.1-admin-api-permissions",
+ "version-2.6.1-admin-api-persistent-topics",
+ "version-2.6.1-admin-api-non-persistent-topics",
+ "version-2.6.1-admin-api-partitioned-topics",
+ "version-2.6.1-admin-api-non-partitioned-topics",
+ "version-2.6.1-admin-api-schemas",
+ "version-2.6.1-admin-api-functions"
+ ],
+ "Adaptors": [
+ "version-2.6.1-adaptors-kafka",
+ "version-2.6.1-adaptors-spark",
+ "version-2.6.1-adaptors-storm"
+ ],
+ "Cookbooks": [
+ "version-2.6.1-cookbooks-tiered-storage",
+ "version-2.6.1-cookbooks-compaction",
+ "version-2.6.1-cookbooks-deduplication",
+ "version-2.6.1-cookbooks-non-persistent",
+ "version-2.6.1-cookbooks-partitioned",
+ "version-2.6.1-cookbooks-retention-expiry",
+ "version-2.6.1-cookbooks-encryption",
+ "version-2.6.1-cookbooks-message-queue",
+ "version-2.6.1-cookbooks-bookkeepermetadata"
+ ],
+ "Development": [
+ "version-2.6.1-develop-tools",
+ "version-2.6.1-develop-binary-protocol",
+ "version-2.6.1-develop-schema",
+ "version-2.6.1-develop-load-manager",
+ "version-2.6.1-develop-cpp"
+ ],
+ "Reference": [
+ "version-2.6.1-reference-terminology",
+ "version-2.6.1-reference-cli-tools",
+ "version-2.6.1-reference-configuration",
+ "version-2.6.1-reference-metrics"
+ ]
+ },
+ "docs-other": {
+ "First Category": [
+ "version-2.6.1-doc4",
+ "version-2.6.1-doc5"
+ ]
+ }
+}
diff --git a/site2/website/versions.json b/site2/website/versions.json
index 7874ef0..5d6a559 100644
--- a/site2/website/versions.json
+++ b/site2/website/versions.json
@@ -1,4 +1,5 @@
[
+ "2.6.1",
"2.6.0",
"2.5.2",
"2.5.1",