You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by li...@apache.org on 2021/12/07 05:07:45 UTC
[pulsar] branch master updated: [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / client/admin api/adaptors (#13145)
This is an automated email from the ASF dual-hosted git repository.
liuyu 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 98e4ffd [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / client/admin api/adaptors (#13145)
98e4ffd is described below
commit 98e4ffdb47b0b3404a121f5f5f35205442ac2a22
Author: Li Li <ur...@gmail.com>
AuthorDate: Tue Dec 7 13:06:42 2021 +0800
[website][upgrade]feat: website upgrade / docs migration - 2.4.1 / client/admin api/adaptors (#13145)
* [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / get started/concepts/schema
Signed-off-by: LiLi <ur...@gmail.com>
* [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / functions/io/sql
Signed-off-by: LiLi <ur...@gmail.com>
* [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / deployment/administration/security
Signed-off-by: LiLi <ur...@gmail.com>
* [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / client/admin api/adaptors
Signed-off-by: LiLi <ur...@gmail.com>
Co-authored-by: Anonymitaet <50...@users.noreply.github.com>
---
.../versioned_docs/version-2.4.1/adaptors-kafka.md | 274 ++++
.../versioned_docs/version-2.4.1/adaptors-spark.md | 91 ++
.../versioned_docs/version-2.4.1/adaptors-storm.md | 96 ++
.../version-2.4.1/admin-api-brokers.md | 272 ++++
.../version-2.4.1/admin-api-clusters.md | 304 +++++
.../version-2.4.1/admin-api-functions.md | 619 +++++++++
.../version-2.4.1/admin-api-namespaces.md | 1374 ++++++++++++++++++++
.../admin-api-non-persistent-topics.md | 286 ++++
.../version-2.4.1/admin-api-overview.md | 97 ++
.../version-2.4.1/admin-api-partitioned-topics.md | 408 ++++++
.../version-2.4.1/admin-api-permissions.md | 181 +++
.../version-2.4.1/admin-api-persistent-topics.md | 8 +
.../version-2.4.1/admin-api-schemas.md | 7 +
.../version-2.4.1/admin-api-tenants.md | 226 ++++
.../version-2.4.1/client-libraries-cpp.md | 407 ++++++
.../version-2.4.1/client-libraries-go.md | 884 +++++++++++++
.../version-2.4.1/client-libraries-java.md | 809 ++++++++++++
.../version-2.4.1/client-libraries-python.md | 311 +++++
.../version-2.4.1/client-libraries-websocket.md | 621 +++++++++
.../versioned_sidebars/version-2.4.1-sidebars.json | 94 ++
20 files changed, 7369 insertions(+)
diff --git a/site2/website-next/versioned_docs/version-2.4.1/adaptors-kafka.md b/site2/website-next/versioned_docs/version-2.4.1/adaptors-kafka.md
new file mode 100644
index 0000000..ad0d886
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/adaptors-kafka.md
@@ -0,0 +1,274 @@
+---
+id: adaptors-kafka
+title: Pulsar adaptor for Apache Kafka
+sidebar_label: "Kafka client wrapper"
+original_id: adaptors-kafka
+---
+
+
+Pulsar provides an easy option for applications that are currently written using the [Apache Kafka](http://kafka.apache.org) Java client API.
+
+## Using the Pulsar Kafka compatibility wrapper
+
+In an existing application, change the regular Kafka client dependency and replace it with the Pulsar Kafka wrapper. Remove the following dependency in `pom.xml`:
+
+```xml
+
+<dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>0.10.2.1</version>
+</dependency>
+
+```
+
+Then include this dependency for the Pulsar Kafka wrapper:
+
+```xml
+
+<dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client-kafka</artifactId>
+ <version>@pulsar:version@</version>
+</dependency>
+
+```
+
+With the new dependency, the existing code works without any changes. You need to adjust the configuration, and make sure it points the
+producers and consumers to Pulsar service rather than Kafka, and uses a particular
+Pulsar topic.
+
+## Using the Pulsar Kafka compatibility wrapper together with existing kafka client
+
+When migrating from Kafka to Pulsar, the application might use the original kafka client
+and the pulsar kafka wrapper together during migration. You should consider using the
+unshaded pulsar kafka client wrapper.
+
+```xml
+
+<dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client-kafka-original</artifactId>
+ <version>@pulsar:version@</version>
+</dependency>
+
+```
+
+When using this dependency, construct producers using `org.apache.kafka.clients.producer.PulsarKafkaProducer`
+instead of `org.apache.kafka.clients.producer.KafkaProducer` and `org.apache.kafka.clients.producer.PulsarKafkaConsumer` for consumers.
+
+## Producer example
+
+```java
+
+// Topic needs to be a regular Pulsar topic
+String topic = "persistent://public/default/my-topic";
+
+Properties props = new Properties();
+// Point to a Pulsar service
+props.put("bootstrap.servers", "pulsar://localhost:6650");
+
+props.put("key.serializer", IntegerSerializer.class.getName());
+props.put("value.serializer", StringSerializer.class.getName());
+
+Producer<Integer, String> producer = new KafkaProducer(props);
+
+for (int i = 0; i < 10; i++) {
+ producer.send(new ProducerRecord<Integer, String>(topic, i, "hello-" + i));
+ log.info("Message {} sent successfully", i);
+}
+
+producer.close();
+
+```
+
+## Consumer example
+
+```java
+
+String topic = "persistent://public/default/my-topic";
+
+Properties props = new Properties();
+// Point to a Pulsar service
+props.put("bootstrap.servers", "pulsar://localhost:6650");
+props.put("group.id", "my-subscription-name");
+props.put("enable.auto.commit", "false");
+props.put("key.deserializer", IntegerDeserializer.class.getName());
+props.put("value.deserializer", StringDeserializer.class.getName());
+
+Consumer<Integer, String> consumer = new KafkaConsumer(props);
+consumer.subscribe(Arrays.asList(topic));
+
+while (true) {
+ ConsumerRecords<Integer, String> records = consumer.poll(100);
+ records.forEach(record -> {
+ log.info("Received record: {}", record);
+ });
+
+ // Commit last offset
+ consumer.commitSync();
+}
+
+```
+
+## Complete Examples
+
+You can find the complete producer and consumer examples [here](https://github.com/apache/pulsar-adapters/tree/master/pulsar-client-kafka-compat/pulsar-client-kafka-tests/src/test/java/org/apache/pulsar/client/kafka/compat/examples).
+
+## Compatibility matrix
+
+Currently the Pulsar Kafka wrapper supports most of the operations offered by the Kafka API.
+
+### Producer
+
+APIs:
+
+| Producer Method | Supported | Notes |
+|:------------------------------------------------------------------------------|:----------|:-------------------------------------------------------------------------|
+| `Future<RecordMetadata> send(ProducerRecord<K, V> record)` | Yes | |
+| `Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)` | Yes | |
+| `void flush()` | Yes | |
+| `List<PartitionInfo> partitionsFor(String topic)` | No | |
+| `Map<MetricName, ? extends Metric> metrics()` | No | |
+| `void close()` | Yes | |
+| `void close(long timeout, TimeUnit unit)` | Yes | |
+
+Properties:
+
+| Config property | Supported | Notes |
+|:----------------------------------------|:----------|:------------------------------------------------------------------------------|
+| `acks` | Ignored | Durability and quorum writes are configured at the namespace level |
+| `auto.offset.reset` | Yes | It uses a default value of `earliest` if you do not give a specific setting. |
+| `batch.size` | Ignored | |
+| `bootstrap.servers` | Yes | |
+| `buffer.memory` | Ignored | |
+| `client.id` | Ignored | |
+| `compression.type` | Yes | Allows `gzip` and `lz4`. No `snappy`. |
+| `connections.max.idle.ms` | Yes | Only support up to 2,147,483,647,000(Integer.MAX_VALUE * 1000) ms of idle time|
+| `interceptor.classes` | Yes | |
+| `key.serializer` | Yes | |
+| `linger.ms` | Yes | Controls the group commit time when batching messages |
+| `max.block.ms` | Ignored | |
+| `max.in.flight.requests.per.connection` | Ignored | In Pulsar ordering is maintained even with multiple requests in flight |
+| `max.request.size` | Ignored | |
+| `metric.reporters` | Ignored | |
+| `metrics.num.samples` | Ignored | |
+| `metrics.sample.window.ms` | Ignored | |
+| `partitioner.class` | Yes | |
+| `receive.buffer.bytes` | Ignored | |
+| `reconnect.backoff.ms` | Ignored | |
+| `request.timeout.ms` | Ignored | |
+| `retries` | Ignored | Pulsar client retries with exponential backoff until the send timeout expires. |
+| `send.buffer.bytes` | Ignored | |
+| `timeout.ms` | Yes | |
+| `value.serializer` | Yes | |
+
+
+### Consumer
+
+The following table lists consumer APIs.
+
+| Consumer Method | Supported | Notes |
+|:--------------------------------------------------------------------------------------------------------|:----------|:------|
+| `Set<TopicPartition> assignment()` | No | |
+| `Set<String> subscription()` | Yes | |
+| `void subscribe(Collection<String> topics)` | Yes | |
+| `void subscribe(Collection<String> topics, ConsumerRebalanceListener callback)` | No | |
+| `void assign(Collection<TopicPartition> partitions)` | No | |
+| `void subscribe(Pattern pattern, ConsumerRebalanceListener callback)` | No | |
+| `void unsubscribe()` | Yes | |
+| `ConsumerRecords<K, V> poll(long timeoutMillis)` | Yes | |
+| `void commitSync()` | Yes | |
+| `void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets)` | Yes | |
+| `void commitAsync()` | Yes | |
+| `void commitAsync(OffsetCommitCallback callback)` | Yes | |
+| `void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)` | Yes | |
+| `void seek(TopicPartition partition, long offset)` | Yes | |
+| `void seekToBeginning(Collection<TopicPartition> partitions)` | Yes | |
+| `void seekToEnd(Collection<TopicPartition> partitions)` | Yes | |
+| `long position(TopicPartition partition)` | Yes | |
+| `OffsetAndMetadata committed(TopicPartition partition)` | Yes | |
+| `Map<MetricName, ? extends Metric> metrics()` | No | |
+| `List<PartitionInfo> partitionsFor(String topic)` | No | |
+| `Map<String, List<PartitionInfo>> listTopics()` | No | |
+| `Set<TopicPartition> paused()` | No | |
+| `void pause(Collection<TopicPartition> partitions)` | No | |
+| `void resume(Collection<TopicPartition> partitions)` | No | |
+| `Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)` | No | |
+| `Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions)` | No | |
+| `Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions)` | No | |
+| `void close()` | Yes | |
+| `void close(long timeout, TimeUnit unit)` | Yes | |
+| `void wakeup()` | No | |
+
+Properties:
+
+| Config property | Supported | Notes |
+|:--------------------------------|:----------|:------------------------------------------------------|
+| `group.id` | Yes | Maps to a Pulsar subscription name |
+| `max.poll.records` | Yes | |
+| `max.poll.interval.ms` | Ignored | Messages are "pushed" from broker |
+| `session.timeout.ms` | Ignored | |
+| `heartbeat.interval.ms` | Ignored | |
+| `bootstrap.servers` | Yes | Needs to point to a single Pulsar service URL |
+| `enable.auto.commit` | Yes | |
+| `auto.commit.interval.ms` | Ignored | With auto-commit, acks are sent immediately to broker |
+| `partition.assignment.strategy` | Ignored | |
+| `auto.offset.reset` | Yes | Only support earliest and latest. |
+| `fetch.min.bytes` | Ignored | |
+| `fetch.max.bytes` | Ignored | |
+| `fetch.max.wait.ms` | Ignored | |
+| `interceptor.classes` | Yes | |
+| `metadata.max.age.ms` | Ignored | |
+| `max.partition.fetch.bytes` | Ignored | |
+| `send.buffer.bytes` | Ignored | |
+| `receive.buffer.bytes` | Ignored | |
+| `client.id` | Ignored | |
+
+
+## Customize Pulsar configurations
+
+You can configure Pulsar authentication provider directly from the Kafka properties.
+
+### Pulsar client properties
+
+| Config property | Default | Notes |
+|:---------------------------------------|:--------|:---------------------------------------------------------------------------------------|
+| [`pulsar.authentication.class`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setAuthentication-org.apache.pulsar.client.api.Authentication-) | | Configure to auth provider. For example, `org.apache.pulsar.client.impl.auth.AuthenticationTls`.|
+| [`pulsar.authentication.params.map`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setAuthentication-java.lang.String-java.util.Map-) | | Map which represents parameters for the Authentication-Plugin. |
+| [`pulsar.authentication.params.string`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setAuthentication-java.lang.String-java.lang.String-) | | String which represents parameters for the Authentication-Plugin, for example, `key1:val1,key2:val2`. |
+| [`pulsar.use.tls`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setUseTls-boolean-) | `false` | Enable TLS transport encryption. |
+| [`pulsar.tls.trust.certs.file.path`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setTlsTrustCertsFilePath-java.lang.String-) | | Path for the TLS trust certificate store. |
+| [`pulsar.tls.allow.insecure.connection`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setTlsAllowInsecureConnection-boolean-) | `false` | Accept self-signed certificates from brokers. |
+| [`pulsar.operation.timeout.ms`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setOperationTimeout-int-java.util.concurrent.TimeUnit-) | `30000` | General operations timeout. |
+| [`pulsar.stats.interval.seconds`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setStatsInterval-long-java.util.concurrent.TimeUnit-) | `60` | Pulsar client lib stats printing interval. |
+| [`pulsar.num.io.threads`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setIoThreads-int-) | `1` | The number of Netty IO threads to use. |
+| [`pulsar.connections.per.broker`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setConnectionsPerBroker-int-) | `1` | The maximum number of connection to each broker. |
+| [`pulsar.use.tcp.nodelay`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setUseTcpNoDelay-boolean-) | `true` | TCP no-delay. |
+| [`pulsar.concurrent.lookup.requests`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setConcurrentLookupRequest-int-) | `50000` | The maximum number of concurrent topic lookups. |
+| [`pulsar.max.number.rejected.request.per.connection`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientConfiguration.html#setMaxNumberOfRejectedRequestPerConnection-int-) | `50` | The threshold of errors to forcefully close a connection. |
+| [`pulsar.keepalive.interval.ms`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ClientBuilder.html#keepAliveInterval-int-java.util.concurrent.TimeUnit-)| `30000` | Keep alive interval for each client-broker-connection. |
+
+
+### Pulsar producer properties
+
+| Config property | Default | Notes |
+|:---------------------------------------|:--------|:---------------------------------------------------------------------------------------|
+| [`pulsar.producer.name`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.html#setProducerName-java.lang.String-) | | Specify the producer name. |
+| [`pulsar.producer.initial.sequence.id`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.html#setInitialSequenceId-long-) | | Specify baseline for sequence ID of this producer. |
+| [`pulsar.producer.max.pending.messages`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.html#setMaxPendingMessages-int-) | `1000` | Set the maximum size of the message queue pending to receive an acknowledgment from the broker. |
+| [`pulsar.producer.max.pending.messages.across.partitions`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.html#setMaxPendingMessagesAcrossPartitions-int-) | `50000` | Set the maximum number of pending messages across all the partitions. |
+| [`pulsar.producer.batching.enabled`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.html#setBatchingEnabled-boolean-) | `true` | Control whether automatic batching of messages is enabled for the producer. |
+| [`pulsar.producer.batching.max.messages`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.html#setBatchingMaxMessages-int-) | `1000` | The maximum number of messages in a batch. |
+| [`pulsar.block.if.producer.queue.full`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.html#setBlockIfQueueFull-boolean-) | | Specify the block producer if queue is full. |
+
+
+### Pulsar consumer Properties
+
+| Config property | Default | Notes |
+|:---------------------------------------|:--------|:---------------------------------------------------------------------------------------|
+| [`pulsar.consumer.name`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerConfiguration.html#setConsumerName-java.lang.String-) | | Specify the consumer name. |
+| [`pulsar.consumer.receiver.queue.size`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerConfiguration.html#setReceiverQueueSize-int-) | 1000 | Set the size of the consumer receiver queue. |
+| [`pulsar.consumer.acknowledgments.group.time.millis`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.html#acknowledgmentGroupTime-long-java.util.concurrent.TimeUnit-) | 100 | Set the maximum amount of group time for consumers to send the acknowledgments to the broker. |
+| [`pulsar.consumer.total.receiver.queue.size.across.partitions`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerConfiguration.html#setMaxTotalReceiverQueueSizeAcrossPartitions-int-) | 50000 | Set the maximum size of the total receiver queue across partitions. |
+| [`pulsar.consumer.subscription.topics.mode`](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.html#subscriptionTopicsMode-Mode-) | PersistentOnly | Set the subscription topic mode for consumers. |
diff --git a/site2/website-next/versioned_docs/version-2.4.1/adaptors-spark.md b/site2/website-next/versioned_docs/version-2.4.1/adaptors-spark.md
new file mode 100644
index 0000000..e14f13b
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/adaptors-spark.md
@@ -0,0 +1,91 @@
+---
+id: adaptors-spark
+title: Pulsar adaptor for Apache Spark
+sidebar_label: "Apache Spark"
+original_id: adaptors-spark
+---
+
+## Spark Streaming receiver
+The Spark Streaming receiver for Pulsar is a custom receiver that enables Apache [Spark Streaming](https://spark.apache.org/streaming/) to receive raw data from Pulsar.
+
+An application can receive data in [Resilient Distributed Dataset](https://spark.apache.org/docs/latest/programming-guide.html#resilient-distributed-datasets-rdds) (RDD) format via the Spark Streaming receiver and can process it in a variety of ways.
+
+### Prerequisites
+
+To use the receiver, include a dependency for the `pulsar-spark` library in your Java configuration.
+
+#### Maven
+
+If you're using Maven, add this to your `pom.xml`:
+
+```xml
+
+<!-- in your <properties> block -->
+<pulsar.version>@pulsar:version@</pulsar.version>
+
+<!-- in your <dependencies> block -->
+<dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-spark</artifactId>
+ <version>${pulsar.version}</version>
+</dependency>
+
+```
+
+#### Gradle
+
+If you're using Gradle, add this to your `build.gradle` file:
+
+```groovy
+
+def pulsarVersion = "@pulsar:version@"
+
+dependencies {
+ compile group: 'org.apache.pulsar', name: 'pulsar-spark', version: pulsarVersion
+}
+
+```
+
+### Usage
+
+Pass an instance of `SparkStreamingPulsarReceiver` to the `receiverStream` method in `JavaStreamingContext`:
+
+```java
+
+ String serviceUrl = "pulsar://localhost:6650/";
+ String topic = "persistent://public/default/test_src";
+ String subs = "test_sub";
+
+ SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("Pulsar Spark Example");
+
+ JavaStreamingContext jsc = new JavaStreamingContext(sparkConf, Durations.seconds(60));
+
+ ConsumerConfigurationData<byte[]> pulsarConf = new ConsumerConfigurationData();
+
+ Set<String> set = new HashSet();
+ set.add(topic);
+ pulsarConf.setTopicNames(set);
+ pulsarConf.setSubscriptionName(subs);
+
+ SparkStreamingPulsarReceiver pulsarReceiver = new SparkStreamingPulsarReceiver(
+ serviceUrl,
+ pulsarConf,
+ new AuthenticationDisabled());
+
+ JavaReceiverInputDStream<byte[]> lineDStream = jsc.receiverStream(pulsarReceiver);
+
+```
+
+For a complete example, click [here](https://github.com/apache/pulsar-adapters/blob/master/examples/spark/src/main/java/org/apache/spark/streaming/receiver/example/SparkStreamingPulsarReceiverExample.java). In this example, the number of messages that contain the string "Pulsar" in received messages is counted.
+
+Note that if needed, other Pulsar authentication classes can be used. For example, in order to use a token during authentication the following parameters for the `SparkStreamingPulsarReceiver` constructor can be set:
+
+```java
+
+SparkStreamingPulsarReceiver pulsarReceiver = new SparkStreamingPulsarReceiver(
+ serviceUrl,
+ pulsarConf,
+ new AuthenticationToken("token:<secret-JWT-token>"));
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/adaptors-storm.md b/site2/website-next/versioned_docs/version-2.4.1/adaptors-storm.md
new file mode 100644
index 0000000..76d5071
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/adaptors-storm.md
@@ -0,0 +1,96 @@
+---
+id: adaptors-storm
+title: Pulsar adaptor for Apache Storm
+sidebar_label: "Apache Storm"
+original_id: adaptors-storm
+---
+
+Pulsar Storm is an adaptor for integrating with [Apache Storm](http://storm.apache.org/) topologies. It provides core Storm implementations for sending and receiving data.
+
+An application can inject data into a Storm topology via a generic Pulsar spout, as well as consume data from a Storm topology via a generic Pulsar bolt.
+
+## Using the Pulsar Storm Adaptor
+
+Include dependency for Pulsar Storm Adaptor:
+
+```xml
+
+<dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-storm</artifactId>
+ <version>${pulsar.version}</version>
+</dependency>
+
+```
+
+## Pulsar Spout
+
+The Pulsar Spout allows for the data published on a topic to be consumed by a Storm topology. It emits a Storm tuple based on the message received and the `MessageToValuesMapper` provided by the client.
+
+The tuples that fail to be processed by the downstream bolts will be re-injected by the spout with an exponential backoff, within a configurable timeout (the default is 60 seconds) or a configurable number of retries, whichever comes first, after which it is acknowledged by the consumer. Here's an example construction of a spout:
+
+```java
+
+MessageToValuesMapper messageToValuesMapper = new MessageToValuesMapper() {
+
+ @Override
+ public Values toValues(Message msg) {
+ return new Values(new String(msg.getData()));
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ // declare the output fields
+ declarer.declare(new Fields("string"));
+ }
+};
+
+// Configure a Pulsar Spout
+PulsarSpoutConfiguration spoutConf = new PulsarSpoutConfiguration();
+spoutConf.setServiceUrl("pulsar://broker.messaging.usw.example.com:6650");
+spoutConf.setTopic("persistent://my-property/usw/my-ns/my-topic1");
+spoutConf.setSubscriptionName("my-subscriber-name1");
+spoutConf.setMessageToValuesMapper(messageToValuesMapper);
+
+// Create a Pulsar Spout
+PulsarSpout spout = new PulsarSpout(spoutConf);
+
+```
+
+For a complete example, click [here](https://github.com/apache/pulsar-adapters/blob/master/pulsar-storm/src/test/java/org/apache/pulsar/storm/PulsarSpoutTest.java).
+
+## Pulsar Bolt
+
+The Pulsar bolt allows data in a Storm topology to be published on a topic. It publishes messages based on the Storm tuple received and the `TupleToMessageMapper` provided by the client.
+
+A partitioned topic can also be used to publish messages on different topics. In the implementation of the `TupleToMessageMapper`, a "key" will need to be provided in the message which will send the messages with the same key to the same topic. Here's an example bolt:
+
+```java
+
+TupleToMessageMapper tupleToMessageMapper = new TupleToMessageMapper() {
+
+ @Override
+ public TypedMessageBuilder<byte[]> toMessage(TypedMessageBuilder<byte[]> msgBuilder, Tuple tuple) {
+ String receivedMessage = tuple.getString(0);
+ // message processing
+ String processedMsg = receivedMessage + "-processed";
+ return msgBuilder.value(processedMsg.getBytes());
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ // declare the output fields
+ }
+};
+
+// Configure a Pulsar Bolt
+PulsarBoltConfiguration boltConf = new PulsarBoltConfiguration();
+boltConf.setServiceUrl("pulsar://broker.messaging.usw.example.com:6650");
+boltConf.setTopic("persistent://my-property/usw/my-ns/my-topic2");
+boltConf.setTupleToMessageMapper(tupleToMessageMapper);
+
+// Create a Pulsar Bolt
+PulsarBolt bolt = new PulsarBolt(boltConf);
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-brokers.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-brokers.md
new file mode 100644
index 0000000..dbac453
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-brokers.md
@@ -0,0 +1,272 @@
+---
+id: admin-api-brokers
+title: Managing Brokers
+sidebar_label: "Brokers"
+original_id: admin-api-brokers
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+> **Important**
+>
+> This page only shows **some frequently used operations**.
+>
+> - For the latest and complete information about `Pulsar admin`, including commands, flags, descriptions, and more information, see [Pulsar admin doc](https://pulsar.apache.org/tools/pulsar-admin/).
+>
+> - For the latest and complete information about `REST API`, including parameters, responses, samples, and more, see {@inject: rest:REST:/} API doc.
+>
+> - For the latest and complete information about `Java admin API`, including classes, methods, descriptions, and more, see [Java admin API doc](https://pulsar.apache.org/api/admin/).
+
+Pulsar brokers consist of two components:
+
+1. An HTTP server exposing a {@inject: rest:REST:/} interface administration and [topic](reference-terminology.md#topic) lookup.
+2. A dispatcher that handles all Pulsar [message](reference-terminology.md#message) transfers.
+
+[Brokers](reference-terminology.md#broker) can be managed via:
+
+* The [`brokers`](reference-pulsar-admin.md#brokers) command of the [`pulsar-admin`](reference-pulsar-admin) tool
+* The `/admin/v2/brokers` endpoint of the admin {@inject: rest:REST:/} API
+* The `brokers` method of the {@inject: javadoc:PulsarAdmin:/admin/org/apache/pulsar/client/admin/PulsarAdmin.html} object in the [Java API](client-libraries-java)
+
+In addition to being configurable when you start them up, brokers can also be [dynamically configured](#dynamic-broker-configuration).
+
+> See the [Configuration](reference-configuration.md#broker) page for a full listing of broker-specific configuration parameters.
+
+## Brokers resources
+
+### List active brokers
+
+Fetch all available active brokers that are serving traffic.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+```shell
+
+$ pulsar-admin brokers list use
+
+```
+
+```
+
+broker1.use.org.com:8080
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/brokers/:cluster|operation/getActiveBrokers?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.brokers().getActiveBrokers(clusterName)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Get the information of the leader broker
+
+Fetch the information of the leader broker, for example, the service url.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+```shell
+
+$ pulsar-admin brokers leader-broker
+
+```
+
+```
+
+BrokerInfo(serviceUrl=broker1.use.org.com:8080)
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/brokers/leaderBroker?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.brokers().getLeaderBroker()
+
+```
+
+For the detail of the code above, see [here](https://github.com/apache/pulsar/blob/master/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BrokersImpl.java#L80)
+
+</TabItem>
+
+</Tabs>
+
+#### list of namespaces owned by a given broker
+
+It finds all namespaces which are owned and served by a given broker.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+```shell
+
+$ pulsar-admin brokers namespaces use \
+ --url broker1.use.org.com:8080
+
+```
+
+```json
+
+{
+ "my-property/use/my-ns/0x00000000_0xffffffff": {
+ "broker_assignment": "shared",
+ "is_controlled": false,
+ "is_active": true
+ }
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/brokers/:cluster/:broker/ownedNamespaces|operation/getOwnedNamespaes?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.brokers().getOwnedNamespaces(cluster,brokerUrl);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Dynamic broker configuration
+
+One way to configure a Pulsar [broker](reference-terminology.md#broker) is to supply a [configuration](reference-configuration.md#broker) when the broker is [started up](reference-cli-tools.md#pulsar-broker).
+
+But since all broker configuration in Pulsar is stored in ZooKeeper, configuration values can also be dynamically updated *while the broker is running*. When you update broker configuration dynamically, ZooKeeper will notify the broker of the change and the broker will then override any existing configuration values.
+
+* The [`brokers`](reference-pulsar-admin.md#brokers) command for the [`pulsar-admin`](reference-pulsar-admin) tool has a variety of subcommands that enable you to manipulate a broker's configuration dynamically, enabling you to [update config values](#update-dynamic-configuration) and more.
+* In the Pulsar admin {@inject: rest:REST:/} API, dynamic configuration is managed through the `/admin/v2/brokers/configuration` endpoint.
+
+### Update dynamic configuration
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+The [`update-dynamic-config`](reference-pulsar-admin.md#brokers-update-dynamic-config) subcommand will update existing configuration. It takes two arguments: the name of the parameter and the new value using the `config` and `value` flag respectively. Here's an example for the [`brokerShutdownTimeoutMs`](reference-configuration.md#broker-brokerShutdownTimeoutMs) parameter:
+
+```shell
+
+$ pulsar-admin brokers update-dynamic-config --config brokerShutdownTimeoutMs --value 100
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|POST|/admin/v2/brokers/configuration/:configName/:configValue|operation/updateDynamicConfiguration?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.brokers().updateDynamicConfiguration(configName, configValue);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### List updated values
+
+Fetch a list of all potentially updatable configuration parameters.
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+```shell
+
+$ pulsar-admin brokers list-dynamic-config
+brokerShutdownTimeoutMs
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/brokers/configuration|operation/getDynamicConfigurationName?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.brokers().getDynamicConfigurationNames();
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### List all
+
+Fetch a list of all parameters that have been dynamically updated.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+```shell
+
+$ pulsar-admin brokers get-all-dynamic-config
+brokerShutdownTimeoutMs:100
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/brokers/configuration/values|operation/getAllDynamicConfigurations?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.brokers().getAllDynamicConfigurations();
+
+```
+
+</TabItem>
+
+</Tabs>
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-clusters.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-clusters.md
new file mode 100644
index 0000000..972c7e1
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-clusters.md
@@ -0,0 +1,304 @@
+---
+id: admin-api-clusters
+title: Managing Clusters
+sidebar_label: "Clusters"
+original_id: admin-api-clusters
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+> **Important**
+>
+> This page only shows **some frequently used operations**.
+>
+> - For the latest and complete information about `Pulsar admin`, including commands, flags, descriptions, and more information, see [Pulsar admin doc](https://pulsar.apache.org/tools/pulsar-admin/).
+>
+> - For the latest and complete information about `REST API`, including parameters, responses, samples, and more, see {@inject: rest:REST:/} API doc.
+>
+> - For the latest and complete information about `Java admin API`, including classes, methods, descriptions, and more, see [Java admin API doc](https://pulsar.apache.org/api/admin/).
+
+Pulsar clusters consist of one or more Pulsar [brokers](reference-terminology.md#broker), one or more [BookKeeper](reference-terminology.md#bookkeeper)
+servers (aka [bookies](reference-terminology.md#bookie)), and a [ZooKeeper](https://zookeeper.apache.org) cluster that provides configuration and coordination management.
+
+Clusters can be managed via:
+
+* The [`clusters`](reference-pulsar-admin.md#clusters) command of the [`pulsar-admin`](reference-pulsar-admin) tool
+* The `/admin/v2/clusters` endpoint of the admin {@inject: rest:REST:/} API
+* The `clusters` method of the {@inject: javadoc:PulsarAdmin:/admin/org/apache/pulsar/client/admin/PulsarAdmin} object in the [Java API](client-libraries-java)
+
+## Clusters resources
+
+### Provision
+
+New clusters can be provisioned using the admin interface.
+
+> Please note that this operation requires superuser privileges.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+You can provision a new cluster using the [`create`](reference-pulsar-admin.md#clusters-create) subcommand. Here's an example:
+
+```shell
+
+$ pulsar-admin clusters create cluster-1 \
+ --url http://my-cluster.org.com:8080 \
+ --broker-url pulsar://my-cluster.org.com:6650
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|PUT|/admin/v2/clusters/:cluster|operation/createCluster?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+ClusterData clusterData = new ClusterData(
+ serviceUrl,
+ serviceUrlTls,
+ brokerServiceUrl,
+ brokerServiceUrlTls
+);
+admin.clusters().createCluster(clusterName, clusterData);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Initialize cluster metadata
+
+When provision a new cluster, you need to initialize that cluster's [metadata](concepts-architecture-overview.md#metadata-store). When initializing cluster metadata, you need to specify all of the following:
+
+* The name of the cluster
+* The local ZooKeeper connection string for the cluster
+* The configuration store connection string for the entire instance
+* The web service URL for the cluster
+* A broker service URL enabling interaction with the [brokers](reference-terminology.md#broker) in the cluster
+
+You must initialize cluster metadata *before* starting up any [brokers](admin-api-brokers) that will belong to the cluster.
+
+> **No cluster metadata initialization through the REST API or the Java admin API**
+>
+> Unlike most other admin functions in Pulsar, cluster metadata initialization cannot be performed via the admin REST API
+> or the admin Java client, as metadata initialization involves communicating with ZooKeeper directly.
+> Instead, you can use the [`pulsar`](reference-cli-tools.md#pulsar) CLI tool, in particular
+> the [`initialize-cluster-metadata`](reference-cli-tools.md#pulsar-initialize-cluster-metadata) command.
+
+Here's an example cluster metadata initialization command:
+
+```shell
+
+bin/pulsar initialize-cluster-metadata \
+ --cluster us-west \
+ --zookeeper zk1.us-west.example.com:2181 \
+ --configuration-store zk1.us-west.example.com:2184 \
+ --web-service-url http://pulsar.us-west.example.com:8080/ \
+ --web-service-url-tls https://pulsar.us-west.example.com:8443/ \
+ --broker-service-url pulsar://pulsar.us-west.example.com:6650/ \
+ --broker-service-url-tls pulsar+ssl://pulsar.us-west.example.com:6651/
+
+```
+
+You'll need to use `--*-tls` flags only if you're using [TLS authentication](security-tls-authentication) in your instance.
+
+### Get configuration
+
+You can fetch the [configuration](reference-configuration) for an existing cluster at any time.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`get`](reference-pulsar-admin.md#clusters-get) subcommand and specify the name of the cluster. Here's an example:
+
+```shell
+
+$ pulsar-admin clusters get cluster-1
+{
+ "serviceUrl": "http://my-cluster.org.com:8080/",
+ "serviceUrlTls": null,
+ "brokerServiceUrl": "pulsar://my-cluster.org.com:6650/",
+ "brokerServiceUrlTls": null
+ "peerClusterNames": null
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/clusters/:cluster|operation/getCluster?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.clusters().getCluster(clusterName);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Update
+
+You can update the configuration for an existing cluster at any time.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`update`](reference-pulsar-admin.md#clusters-update) subcommand and specify new configuration values using flags.
+
+```shell
+
+$ pulsar-admin clusters update cluster-1 \
+ --url http://my-cluster.org.com:4081 \
+ --broker-url pulsar://my-cluster.org.com:3350
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|POST|/admin/v2/clusters/:cluster|operation/updateCluster?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+ClusterData clusterData = new ClusterData(
+ serviceUrl,
+ serviceUrlTls,
+ brokerServiceUrl,
+ brokerServiceUrlTls
+);
+admin.clusters().updateCluster(clusterName, clusterData);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Delete
+
+Clusters can be deleted from a Pulsar [instance](reference-terminology.md#instance).
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`delete`](reference-pulsar-admin.md#clusters-delete) subcommand and specify the name of the cluster.
+
+```
+
+$ pulsar-admin clusters delete cluster-1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|DELETE|/admin/v2/clusters/:cluster|operation/deleteCluster?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.clusters().deleteCluster(clusterName);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### List
+
+You can fetch a list of all clusters in a Pulsar [instance](reference-terminology.md#instance).
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`list`](reference-pulsar-admin.md#clusters-list) subcommand.
+
+```shell
+
+$ pulsar-admin clusters list
+cluster-1
+cluster-2
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/clusters|operation/getClusters?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.clusters().getClusters();
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Update peer-cluster data
+
+Peer clusters can be configured for a given cluster in a Pulsar [instance](reference-terminology.md#instance).
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`update-peer-clusters`](reference-pulsar-admin.md#clusters-update-peer-clusters) subcommand and specify the list of peer-cluster names.
+
+```
+
+$ pulsar-admin update-peer-clusters cluster-1 --peer-clusters cluster-2
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|POST|/admin/v2/clusters/:cluster/peers|operation/setPeerClusterNames?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.clusters().updatePeerClusterNames(clusterName, peerClusterList);
+
+```
+
+</TabItem>
+
+</Tabs>
\ No newline at end of file
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-functions.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-functions.md
new file mode 100644
index 0000000..6f3cacf
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-functions.md
@@ -0,0 +1,619 @@
+---
+id: admin-api-functions
+title: Managing functions
+sidebar_label: "Functions"
+original_id: admin-api-functions
+---
+
+**Pulsar Functions** are lightweight compute processes that
+
+* consume messages from one or more Pulsar topics
+* apply a user-supplied processing logic to each message
+* publish the results of the computation to another topic
+
+Functions can be managed via the following methods.
+
+Method | Description
+---|---
+**Admin CLI** | The [`functions`](reference-pulsar-admin.md#functions) command of the [`pulsar-admin`](reference-pulsar-admin) tool.
+**REST API** |The `/admin/v3/functions` endpoint of the admin {@inject: rest:REST:/} API.
+**Java Admin API**| The `functions` method of the {@inject: javadoc:PulsarAdmin:/admin/org/apache/pulsar/client/admin/PulsarAdmin} object in the [Java API](client-libraries-java).
+
+## Function resources
+
+You can perform the following operations on functions.
+
+### Create a function
+
+You can create a Pulsar function in cluster mode (deploy it on a Pulsar cluster) using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`create`](reference-pulsar-admin.md#functions-create) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions create \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --inputs test-input-topic \
+ --output persistent://public/default/test-output-topic \
+ --classname org.apache.pulsar.functions.api.examples.ExclamationFunction \
+ --jar /examples/api-examples.jar
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+FunctionConfig functionConfig = new FunctionConfig();
+functionConfig.setTenant(tenant);
+functionConfig.setNamespace(namespace);
+functionConfig.setName(functionName);
+functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+functionConfig.setParallelism(1);
+functionConfig.setClassName("org.apache.pulsar.functions.api.examples.ExclamationFunction");
+functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE);
+functionConfig.setTopicsPattern(sourceTopicPattern);
+functionConfig.setSubName(subscriptionName);
+functionConfig.setAutoAck(true);
+functionConfig.setOutput(sinkTopic);
+admin.functions().createFunction(functionConfig, fileName);
+
+```
+
+### Update a function
+
+You can update a Pulsar function that has been deployed to a Pulsar cluster using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`update`](reference-pulsar-admin.md#functions-update) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions update \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --output persistent://public/default/update-output-topic \
+ # other options
+
+```
+
+#### REST Admin API
+
+{@inject: endpoint|PUT|/admin/v3/functions/:tenant/:namespace/functionName?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+FunctionConfig functionConfig = new FunctionConfig();
+functionConfig.setTenant(tenant);
+functionConfig.setNamespace(namespace);
+functionConfig.setName(functionName);
+functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+functionConfig.setParallelism(1);
+functionConfig.setClassName("org.apache.pulsar.functions.api.examples.ExclamationFunction");
+UpdateOptions updateOptions = new UpdateOptions();
+updateOptions.setUpdateAuthData(updateAuthData);
+admin.functions().updateFunction(functionConfig, userCodeFile, updateOptions);
+
+```
+
+### Start an instance of a function
+
+You can start a stopped function instance with `instance-id` using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`start`](reference-pulsar-admin.md#functions-start) subcommand.
+
+```shell
+
+$ pulsar-admin functions start \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --instance-id 1
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/:instanceId/start?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().startFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
+
+```
+
+### Start all instances of a function
+
+You can start all stopped function instances using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`start`](reference-pulsar-admin.md#functions-start) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions start \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/start?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+admin.functions().startFunction(tenant, namespace, functionName);
+
+```
+
+### Stop an instance of a function
+
+You can stop a function instance with `instance-id` using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`stop`](reference-pulsar-admin.md#functions-stop) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions stop \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --instance-id 1
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/:instanceId/stop?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().stopFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
+
+```
+
+### Stop all instances of a function
+
+You can stop all function instances using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`stop`](reference-pulsar-admin.md#functions-stop) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions stop \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/stop?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().stopFunction(tenant, namespace, functionName);
+
+```
+
+### Restart an instance of a function
+
+Restart a function instance with `instance-id` using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`restart`](reference-pulsar-admin.md#functions-restart) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions restart \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --instance-id 1
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/:instanceId/restart?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().restartFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
+
+```
+
+### Restart all instances of a function
+
+You can restart all function instances using Admin CLI, REST API or Java admin API.
+
+#### Admin CLI
+
+Use the [`restart`](reference-pulsar-admin.md#functions-restart) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions restart \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/restart?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().restartFunction(tenant, namespace, functionName);
+
+```
+
+### List all functions
+
+You can list all Pulsar functions running under a specific tenant and namespace using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`list`](reference-pulsar-admin.md#functions-list) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions list \
+ --tenant public \
+ --namespace default
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().getFunctions(tenant, namespace);
+
+```
+
+### Delete a function
+
+You can delete a Pulsar function that is running on a Pulsar cluster using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`delete`](reference-pulsar-admin.md#functions-delete) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions delete \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions)
+
+```
+
+#### REST API
+
+{@inject: endpoint|DELETE|/admin/v3/functions/:tenant/:namespace/functionName?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().deleteFunction(tenant, namespace, functionName);
+
+```
+
+### Get info about a function
+
+You can get information about a Pulsar function currently running in cluster mode using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`get`](reference-pulsar-admin.md#functions-get) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions get \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions)
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/functionName?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().getFunction(tenant, namespace, functionName);
+
+```
+
+### Get status of an instance of a function
+
+You can get the current status of a Pulsar function instance with `instance-id` using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`status`](reference-pulsar-admin.md#functions-status) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions status \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --instance-id 1
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/functionName/:instanceId/status?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().getFunctionStatus(tenant, namespace, functionName, Integer.parseInt(instanceId));
+
+```
+
+### Get status of all instances of a function
+
+You can get the current status of a Pulsar function instance using Admin CLI, REST API or Java Admin API.
+
+#### Admin CLI
+
+Use the [`status`](reference-pulsar-admin.md#functions-status) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions status \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions)
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/functionName/status?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().getFunctionStatus(tenant, namespace, functionName);
+
+```
+
+### Get stats of an instance of a function
+
+You can get the current stats of a Pulsar Function instance with `instance-id` using Admin CLI, REST API or Java admin API.
+
+#### Admin CLI
+
+Use the [`stats`](reference-pulsar-admin.md#functions-stats) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions stats \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --instance-id 1
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/functionName/:instanceId/stats?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().getFunctionStats(tenant, namespace, functionName, Integer.parseInt(instanceId));
+
+```
+
+### Get stats of all instances of a function
+
+You can get the current stats of a Pulsar function using Admin CLI, REST API or Java admin API.
+
+#### Admin CLI
+
+Use the [`stats`](reference-pulsar-admin.md#functions-stats) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions stats \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions)
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/functionName/stats?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().getFunctionStats(tenant, namespace, functionName);
+
+```
+
+### Trigger a function
+
+You can trigger a specified Pulsar function with a supplied value using Admin CLI, REST API or Java admin API.
+
+#### Admin CLI
+
+Use the [`trigger`](reference-pulsar-admin.md#functions-trigger) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions trigger \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --topic (the name of input topic) \
+ --trigger-value \"hello pulsar\"
+ # or --trigger-file (the path of trigger file)
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/trigger?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+admin.functions().triggerFunction(tenant, namespace, functionName, topic, triggerValue, triggerFile);
+
+```
+
+### Put state associated with a function
+
+You can put the state associated with a Pulsar function using Admin CLI, REST API or Java admin API.
+
+#### Admin CLI
+
+Use the [`putstate`](reference-pulsar-admin.md#functions-putstate) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions putstate \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --state "{\"key\":\"pulsar\", \"stringValue\":\"hello pulsar\"}"
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v3/functions/:tenant/:namespace/functionName/state/:key?version=@pulsar:version_number@}
+
+#### Java Admin API
+
+```java
+
+TypeReference<FunctionState> typeRef = new TypeReference<FunctionState>() {};
+FunctionState stateRepr = ObjectMapperFactory.getThreadLocal().readValue(state, typeRef);
+admin.functions().putFunctionState(tenant, namespace, functionName, stateRepr);
+
+```
+
+### Fetch state associated with a function
+
+You can fetch the current state associated with a Pulsar function using Admin CLI, REST API or Java admin API.
+
+#### Admin CLI
+
+Use the [`querystate`](reference-pulsar-admin.md#functions-querystate) subcommand.
+
+**Example**
+
+```shell
+
+$ pulsar-admin functions querystate \
+ --tenant public \
+ --namespace default \
+ --name (the name of Pulsar Functions) \
+ --key (the key of state)
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v3/functions/:tenant/:namespace/functionName/state/:key?version=@pulsar:version_number@}
+
+#### Java Admin CLI
+
+```java
+
+admin.functions().getFunctionState(tenant, namespace, functionName, key);
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-namespaces.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-namespaces.md
new file mode 100644
index 0000000..216cb6f
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-namespaces.md
@@ -0,0 +1,1374 @@
+---
+id: admin-api-namespaces
+title: Managing Namespaces
+sidebar_label: "Namespaces"
+original_id: admin-api-namespaces
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+> **Important**
+>
+> This page only shows **some frequently used operations**.
+>
+> - For the latest and complete information about `Pulsar admin`, including commands, flags, descriptions, and more information, see [Pulsar admin doc](https://pulsar.apache.org/tools/pulsar-admin/).
+>
+> - For the latest and complete information about `REST API`, including parameters, responses, samples, and more, see {@inject: rest:REST:/} API doc.
+>
+> - For the latest and complete information about `Java admin API`, including classes, methods, descriptions, and more, see [Java admin API doc](https://pulsar.apache.org/api/admin/).
+
+Pulsar [namespaces](reference-terminology.md#namespace) are logical groupings of [topics](reference-terminology.md#topic).
+
+Namespaces can be managed via:
+
+* The [`namespaces`](reference-pulsar-admin.md#clusters) command of the [`pulsar-admin`](reference-pulsar-admin) tool
+* The `/admin/v2/namespaces` endpoint of the admin {@inject: rest:REST:/} API
+* The `namespaces` method of the {@inject: javadoc:PulsarAdmin:/admin/org/apache/pulsar/client/admin/PulsarAdmin} object in the [Java API](client-libraries-java)
+
+## Namespaces resources
+
+### Create namespaces
+
+You can create new namespaces under a given [tenant](reference-terminology.md#tenant).
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`create`](reference-pulsar-admin.md#namespaces-create) subcommand and specify the namespace by name:
+
+```shell
+
+$ pulsar-admin namespaces create test-tenant/test-namespace
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|PUT|/admin/v2/namespaces/:tenant/:namespace|operation/createNamespace?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().createNamespace(namespace);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Get policies
+
+You can fetch the current policies associated with a namespace at any time.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`policies`](reference-pulsar-admin.md#namespaces-policies) subcommand and specify the namespace:
+
+```shell
+
+$ pulsar-admin namespaces policies test-tenant/test-namespace
+{
+ "auth_policies": {
+ "namespace_auth": {},
+ "destination_auth": {}
+ },
+ "replication_clusters": [],
+ "bundles_activated": true,
+ "bundles": {
+ "boundaries": [
+ "0x00000000",
+ "0xffffffff"
+ ],
+ "numBundles": 1
+ },
+ "backlog_quota_map": {},
+ "persistence": null,
+ "latency_stats_sample_rate": {},
+ "message_ttl_in_seconds": 0,
+ "retention_policies": null,
+ "deleted": false
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace|operation/getPolicies?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getPolicies(namespace);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### List namespaces
+
+You can list all namespaces within a given Pulsar [tenant](reference-terminology.md#tenant).
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`list`](reference-pulsar-admin.md#namespaces-list) subcommand and specify the tenant:
+
+```shell
+
+$ pulsar-admin namespaces list test-tenant
+test-tenant/ns1
+test-tenant/ns2
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant|operation/getTenantNamespaces?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getNamespaces(tenant);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Delete namespaces
+
+You can delete existing namespaces from a tenant.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`delete`](reference-pulsar-admin.md#namespaces-delete) subcommand and specify the namespace:
+
+```shell
+
+$ pulsar-admin namespaces delete test-tenant/ns1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|DELETE|/admin/v2/namespaces/:tenant/:namespace|operation/deleteNamespace?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().deleteNamespace(namespace);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure replication clusters
+
+#### Set replication cluster
+
+It sets replication clusters for a namespace, so Pulsar can internally replicate publish message from one colo to another colo.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-clusters test-tenant/ns1 \
+ --clusters cl1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/replication|operation/setNamespaceReplicationClusters?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setNamespaceReplicationClusters(namespace, clusters);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get replication cluster
+
+It gives a list of replication clusters for a given namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-clusters test-tenant/cl1/ns1
+
+```
+
+```
+
+cl2
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/replication|operation/getNamespaceReplicationClusters?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getNamespaceReplicationClusters(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure backlog quota policies
+
+#### Set backlog quota policies
+
+Backlog quota helps the broker to restrict bandwidth/storage of a namespace once it reaches a certain threshold limit. Admin can set the limit and take corresponding action after the limit is reached.
+
+ 1. producer_request_hold: broker will hold and not persist produce request payload
+
+ 2. producer_exception: broker disconnects with the client by giving an exception.
+
+ 3. consumer_backlog_eviction: broker will start discarding backlog messages
+
+ Backlog quota restriction can be taken care by defining restriction of backlog-quota-type: destination_storage
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-backlog-quota --limit 10G --limitTime 36000 --policy producer_request_hold test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/backlogQuota|operation/setBacklogQuota?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setBacklogQuota(namespace, new BacklogQuota(limit, limitTime, policy))
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get backlog quota policies
+
+It shows a configured backlog quota for a given namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-backlog-quotas test-tenant/ns1
+
+```
+
+```json
+
+{
+ "destination_storage": {
+ "limit": 10,
+ "policy": "producer_request_hold"
+ }
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/backlogQuotaMap|operation/getBacklogQuotaMap?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getBacklogQuotaMap(namespace);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Remove backlog quota policies
+
+It removes backlog quota policies for a given namespace
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces remove-backlog-quota test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|DELETE|/admin/v2/namespaces/:tenant/:namespace/backlogQuota|operation/removeBacklogQuota?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().removeBacklogQuota(namespace, backlogQuotaType)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure persistence policies
+
+#### Set persistence policies
+
+Persistence policies allow to configure persistency-level for all topic messages under a given namespace.
+
+ - Bookkeeper-ack-quorum: Number of acks (guaranteed copies) to wait for each entry, default: 0
+
+ - Bookkeeper-ensemble: Number of bookies to use for a topic, default: 0
+
+ - Bookkeeper-write-quorum: How many writes to make of each entry, default: 0
+
+ - Ml-mark-delete-max-rate: Throttling rate of mark-delete operation (0 means no throttle), default: 0.0
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-persistence --bookkeeper-ack-quorum 2 --bookkeeper-ensemble 3 --bookkeeper-write-quorum 2 --ml-mark-delete-max-rate 0 test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/persistence|operation/setPersistence?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setPersistence(namespace,new PersistencePolicies(bookkeeperEnsemble, bookkeeperWriteQuorum,bookkeeperAckQuorum,managedLedgerMaxMarkDeleteRate))
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get persistence policies
+
+It shows the configured persistence policies of a given namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-persistence test-tenant/ns1
+
+```
+
+```json
+
+{
+ "bookkeeperEnsemble": 3,
+ "bookkeeperWriteQuorum": 2,
+ "bookkeeperAckQuorum": 2,
+ "managedLedgerMaxMarkDeleteRate": 0
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/persistence|operation/getPersistence?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getPersistence(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure namespace bundles
+
+#### Unload namespace bundles
+
+The namespace bundle is a virtual group of topics which belong to the same namespace. If the broker gets overloaded with the number of bundles, this command can help unload a bundle from that broker, so it can be served by some other less-loaded brokers. The namespace bundle ID ranges from 0x00000000 to 0xffffffff.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces unload --bundle 0x00000000_0xffffffff test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|PUT|/admin/v2/namespaces/:tenant/:namespace/{bundle}/unload|operation/unloadNamespaceBundle?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().unloadNamespaceBundle(namespace, bundle)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Split namespace bundles
+
+Each namespace bundle can contain multiple topics and each bundle can be served by only one broker.
+If a single bundle is creating an excessive load on a broker, an admin splits the bundle using this command permitting one or more of the new bundles to be unloaded thus spreading the load across the brokers.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces split-bundle --bundle 0x00000000_0xffffffff test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|PUT|/admin/v2/namespaces/:tenant/:namespace/{bundle}/split|operation/splitNamespaceBundle?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().splitNamespaceBundle(namespace, bundle)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure message TTL
+
+#### Set message-ttl
+
+It configures message’s time to live (in seconds) duration.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-message-ttl --messageTTL 100 test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/messageTTL|operation/setNamespaceMessageTTL?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setNamespaceMessageTTL(namespace, messageTTL)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get message-ttl
+
+It gives a message ttl of configured namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-message-ttl test-tenant/ns1
+
+```
+
+```
+
+100
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/messageTTL|operation/getNamespaceMessageTTL?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getNamespaceMessageTTL(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Remove message-ttl
+
+Remove a message TTL of the configured namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces remove-message-ttl test-tenant/ns1
+
+```
+
+```
+
+100
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|DELETE|/admin/v2/namespaces/:tenant/:namespace/messageTTL|operation/removeNamespaceMessageTTL?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().removeNamespaceMessageTTL(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+
+### Clear backlog
+
+#### Clear namespace backlog
+
+It clears all message backlog for all the topics that belong to a specific namespace. You can also clear backlog for a specific subscription as well.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces clear-backlog --sub my-subscription test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/clearBacklog|operation/clearNamespaceBacklogForSubscription?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().clearNamespaceBacklogForSubscription(namespace, subscription)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Clear bundle backlog
+
+It clears all message backlog for all the topics that belong to a specific NamespaceBundle. You can also clear backlog for a specific subscription as well.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces clear-backlog --bundle 0x00000000_0xffffffff --sub my-subscription test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/{bundle}/clearBacklog|operation?version=@pulsar:version_number@/clearNamespaceBundleBacklogForSubscription}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().clearNamespaceBundleBacklogForSubscription(namespace, bundle, subscription)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure retention
+
+#### Set retention
+
+Each namespace contains multiple topics and the retention size (storage size) of each topic should not exceed a specific threshold or it should be stored for a certain period. This command helps configure the retention size and time of topics in a given namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin set-retention --size 100 --time 10 test-tenant/ns1
+
+```
+
+```
+
+N/A
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/retention|operation/setRetention?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setRetention(namespace, new RetentionPolicies(retentionTimeInMin, retentionSizeInMB))
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get retention
+
+It shows retention information of a given namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-retention test-tenant/ns1
+
+```
+
+```json
+
+{
+ "retentionTimeInMinutes": 10,
+ "retentionSizeInMB": 100
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/retention|operation/getRetention?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getRetention(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure dispatch throttling for topics
+
+#### Set dispatch throttling for topics
+
+It sets message dispatch rate for all the topics under a given namespace.
+The dispatch rate can be restricted by the number of messages per X seconds (`msg-dispatch-rate`) or by the 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.
+
+:::
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-dispatch-rate test-tenant/ns1 \
+ --msg-dispatch-rate 1000 \
+ --byte-dispatch-rate 1048576 \
+ --dispatch-rate-period 1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/dispatchRate|operation/setDispatchRate?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setDispatchRate(namespace, new DispatchRate(1000, 1048576, 1))
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get configured message-rate for topics
+
+It shows configured message-rate for the namespace (topics under this namespace can dispatch this many messages per second)
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-dispatch-rate test-tenant/ns1
+
+```
+
+```json
+
+{
+ "dispatchThrottlingRatePerTopicInMsg" : 1000,
+ "dispatchThrottlingRatePerTopicInByte" : 1048576,
+ "ratePeriodInSecond" : 1
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/dispatchRate|operation/getDispatchRate?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getDispatchRate(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure dispatch throttling for subscription
+
+#### Set dispatch throttling for subscription
+
+It sets message dispatch rate for all the subscription of topics under a given namespace.
+The dispatch rate can be restricted by the number of messages per X seconds (`msg-dispatch-rate`) or by the 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.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-subscription-dispatch-rate test-tenant/ns1 \
+ --msg-dispatch-rate 1000 \
+ --byte-dispatch-rate 1048576 \
+ --dispatch-rate-period 1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/subscriptionDispatchRate|operation/setDispatchRate?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setSubscriptionDispatchRate(namespace, new DispatchRate(1000, 1048576, 1))
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get configured message-rate for subscription
+
+It shows configured message-rate for the namespace (topics under this namespace can dispatch this many messages per second)
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-subscription-dispatch-rate test-tenant/ns1
+
+```
+
+```json
+
+{
+ "dispatchThrottlingRatePerTopicInMsg" : 1000,
+ "dispatchThrottlingRatePerTopicInByte" : 1048576,
+ "ratePeriodInSecond" : 1
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/subscriptionDispatchRate|operation/getDispatchRate?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getSubscriptionDispatchRate(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure dispatch throttling for replicator
+
+#### Set dispatch throttling for replicator
+
+It sets message dispatch rate for all the replicator between replication clusters under a given namespace.
+The dispatch rate can be restricted by the number of messages per X seconds (`msg-dispatch-rate`) or by the 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.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-replicator-dispatch-rate test-tenant/ns1 \
+ --msg-dispatch-rate 1000 \
+ --byte-dispatch-rate 1048576 \
+ --dispatch-rate-period 1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/replicatorDispatchRate|operation/setDispatchRate?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setReplicatorDispatchRate(namespace, new DispatchRate(1000, 1048576, 1))
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Get configured message-rate for replicator
+
+It shows configured message-rate for the namespace (topics under this namespace can dispatch this many messages per second)
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-replicator-dispatch-rate test-tenant/ns1
+
+```
+
+```json
+
+{
+ "dispatchThrottlingRatePerTopicInMsg" : 1000,
+ "dispatchThrottlingRatePerTopicInByte" : 1048576,
+ "ratePeriodInSecond" : 1
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/replicatorDispatchRate|operation/getDispatchRate?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getReplicatorDispatchRate(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Configure deduplication snapshot interval
+
+#### Get deduplication snapshot interval
+
+It shows configured `deduplicationSnapshotInterval` for a namespace (Each topic under the namespace will take a deduplication snapshot according to this interval)
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces get-deduplication-snapshot-interval test-tenant/ns1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/deduplicationSnapshotInterval?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getDeduplicationSnapshotInterval(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Set deduplication snapshot interval
+
+Set configured `deduplicationSnapshotInterval` for a namespace. Each topic under the namespace will take a deduplication snapshot according to this interval.
+`brokerDeduplicationEnabled` must be set to `true` for this property to take effect.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces set-deduplication-snapshot-interval test-tenant/ns1 --interval 1000
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/deduplicationSnapshotInterval?version=@pulsar:version_number@}
+
+```
+
+```json
+
+{
+ "interval": 1000
+}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().setDeduplicationSnapshotInterval(namespace, 1000)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+#### Remove deduplication snapshot interval
+
+Remove configured `deduplicationSnapshotInterval` of a namespace (Each topic under the namespace will take a deduplication snapshot according to this interval)
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```
+
+$ pulsar-admin namespaces remove-deduplication-snapshot-interval test-tenant/ns1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+```
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/deduplicationSnapshotInterval?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().removeDeduplicationSnapshotInterval(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Namespace isolation
+
+You can use the [Pulsar isolation policy](administration-isolation) to allocate resources (broker and bookie) for a namespace.
+
+### Unload namespaces from a broker
+
+You can unload a namespace, or a [namespace bundle](reference-terminology.md#namespace-bundle), from the Pulsar [broker](reference-terminology.md#broker) that is currently responsible for it.
+
+#### pulsar-admin
+
+Use the [`unload`](reference-pulsar-admin.md#unload) subcommand of the [`namespaces`](reference-pulsar-admin.md#namespaces) command.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST","value":"REST"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+```shell
+
+$ pulsar-admin namespaces unload my-tenant/my-ns
+
+```
+
+</TabItem>
+<TabItem value="REST">
+
+```
+
+{@inject: endpoint|PUT|/admin/v2/namespaces/:tenant/:namespace/unload|operation/unloadNamespace?version=@pulsar:version_number@}
+
+```
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().unload(namespace)
+
+```
+
+</TabItem>
+
+</Tabs>
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-non-persistent-topics.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-non-persistent-topics.md
new file mode 100644
index 0000000..ecc7d3e
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-non-persistent-topics.md
@@ -0,0 +1,286 @@
+---
+id: admin-api-non-persistent-topics
+title: Managing non-persistent topics
+sidebar_label: "Non-Persistent topics"
+original_id: admin-api-non-persistent-topics
+---
+
+Non-persistent can be used in applications that only want to consume real time published messages and
+do not need persistent guarantee that can also reduce message-publish latency by removing overhead of
+persisting messages.
+
+In all of the instructions and commands below, the topic name structure is:
+
+```shell
+
+non-persistent://tenant/namespace/topic
+
+```
+
+## Non-persistent topics resources
+
+### Get stats
+
+It shows current statistics of a given non-partitioned topic.
+
+ - **msgRateIn**: The sum of all local and replication publishers' publish rates in messages per second
+
+ - **msgThroughputIn**: Same as above, but in bytes per second instead of messages per second
+
+ - **msgRateOut**: The sum of all local and replication consumers' dispatch rates in messages per second
+
+ - **msgThroughputOut**: Same as above, but in bytes per second instead of messages per second
+
+ - **averageMsgSize**: The average size in bytes of messages published within the last interval
+
+ - **publishers**: The list of all local publishers into the topic. There can be zero or thousands
+
+ - **averageMsgSize**: Average message size in bytes from this publisher within the last interval
+
+ - **producerId**: Internal identifier for this producer on this topic
+
+ - **producerName**: Internal identifier for this producer, generated by the client library
+
+ - **address**: IP address and source port for the connection of this producer
+
+ - **connectedSince**: Timestamp this producer was created or last reconnected
+
+ - **subscriptions**: The list of all local subscriptions to the topic
+
+ - **my-subscription**: The name of this subscription (client defined)
+
+ - **type**: This subscription type
+
+ - **consumers**: The list of connected consumers for this subscription
+
+ - **consumerName**: Internal identifier for this consumer, generated by the client library
+
+ - **availablePermits**: The number of messages this consumer has space for in the client library's listen queue. A value less than 1 means the client library's queue is full and receive() isn't being called. A non-negative value means this consumer is ready to be dispatched messages.
+
+ - **replication**: This section gives the stats for cross-colo replication of this topic
+
+ - **connected**: Whether the outbound replicator is connected
+
+ - **inboundConnection**: The IP and port of the broker in the remote cluster's publisher connection to this broker
+
+ - **inboundConnectedSince**: The TCP connection being used to publish messages to the remote cluster. If there are no local publishers connected, this connection is automatically closed after a minute.
+
+ - **msgDropRate**: for publisher: publish: broker only allows configured number of in flight per connection, and drops all other published messages above the threshold. Broker also drops messages for subscriptions in case of unavailable limit and connection is not writable.
+
+```json
+
+{
+ "msgRateIn": 4641.528542257553,
+ "msgThroughputIn": 44663039.74947473,
+ "msgRateOut": 0,
+ "msgThroughputOut": 0,
+ "averageMsgSize": 1232439.816728665,
+ "storageSize": 135532389160,
+ "msgDropRate" : 0.0,
+ "publishers": [
+ {
+ "msgRateIn": 57.855383881403576,
+ "msgThroughputIn": 558994.7078932219,
+ "averageMsgSize": 613135,
+ "producerId": 0,
+ "producerName": null,
+ "address": null,
+ "connectedSince": null,
+ "msgDropRate" : 0.0
+ }
+ ],
+ "subscriptions": {
+ "my-topic_subscription": {
+ "msgRateOut": 0,
+ "msgThroughputOut": 0,
+ "msgBacklog": 116632,
+ "type": null,
+ "msgRateExpired": 36.98245516804671,
+ "consumers" : [ {
+ "msgRateOut" : 20343.506296021893,
+ "msgThroughputOut" : 2.0979855364233278E7,
+ "msgRateRedeliver" : 0.0,
+ "consumerName" : "fe3c0",
+ "availablePermits" : 950,
+ "unackedMessages" : 0,
+ "blockedConsumerOnUnackedMsgs" : false,
+ "address" : "/10.73.210.249:60578",
+ "connectedSince" : "2017-07-26 15:13:48.026-0700",
+ "clientVersion" : "1.19-incubating-SNAPSHOT"
+ } ],
+ "msgDropRate" : 432.2390921571593
+
+ }
+ },
+ "replication": {}
+}
+
+```
+
+#### pulsar-admin
+
+Topic stats can be fetched using [`stats`](reference-pulsar-admin.md#stats) command.
+
+```shell
+
+$ pulsar-admin non-persistent stats \
+ non-persistent://test-tenant/ns1/tp1 \
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v2/non-persistent/:tenant/:namespace/:topic/stats|operation/getStats?version=@pulsar:version_number@}
+
+
+#### Java
+
+```java
+
+String topic = "non-persistent://my-tenant/my-namespace/my-topic";
+admin.nonPersistentTopics().getStats(topic);
+
+```
+
+### Get internal stats
+
+It shows detailed statistics of a topic.
+
+#### pulsar-admin
+
+Topic internal-stats can be fetched using [`stats-internal`](reference-pulsar-admin.md#stats-internal) command.
+
+```shell
+
+$ pulsar-admin non-persistent stats-internal \
+ non-persistent://test-tenant/ns1/tp1 \
+
+{
+ "entriesAddedCounter" : 48834,
+ "numberOfEntries" : 0,
+ "totalSize" : 0,
+ "cursors" : {
+ "s1" : {
+ "waitingReadOp" : false,
+ "pendingReadOps" : 0,
+ "messagesConsumedCounter" : 0,
+ "cursorLedger" : 0,
+ "cursorLedgerLastEntry" : 0
+ }
+ }
+}
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v2/non-persistent/:tenant/:namespace/:topic/internalStats|operation/getInternalStats?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+String topic = "non-persistent://my-tenant/my-namespace/my-topic";
+admin.nonPersistentTopics().getInternalStats(topic);
+
+```
+
+### Create partitioned topic
+
+Partitioned topics in Pulsar must be explicitly created. When creating a new partitioned topic you need to provide a name for the topic as well as the desired number of partitions.
+
+:::note
+
+By default, after 60 seconds of creation, topics are considered inactive and deleted automatically to prevent from generating trash data.
+To disable this feature, set `brokerDeleteInactiveTopicsEnabled` to `false`.
+To change the frequency of checking inactive topics, set `brokerDeleteInactiveTopicsFrequencySeconds` to your desired value.
+For more information about these two parameters, see [here](reference-configuration.md#broker).
+
+:::
+
+#### pulsar-admin
+
+```shell
+
+$ bin/pulsar-admin non-persistent create-partitioned-topic \
+ non-persistent://my-tenant/my-namespace/my-topic \
+ --partitions 4
+
+```
+
+#### REST API
+
+{@inject: endpoint|PUT|/admin/v2/non-persistent/:tenant/:namespace/:topic/partitions|operation/createPartitionedTopic?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+String topicName = "non-persistent://my-tenant/my-namespace/my-topic";
+int numPartitions = 4;
+admin.nonPersistentTopics().createPartitionedTopic(topicName, numPartitions);
+
+```
+
+### Get metadata
+
+Partitioned topics have metadata associated with them that you can fetch as a JSON object. The following metadata fields are currently available:
+
+Field | Meaning
+:-----|:-------
+`partitions` | The number of partitions into which the topic is divided
+
+#### pulsar-admin
+
+```shell
+
+$ pulsar-admin non-persistent get-partitioned-topic-metadata \
+ non-persistent://my-tenant/my-namespace/my-topic
+{
+ "partitions": 4
+}
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v2/non-persistent/:tenant/:namespace/:topic/partitions|operation/getPartitionedMetadata?version=@pulsar:version_number@}
+
+
+#### Java
+
+```java
+
+String topicName = "non-persistent://my-tenant/my-namespace/my-topic";
+admin.nonPersistentTopics().getPartitionedTopicMetadata(topicName);
+
+```
+
+### Unload topic
+
+It unloads a topic.
+
+#### pulsar-admin
+
+Topic can be unloaded using [`unload`](reference-pulsar-admin.md#unload) command.
+
+```shell
+
+$ pulsar-admin non-persistent unload \
+ non-persistent://test-tenant/ns1/tp1 \
+
+```
+
+#### REST API
+
+{@inject: endpoint|PUT|/admin/v2/non-persistent/:tenant/:namespace/:topic/unload|operation/unloadTopic?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+String topic = "non-persistent://my-tenantmy-namespace/my-topic";
+admin.nonPersistentTopics().unload(topic);
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-overview.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-overview.md
new file mode 100644
index 0000000..b6186cd
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-overview.md
@@ -0,0 +1,97 @@
+---
+id: admin-api-overview
+title: The Pulsar admin interface
+sidebar_label: "Overview"
+original_id: admin-api-overview
+---
+
+The Pulsar admin interface enables you to manage all of the important entities in a Pulsar [instance](reference-terminology.md#instance), such as [tenants](reference-terminology.md#tenant), [topics](reference-terminology.md#topic), and [namespaces](reference-terminology.md#namespace).
+
+You can currently interact with the admin interface via:
+
+- Making HTTP calls against the admin {@inject: rest:REST:/} API provided by Pulsar [brokers](reference-terminology.md#broker). For some restful apis, they might be redirected to topic owner brokers for serving
+ with [`307 Temporary Redirect`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/307), hence the HTTP callers should handle `307 Temporary Redirect`. If you are using `curl`, you should specify `-L`
+ to handle redirections.
+- The `pulsar-admin` CLI tool, which is available in the `bin` folder of your [Pulsar installation](getting-started-standalone):
+
+```shell
+
+$ bin/pulsar-admin
+
+```
+
+Full documentation for this tool can be found in the [Pulsar command-line tools](reference-pulsar-admin) doc.
+
+- A Java client interface.
+
+> #### The REST API is the admin interface
+> Under the hood, both the `pulsar-admin` CLI tool and the Java client both use the REST API. If you’d like to implement your own admin interface client, you should use the REST API as well. Full documentation can be found here.
+
+In this document, examples from each of the three available interfaces will be shown.
+
+## Admin setup
+
+Each of Pulsar's three admin interfaces---the [`pulsar-admin`](reference-pulsar-admin) CLI tool, the [Java admin API](/api/admin), and the {@inject: rest:REST:/} API ---requires some special setup if you have [authentication](security-overview.md#authentication-providers) enabled in your Pulsar [instance](reference-terminology.md#instance).
+
+### pulsar-admin
+
+If you have [authentication](security-overview.md#authentication-providers) enabled, you will need to provide an auth configuration to use the [`pulsar-admin`](reference-pulsar-admin) tool. By default, the configuration for the `pulsar-admin` tool is found in the [`conf/client.conf`](reference-configuration.md#client) file. Here are the available parameters:
+
+|Name|Description|Default|
+|----|-----------|-------|
+|webServiceUrl|The web URL for the cluster.|http://localhost:8080/|
+|brokerServiceUrl|The Pulsar protocol URL for the cluster.|pulsar://localhost:6650/|
+|authPlugin|The authentication plugin.| |
+|authParams|The authentication parameters for the cluster, as a comma-separated string.| |
+|useTls|Whether or not TLS authentication will be enforced in the cluster.|false|
+|tlsAllowInsecureConnection|Accept untrusted TLS certificate from client.|false|
+|tlsTrustCertsFilePath|Path for the trusted TLS certificate file.| |
+
+### REST API
+
+You can find documentation for the REST API exposed by Pulsar [brokers](reference-terminology.md#broker) in this reference {@inject: rest:document:/}.
+
+### Java admin client
+
+To use the Java admin API, instantiate a {@inject: javadoc:PulsarAdmin:/admin/org/apache/pulsar/client/admin/PulsarAdmin} object, specifying a URL for a Pulsar [broker](reference-terminology.md#broker) and a {@inject: javadoc:PulsarAdminBuilder:/admin/org/apache/pulsar/client/admin/PulsarAdminBuilder}. Here's a minimal example using `localhost`:
+
+```java
+
+String url = "http://localhost:8080";
+// Pass auth-plugin class fully-qualified name if Pulsar-security enabled
+String authPluginClassName = "com.org.MyAuthPluginClass";
+// Pass auth-param if auth-plugin class requires it
+String authParams = "param1=value1";
+boolean useTls = false;
+boolean tlsAllowInsecureConnection = false;
+String tlsTrustCertsFilePath = null;
+PulsarAdmin admin = PulsarAdmin.builder()
+.authentication(authPluginClassName,authParams)
+.serviceHttpUrl(url)
+.tlsTrustCertsFilePath(tlsTrustCertsFilePath)
+.allowTlsInsecureConnection(tlsAllowInsecureConnection)
+.build();
+
+```
+
+If you have multiple brokers to use, you can use multi-host like Pulsar service. For example,
+
+```java
+
+String url = "http://localhost:8080,localhost:8081,localhost:8082";
+// Pass auth-plugin class fully-qualified name if Pulsar-security enabled
+String authPluginClassName = "com.org.MyAuthPluginClass";
+// Pass auth-param if auth-plugin class requires it
+String authParams = "param1=value1";
+boolean useTls = false;
+boolean tlsAllowInsecureConnection = false;
+String tlsTrustCertsFilePath = null;
+PulsarAdmin admin = PulsarAdmin.builder()
+.authentication(authPluginClassName,authParams)
+.serviceHttpUrl(url)
+.tlsTrustCertsFilePath(tlsTrustCertsFilePath)
+.allowTlsInsecureConnection(tlsAllowInsecureConnection)
+.build();
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-partitioned-topics.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-partitioned-topics.md
new file mode 100644
index 0000000..ecb9ad8
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-partitioned-topics.md
@@ -0,0 +1,408 @@
+---
+id: admin-api-partitioned-topics
+title: Managing partitioned topics
+sidebar_label: "Partitioned topics"
+original_id: admin-api-partitioned-topics
+---
+
+
+You can use Pulsar's [admin API](admin-api-overview) to create and manage partitioned topics.
+
+In all of the instructions and commands below, the topic name structure is:
+
+```shell
+
+persistent://tenant/namespace/topic
+
+```
+
+## Partitioned topics resources
+
+### Create
+
+Partitioned topics in Pulsar must be explicitly created. When creating a new partitioned topic you
+need to provide a name for the topic as well as the desired number of partitions.
+
+:::note
+
+By default, after 60 seconds of creation, topics are considered inactive and deleted automatically to prevent from generating trash data.
+To disable this feature, set `brokerDeleteInactiveTopicsEnabled` to `false`.
+To change the frequency of checking inactive topics, set `brokerDeleteInactiveTopicsFrequencySeconds` to your desired value.
+For more information about these two parameters, see [here](reference-configuration.md#broker).
+
+:::
+
+#### pulsar-admin
+
+You can create partitioned topics using the [`create-partitioned-topic`](reference-pulsar-admin.md#create-partitioned-topic)
+command and specifying the topic name as an argument and the number of partitions using the `-p` or `--partitions` flag.
+Here's an example:
+
+```shell
+
+$ bin/pulsar-admin topics create-partitioned-topic \
+ persistent://my-tenant/my-namespace/my-topic \
+ --partitions 4
+
+```
+
+#### REST API
+
+{@inject: endpoint|PUT|/admin/v2/:schema/:tenant/:namespace/:topic/partitions|operation/createPartitionedTopic?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+String topicName = "persistent://my-tenant/my-namespace/my-topic";
+int numPartitions = 4;
+admin.persistentTopics().createPartitionedTopic(topicName, numPartitions);
+
+```
+
+## Nonpartitioned topics resources
+
+### Create
+
+Nonpartitioned topics in Pulsar must be explicitly created if allowAutoTopicCreation or createIfMissing is disabled.
+When creating a non-partitioned topic, you need to provide a topic name.
+
+:::note
+
+By default, after 60 seconds of creation, topics are considered inactive and deleted automatically to prevent from generating trash data.
+To disable this feature, set `brokerDeleteInactiveTopicsEnabled` to `false`.
+To change the frequency of checking inactive topics, set `brokerDeleteInactiveTopicsFrequencySeconds` to your desired value.
+For more information about these two parameters, see [here](reference-configuration.md#broker).
+
+:::
+
+#### pulsar-admin
+
+You can create non-partitioned topics using the [`create`](reference-pulsar-admin.md#create)
+command and specifying the topic name as an argument. This is an example command:
+
+```shell
+
+$ bin/pulsar-admin topics create persistent://my-tenant/my-namespace/my-topic
+
+```
+
+#### REST API
+
+{@inject: endpoint|PUT|admin/v2/:schema/:tenant/:namespace/:topic|operation/createNonPartitionedTopic?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+String topicName = "persistent://my-tenant/my-namespace/my-topic";
+admin.topics().createNonPartitionedTopic(topicName);
+
+```
+
+### Get metadata
+
+Partitioned topics have metadata associated with them that you can fetch as a JSON object.
+The following metadata fields are currently available:
+
+Field | Meaning
+:-----|:-------
+`partitions` | The number of partitions into which the topic is divided
+
+#### pulsar-admin
+
+You can see the number of partitions in a partitioned topic using the [`get-partitioned-topic-metadata`](reference-pulsar-admin.md#get-partitioned-topic-metadata)
+subcommand. Here's an example:
+
+```shell
+
+$ pulsar-admin topics get-partitioned-topic-metadata \
+ persistent://my-tenant/my-namespace/my-topic
+{
+ "partitions": 4
+}
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v2/:schema/:tenant/:namespace/:topic/partitions|operation/getPartitionedMetadata?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+String topicName = "persistent://my-tenant/my-namespace/my-topic";
+admin.persistentTopics().getPartitionedTopicMetadata(topicName);
+
+```
+
+### Update
+
+You can update the number of partitions on an existing partitioned topic
+*if* the topic is non-global. To update, the new number of partitions must be greater
+than the existing number.
+
+Decrementing the number of partitions would deleting the topic, which is not supported in Pulsar.
+
+Already created partitioned producers and consumers will automatically find the newly created partitions.
+
+#### pulsar-admin
+
+Partitioned topics can be updated using the [`update-partitioned-topic`](reference-pulsar-admin.md#update-partitioned-topic) command.
+
+```shell
+
+$ pulsar-admin topics update-partitioned-topic \
+ persistent://my-tenant/my-namespace/my-topic \
+ --partitions 8
+
+```
+
+#### REST API
+
+{@inject: endpoint|POST|/admin/v2/:schema/:tenant/:cluster/:namespace/:destination/partitions|operation/updatePartitionedTopic?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+admin.persistentTopics().updatePartitionedTopic(persistentTopic, numPartitions);
+
+```
+
+### Delete
+
+#### pulsar-admin
+
+Partitioned topics can be deleted using the [`delete-partitioned-topic`](reference-pulsar-admin.md#delete-partitioned-topic) command, specifying the topic by name:
+
+```shell
+
+$ bin/pulsar-admin topics delete-partitioned-topic \
+ persistent://my-tenant/my-namespace/my-topic
+
+```
+
+#### REST API
+
+{@inject: endpoint|DELETE|/admin/v2/:schema/:topic/:namespace/:destination/partitions|operation/deletePartitionedTopic?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+admin.persistentTopics().delete(persistentTopic);
+
+```
+
+### List
+
+It provides a list of persistent topics existing under a given namespace.
+
+#### pulsar-admin
+
+```shell
+
+$ pulsar-admin topics list tenant/namespace
+persistent://tenant/namespace/topic1
+persistent://tenant/namespace/topic2
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v2/:schema/:tenant/:namespace|operation/getPartitionedTopicList?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+admin.persistentTopics().getList(namespace);
+
+```
+
+### Stats
+
+It shows current statistics of a given partitioned topic. Here's an example payload:
+
+```json
+
+{
+ "msgRateIn": 4641.528542257553,
+ "msgThroughputIn": 44663039.74947473,
+ "msgRateOut": 0,
+ "msgThroughputOut": 0,
+ "averageMsgSize": 1232439.816728665,
+ "storageSize": 135532389160,
+ "publishers": [
+ {
+ "msgRateIn": 57.855383881403576,
+ "msgThroughputIn": 558994.7078932219,
+ "averageMsgSize": 613135,
+ "producerId": 0,
+ "producerName": null,
+ "address": null,
+ "connectedSince": null
+ }
+ ],
+ "subscriptions": {
+ "my-topic_subscription": {
+ "msgRateOut": 0,
+ "msgThroughputOut": 0,
+ "msgBacklog": 116632,
+ "type": null,
+ "msgRateExpired": 36.98245516804671,
+ "consumers": []
+ }
+ },
+ "replication": {}
+}
+
+```
+
+The following stats are available:
+
+|Stat|Description|
+|----|-----------|
+|msgRateIn|The sum of all local and replication publishers’ publish rates in messages per second|
+|msgThroughputIn|Same as msgRateIn but in bytes per second instead of messages per second|
+|msgRateOut|The sum of all local and replication consumers’ dispatch rates in messages per second|
+|msgThroughputOut|Same as msgRateOut but in bytes per second instead of messages per second|
+|averageMsgSize|Average message size, in bytes, from this publisher within the last interval|
+|storageSize|The sum of the ledgers’ storage size for this topic|
+|publishers|The list of all local publishers into the topic. There can be anywhere from zero to thousands.|
+|producerId|Internal identifier for this producer on this topic|
+|producerName|Internal identifier for this producer, generated by the client library|
+|address|IP address and source port for the connection of this producer|
+|connectedSince|Timestamp this producer was created or last reconnected|
+|subscriptions|The list of all local subscriptions to the topic|
+|my-subscription|The name of this subscription (client defined)|
+|msgBacklog|The count of messages in backlog for this subscription|
+|type|This subscription type|
+|msgRateExpired|The rate at which messages were discarded instead of dispatched from this subscription due to TTL|
+|consumers|The list of connected consumers for this subscription|
+|consumerName|Internal identifier for this consumer, generated by the client library|
+|availablePermits|The number of messages this consumer has space for in the client library’s listen queue. A value of 0 means the client library’s queue is full and receive() isn’t being called. A nonzero value means this consumer is ready to be dispatched messages.|
+|replication|This section gives the stats for cross-colo replication of this topic|
+|replicationBacklog|The outbound replication backlog in messages|
+|connected|Whether the outbound replicator is connected|
+|replicationDelayInSeconds|How long the oldest message has been waiting to be sent through the connection, if connected is true|
+|inboundConnection|The IP and port of the broker in the remote cluster’s publisher connection to this broker|
+|inboundConnectedSince|The TCP connection being used to publish messages to the remote cluster. If there are no local publishers connected, this connection is automatically closed after a minute.|
+
+#### pulsar-admin
+
+The stats for the partitioned topic and its connected producers and consumers can be fetched by using the [`partitioned-stats`](reference-pulsar-admin.md#partitioned-stats) command, specifying the topic by name:
+
+```shell
+
+$ pulsar-admin topics partitioned-stats \
+ persistent://test-tenant/namespace/topic \
+ --per-partition
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v2/:schema/:tenant/:namespace/:topic/partitioned-stats|operation/getPartitionedStats?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+admin.persistentTopics().getStats(persistentTopic);
+
+```
+
+### Internal stats
+
+It shows detailed statistics of a topic.
+
+|Stat|Description|
+|----|-----------|
+|entriesAddedCounter|Messages published since this broker loaded this topic|
+|numberOfEntries|Total number of messages being tracked|
+|totalSize|Total storage size in bytes of all messages|
+|currentLedgerEntries|Count of messages written to the ledger currently open for writing|
+|currentLedgerSize|Size in bytes of messages written to ledger currently open for writing|
+|lastLedgerCreatedTimestamp|Time when last ledger was created|
+|lastLedgerCreationFailureTimestamp|time when last ledger was failed|
+|waitingCursorsCount|How many cursors are caught up and waiting for a new message to be published|
+|pendingAddEntriesCount|How many messages have (asynchronous) write requests we are waiting on completion|
+|lastConfirmedEntry|The ledgerid:entryid of the last message successfully written. If the entryid is -1, then the ledger has been opened or is currently being opened but has no entries written yet.|
+|state|The state of the cursor ledger. Open means we have a cursor ledger for saving updates of the markDeletePosition.|
+|ledgers|The ordered list of all ledgers for this topic holding its messages|
+|cursors|The list of all cursors on this topic. There will be one for every subscription you saw in the topic stats.|
+|markDeletePosition|The ack position: the last message the subscriber acknowledged receiving|
+|readPosition|The latest position of subscriber for reading message|
+|waitingReadOp|This is true when the subscription has read the latest message published to the topic and is waiting on new messages to be published.|
+|pendingReadOps|The counter for how many outstanding read requests to the BookKeepers we have in progress|
+|messagesConsumedCounter|Number of messages this cursor has acked since this broker loaded this topic|
+|cursorLedger|The ledger being used to persistently store the current markDeletePosition|
+|cursorLedgerLastEntry|The last entryid used to persistently store the current markDeletePosition|
+|individuallyDeletedMessages|If Acks are being done out of order, shows the ranges of messages Acked between the markDeletePosition and the read-position|
+|lastLedgerSwitchTimestamp|The last time the cursor ledger was rolled over|
+
+```json
+
+{
+ "entriesAddedCounter": 20449518,
+ "numberOfEntries": 3233,
+ "totalSize": 331482,
+ "currentLedgerEntries": 3233,
+ "currentLedgerSize": 331482,
+ "lastLedgerCreatedTimestamp": "2016-06-29 03:00:23.825",
+ "lastLedgerCreationFailureTimestamp": null,
+ "waitingCursorsCount": 1,
+ "pendingAddEntriesCount": 0,
+ "lastConfirmedEntry": "324711539:3232",
+ "state": "LedgerOpened",
+ "ledgers": [
+ {
+ "ledgerId": 324711539,
+ "entries": 0,
+ "size": 0
+ }
+ ],
+ "cursors": {
+ "my-subscription": {
+ "markDeletePosition": "324711539:3133",
+ "readPosition": "324711539:3233",
+ "waitingReadOp": true,
+ "pendingReadOps": 0,
+ "messagesConsumedCounter": 20449501,
+ "cursorLedger": 324702104,
+ "cursorLedgerLastEntry": 21,
+ "individuallyDeletedMessages": "[(324711539:3134‥324711539:3136], (324711539:3137‥324711539:3140], ]",
+ "lastLedgerSwitchTimestamp": "2016-06-29 01:30:19.313",
+ "state": "Open"
+ }
+ }
+}
+
+```
+
+#### pulsar-admin
+
+The internal stats for the partitioned topic can be fetched by using the [`stats-internal`](reference-pulsar-admin.md#stats-internal) command, specifying the topic by name:
+
+```shell
+
+$ pulsar-admin topics stats-internal \
+ persistent://test-tenant/namespace/topic
+
+```
+
+#### REST API
+
+{@inject: endpoint|GET|/admin/v2/:schema/:tenant/:namespace/:topic/internalStats|operation/getInternalStats?version=@pulsar:version_number@}
+
+#### Java
+
+```java
+
+admin.persistentTopics().getInternalStats(persistentTopic);
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-permissions.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-permissions.md
new file mode 100644
index 0000000..e2ca469
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-permissions.md
@@ -0,0 +1,181 @@
+---
+id: admin-api-permissions
+title: Managing permissions
+sidebar_label: "Permissions"
+original_id: admin-api-permissions
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+> **Important**
+>
+> This page only shows **some frequently used operations**.
+>
+> - For the latest and complete information about `Pulsar admin`, including commands, flags, descriptions, and more information, see [Pulsar admin doc](https://pulsar.apache.org/tools/pulsar-admin/).
+>
+> - For the latest and complete information about `REST API`, including parameters, responses, samples, and more, see {@inject: rest:REST:/} API doc.
+>
+> - For the latest and complete information about `Java admin API`, including classes, methods, descriptions, and more, see [Java admin API doc](https://pulsar.apache.org/api/admin/).
+
+Pulsar allows you to grant namespace-level or topic-level permission to users.
+
+- If you grant a namespace-level permission to a user, then the user can access all the topics under the namespace.
+
+- If you grant a topic-level permission to a user, then the user can access only the topic.
+
+The chapters below demonstrate how to grant namespace-level permissions to users. For how to grant topic-level permissions to users, see [manage topics](admin-api-topics.md/#grant-permission).
+
+## Grant permissions
+
+You can grant permissions to specific roles for lists of operations such as `produce` and `consume`.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`grant-permission`](reference-pulsar-admin.md#grant-permission) subcommand and specify a namespace, actions using the `--actions` flag, and a role using the `--role` flag:
+
+```shell
+
+$ pulsar-admin namespaces grant-permission test-tenant/ns1 \
+ --actions produce,consume \
+ --role admin10
+
+```
+
+Wildcard authorization can be performed when `authorizationAllowWildcardsMatching` is set to `true` in `broker.conf`.
+
+e.g.
+
+```shell
+
+$ pulsar-admin namespaces grant-permission test-tenant/ns1 \
+ --actions produce,consume \
+ --role 'my.role.*'
+
+```
+
+Then, roles `my.role.1`, `my.role.2`, `my.role.foo`, `my.role.bar`, etc. can produce and consume.
+
+```shell
+
+$ pulsar-admin namespaces grant-permission test-tenant/ns1 \
+ --actions produce,consume \
+ --role '*.role.my'
+
+```
+
+Then, roles `1.role.my`, `2.role.my`, `foo.role.my`, `bar.role.my`, etc. can produce and consume.
+
+**Note**: A wildcard matching works at **the beginning or end of the role name only**.
+
+e.g.
+
+```shell
+
+$ pulsar-admin namespaces grant-permission test-tenant/ns1 \
+ --actions produce,consume \
+ --role 'my.*.role'
+
+```
+
+In this case, only the role `my.*.role` has permissions.
+Roles `my.1.role`, `my.2.role`, `my.foo.role`, `my.bar.role`, etc. **cannot** produce and consume.
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|POST|/admin/v2/namespaces/:tenant/:namespace/permissions/:role|operation/grantPermissionOnNamespace?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().grantPermissionOnNamespace(namespace, role, getAuthActions(actions));
+
+```
+
+</TabItem>
+
+</Tabs>
+
+## Get permissions
+
+You can see which permissions have been granted to which roles in a namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`permissions`](reference-pulsar-admin#permissions) subcommand and specify a namespace:
+
+```shell
+
+$ pulsar-admin namespaces permissions test-tenant/ns1
+{
+ "admin10": [
+ "produce",
+ "consume"
+ ]
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/namespaces/:tenant/:namespace/permissions|operation/getPermissions?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().getPermissions(namespace);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+## Revoke permissions
+
+You can revoke permissions from specific roles, which means that those roles will no longer have access to the specified namespace.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"Java","value":"Java"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`revoke-permission`](reference-pulsar-admin.md#revoke-permission) subcommand and specify a namespace and a role using the `--role` flag:
+
+```shell
+
+$ pulsar-admin namespaces revoke-permission test-tenant/ns1 \
+ --role admin10
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|DELETE|/admin/v2/namespaces/:tenant/:namespace/permissions/:role|operation/revokePermissionsOnNamespace?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java">
+
+```java
+
+admin.namespaces().revokePermissionsOnNamespace(namespace, role);
+
+```
+
+</TabItem>
+
+</Tabs>
\ No newline at end of file
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-persistent-topics.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-persistent-topics.md
new file mode 100644
index 0000000..b6d293b
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-persistent-topics.md
@@ -0,0 +1,8 @@
+---
+id: admin-api-persistent-topics
+title: Managing persistent topics
+sidebar_label: "Persistent topics"
+original_id: admin-api-persistent-topics
+---
+
+For details of the content, refer to [manage topics](admin-api-topics).
\ No newline at end of file
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-schemas.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-schemas.md
new file mode 100644
index 0000000..9ffe21f
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-schemas.md
@@ -0,0 +1,7 @@
+---
+id: admin-api-schemas
+title: Managing Schemas
+sidebar_label: "Schemas"
+original_id: admin-api-schemas
+---
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/admin-api-tenants.md b/site2/website-next/versioned_docs/version-2.4.1/admin-api-tenants.md
new file mode 100644
index 0000000..fe68336
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/admin-api-tenants.md
@@ -0,0 +1,226 @@
+---
+id: admin-api-tenants
+title: Managing Tenants
+sidebar_label: "Tenants"
+original_id: admin-api-tenants
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+> **Important**
+>
+> This page only shows **some frequently used operations**.
+>
+> - For the latest and complete information about `Pulsar admin`, including commands, flags, descriptions, and more information, see [Pulsar admin doc](https://pulsar.apache.org/tools/pulsar-admin/).
+>
+> - For the latest and complete information about `REST API`, including parameters, responses, samples, and more, see {@inject: rest:REST:/} API doc.
+>
+> - For the latest and complete information about `Java admin API`, including classes, methods, descriptions, and more, see [Java admin API doc](https://pulsar.apache.org/api/admin/).
+
+Tenants, like namespaces, can be managed using the [admin API](admin-api-overview). There are currently two configurable aspects of tenants:
+
+* Admin roles
+* Allowed clusters
+
+## Tenant resources
+
+### List
+
+You can list all of the tenants associated with an [instance](reference-terminology.md#instance).
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`list`](reference-pulsar-admin.md#tenants-list) subcommand.
+
+```shell
+
+$ pulsar-admin tenants list
+my-tenant-1
+my-tenant-2
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/tenants|operation/getTenants?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.tenants().getTenants();
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Create
+
+You can create a new tenant.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`create`](reference-pulsar-admin.md#tenants-create) subcommand:
+
+```shell
+
+$ pulsar-admin tenants create my-tenant
+
+```
+
+When creating a tenant, you can assign admin roles using the `-r`/`--admin-roles` flag. You can specify multiple roles as a comma-separated list. Here are some examples:
+
+```shell
+
+$ pulsar-admin tenants create my-tenant \
+ --admin-roles role1,role2,role3
+
+$ pulsar-admin tenants create my-tenant \
+ -r role1
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|POST|/admin/v2/tenants/:tenant|operation/createTenant?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.tenants().createTenant(tenantName, tenantInfo);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Get configuration
+
+You can fetch the [configuration](reference-configuration) for an existing tenant at any time.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`get`](reference-pulsar-admin.md#tenants-get) subcommand and specify the name of the tenant. Here's an example:
+
+```shell
+
+$ pulsar-admin tenants get my-tenant
+{
+ "adminRoles": [
+ "admin1",
+ "admin2"
+ ],
+ "allowedClusters": [
+ "cl1",
+ "cl2"
+ ]
+}
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|GET|/admin/v2/tenants/:cluster|operation/getTenant?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.tenants().getTenantInfo(tenantName);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Delete
+
+Tenants can be deleted from a Pulsar [instance](reference-terminology.md#instance).
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`delete`](reference-pulsar-admin.md#tenants-delete) subcommand and specify the name of the tenant.
+
+```shell
+
+$ pulsar-admin tenants delete my-tenant
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|DELETE|/admin/v2/tenants/:cluster|operation/deleteTenant?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.Tenants().deleteTenant(tenantName);
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Update
+
+You can update a tenant's configuration.
+
+<Tabs
+ defaultValue="pulsar-admin"
+ values={[{"label":"pulsar-admin","value":"pulsar-admin"},{"label":"REST API","value":"REST API"},{"label":"JAVA","value":"JAVA"}]}>
+<TabItem value="pulsar-admin">
+
+Use the [`update`](reference-pulsar-admin.md#tenants-update) subcommand.
+
+```shell
+
+$ pulsar-admin tenants update my-tenant
+
+```
+
+</TabItem>
+<TabItem value="REST API">
+
+{@inject: endpoint|DELETE|/admin/v2/tenants/:cluster|operation/updateTenant?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="JAVA">
+
+```java
+
+admin.tenants().updateTenant(tenantName, tenantInfo);
+
+```
+
+</TabItem>
+
+</Tabs>
diff --git a/site2/website-next/versioned_docs/version-2.4.1/client-libraries-cpp.md b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-cpp.md
new file mode 100644
index 0000000..333ec67
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-cpp.md
@@ -0,0 +1,407 @@
+---
+id: client-libraries-cpp
+title: Pulsar C++ client
+sidebar_label: "C++"
+original_id: client-libraries-cpp
+---
+
+You can use Pulsar C++ client to create Pulsar producers and consumers in C++.
+
+All the methods in producer, consumer, and reader of a C++ client are thread-safe.
+
+## Supported platforms
+
+Pulsar C++ client is supported on **Linux** and **MacOS** platforms.
+
+[Doxygen](http://www.doxygen.nl/)-generated API docs for the C++ client are available [here](/api/cpp).
+
+## System requirements
+
+You need to install the following components before using the C++ client:
+
+* [CMake](https://cmake.org/)
+* [Boost](http://www.boost.org/)
+* [Protocol Buffers](https://developers.google.com/protocol-buffers/) 2.6
+* [libcurl](https://curl.haxx.se/libcurl/)
+* [Google Test](https://github.com/google/googletest)
+
+## Linux
+
+### Compilation
+
+1. Clone the Pulsar repository.
+
+```shell
+
+$ git clone https://github.com/apache/pulsar
+
+```
+
+2. Install all necessary dependencies.
+
+```shell
+
+$ apt-get install cmake libssl-dev libcurl4-openssl-dev liblog4cxx-dev \
+ libprotobuf-dev protobuf-compiler libboost-all-dev google-mock libgtest-dev libjsoncpp-dev
+
+```
+
+3. Compile and install [Google Test](https://github.com/google/googletest).
+
+```shell
+
+# libgtest-dev version is 1.18.0 or above
+$ cd /usr/src/googletest
+$ sudo cmake .
+$ sudo make
+$ sudo cp ./googlemock/libgmock.a ./googlemock/gtest/libgtest.a /usr/lib/
+
+# less than 1.18.0
+$ cd /usr/src/gtest
+$ sudo cmake .
+$ sudo make
+$ sudo cp libgtest.a /usr/lib
+
+$ cd /usr/src/gmock
+$ sudo cmake .
+$ sudo make
+$ sudo cp libgmock.a /usr/lib
+
+```
+
+4. Compile the Pulsar client library for C++ inside the Pulsar repository.
+
+```shell
+
+$ cd pulsar-client-cpp
+$ cmake .
+$ make
+
+```
+
+After you install the components successfully, the files `libpulsar.so` and `libpulsar.a` are in the `lib` folder of the repository. The tools `perfProducer` and `perfConsumer` are in the `perf` directory.
+
+### Install Dependencies
+
+> Since 2.1.0 release, Pulsar ships pre-built RPM and Debian packages. You can download and install those packages directly.
+
+After you download and install RPM or DEB, the `libpulsar.so`, `libpulsarnossl.so`, `libpulsar.a`, and `libpulsarwithdeps.a` libraries are in your `/usr/lib` directory.
+
+By default, they are built in code path `${PULSAR_HOME}/pulsar-client-cpp`. You can build with the command below.
+
+ `cmake . -DBUILD_TESTS=OFF -DLINK_STATIC=ON && make pulsarShared pulsarSharedNossl pulsarStatic pulsarStaticWithDeps -j 3`.
+
+These libraries rely on some other libraries. If you want to get detailed version of dependencies, see [RPM](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/pkg/rpm/Dockerfile) or [DEB](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/pkg/deb/Dockerfile) files.
+
+1. `libpulsar.so` is a shared library, containing statically linked `boost` and `openssl`. It also dynamically links all other necessary libraries. You can use this Pulsar library with the command below.
+
+```bash
+
+ g++ --std=c++11 PulsarTest.cpp -o test /usr/lib/libpulsar.so -I/usr/local/ssl/include
+
+```
+
+2. `libpulsarnossl.so` is a shared library, similar to `libpulsar.so` except that the libraries `openssl` and `crypto` are dynamically linked. You can use this Pulsar library with the command below.
+
+```bash
+
+ g++ --std=c++11 PulsarTest.cpp -o test /usr/lib/libpulsarnossl.so -lssl -lcrypto -I/usr/local/ssl/include -L/usr/local/ssl/lib
+
+```
+
+3. `libpulsar.a` is a static library. You need to load dependencies before using this library. You can use this Pulsar library with the command below.
+
+```bash
+
+ g++ --std=c++11 PulsarTest.cpp -o test /usr/lib/libpulsar.a -lssl -lcrypto -ldl -lpthread -I/usr/local/ssl/include -L/usr/local/ssl/lib -lboost_system -lboost_regex -lcurl -lprotobuf -lzstd -lz
+
+```
+
+4. `libpulsarwithdeps.a` is a static library, based on `libpulsar.a`. It is archived in the dependencies of `libboost_regex`, `libboost_system`, `libcurl`, `libprotobuf`, `libzstd` and `libz`. You can use this Pulsar library with the command below.
+
+```bash
+
+ g++ --std=c++11 PulsarTest.cpp -o test /usr/lib/libpulsarwithdeps.a -lssl -lcrypto -ldl -lpthread -I/usr/local/ssl/include -L/usr/local/ssl/lib
+
+```
+
+The `libpulsarwithdeps.a` does not include library openssl related libraries `libssl` and `libcrypto`, because these two libraries are related to security. It is more reasonable and easier to use the versions provided by the local system to handle security issues and upgrade libraries.
+
+### Install RPM
+
+1. Download a RPM package from the links in the table.
+
+| Link | Crypto files |
+|------|--------------|
+| [client](@pulsar:dist_rpm:client@) | [asc](@pulsar:dist_rpm:client@.asc), [sha512](@pulsar:dist_rpm:client@.sha512) |
+| [client-debuginfo](@pulsar:dist_rpm:client-debuginfo@) | [asc](@pulsar:dist_rpm:client-debuginfo@.asc), [sha512](@pulsar:dist_rpm:client-debuginfo@.sha512) |
+| [client-devel](@pulsar:dist_rpm:client-devel@) | [asc](@pulsar:dist_rpm:client-devel@.asc), [sha512](@pulsar:dist_rpm:client-devel@.sha512) |
+
+2. Install the package using the following command.
+
+```bash
+
+$ rpm -ivh apache-pulsar-client*.rpm
+
+```
+
+After you install RPM successfully, Pulsar libraries are in the `/usr/lib` directory.
+
+### Install Debian
+
+1. Download a Debian package from the links in the table.
+
+| Link | Crypto files |
+|------|--------------|
+| [client](@pulsar:deb:client@) | [asc](@pulsar:dist_deb:client@.asc), [sha512](@pulsar:dist_deb:client@.sha512) |
+| [client-devel](@pulsar:deb:client-devel@) | [asc](@pulsar:dist_deb:client-devel@.asc), [sha512](@pulsar:dist_deb:client-devel@.sha512) |
+
+2. Install the package using the following command.
+
+```bash
+
+$ apt install ./apache-pulsar-client*.deb
+
+```
+
+After you install DEB successfully, Pulsar libraries are in the `/usr/lib` directory.
+
+### Build
+
+> If you want to build RPM and Debian packages from the latest master, follow the instructions below. You should run all the instructions at the root directory of your cloned Pulsar repository.
+
+There are recipes that build RPM and Debian packages containing a
+statically linked `libpulsar.so` / `libpulsarnossl.so` / `libpulsar.a` / `libpulsarwithdeps.a` with all required dependencies.
+
+To build the C++ library packages, you need to build the Java packages first.
+
+```shell
+
+mvn install -DskipTests
+
+```
+
+#### RPM
+
+To build the RPM inside a Docker container, use the command below. The RPMs are in the `pulsar-client-cpp/pkg/rpm/RPMS/x86_64/` path.
+
+```shell
+
+pulsar-client-cpp/pkg/rpm/docker-build-rpm.sh
+
+```
+
+| Package name | Content |
+|-----|-----|
+| pulsar-client | Shared library `libpulsar.so` and `libpulsarnossl.so` |
+| pulsar-client-devel | Static library `libpulsar.a`, `libpulsarwithdeps.a`and C++ and C headers |
+| pulsar-client-debuginfo | Debug symbols for `libpulsar.so` |
+
+#### Debian
+
+To build Debian packages, enter the following command.
+
+```shell
+
+pulsar-client-cpp/pkg/deb/docker-build-deb.sh
+
+```
+
+Debian packages are created in the `pulsar-client-cpp/pkg/deb/BUILD/DEB/` path.
+
+| Package name | Content |
+|-----|-----|
+| pulsar-client | Shared library `libpulsar.so` and `libpulsarnossl.so` |
+| pulsar-client-dev | Static library `libpulsar.a`, `libpulsarwithdeps.a` and C++ and C headers |
+
+## MacOS
+
+### Compilation
+
+1. Clone the Pulsar repository.
+
+```shell
+
+$ git clone https://github.com/apache/pulsar
+
+```
+
+2. Install all necessary dependencies.
+
+```shell
+
+# OpenSSL installation
+$ brew install openssl
+$ export OPENSSL_INCLUDE_DIR=/usr/local/opt/openssl/include/
+$ export OPENSSL_ROOT_DIR=/usr/local/opt/openssl/
+
+# Protocol Buffers installation
+$ brew tap homebrew/versions
+$ brew install protobuf260
+$ brew install boost
+$ brew install log4cxx
+
+# Google Test installation
+$ git clone https://github.com/google/googletest.git
+$ cd googletest
+$ cmake .
+$ make install
+
+```
+
+3. Compile the Pulsar client library in the repository that you cloned.
+
+```shell
+
+$ cd pulsar-client-cpp
+$ cmake .
+$ make
+
+```
+
+### Install `libpulsar`
+
+Pulsar releases are available in the [Homebrew](https://brew.sh/) core repository. You can install the C++ client library with the following command. The package is installed with the library and headers.
+
+```shell
+
+brew install libpulsar
+
+```
+
+## Connection URLs
+
+To connect Pulsar using client libraries, you need to specify a Pulsar protocol URL.
+
+Pulsar protocol URLs are assigned to specific clusters, you can use the Pulsar URI scheme. The default port is `6650`. The following is an example for localhost.
+
+```http
+
+pulsar://localhost:6650
+
+```
+
+In a Pulsar cluster in production, the URL looks as follows.
+
+```http
+
+pulsar://pulsar.us-west.example.com:6650
+
+```
+
+If you use TLS authentication, you need to add `ssl`, and the default port is `6651`. The following is an example.
+
+```http
+
+pulsar+ssl://pulsar.us-west.example.com:6651
+
+```
+
+## Create a consumer
+
+To use Pulsar as a consumer, you need to create a consumer on the C++ client. The following is an example.
+
+```c++
+
+Client client("pulsar://localhost:6650");
+
+Consumer consumer;
+Result result = client.subscribe("my-topic", "my-subscription-name", consumer);
+if (result != ResultOk) {
+ LOG_ERROR("Failed to subscribe: " << result);
+ return -1;
+}
+
+Message msg;
+
+while (true) {
+ consumer.receive(msg);
+ LOG_INFO("Received: " << msg
+ << " with payload '" << msg.getDataAsString() << "'");
+
+ consumer.acknowledge(msg);
+}
+
+client.close();
+
+```
+
+## Create a producer
+
+To use Pulsar as a producer, you need to create a producer on the C++ client. The following is an example.
+
+```c++
+
+Client client("pulsar://localhost:6650");
+
+Producer producer;
+Result result = client.createProducer("my-topic", producer);
+if (result != ResultOk) {
+ LOG_ERROR("Error creating producer: " << result);
+ return -1;
+}
+
+// Publish 10 messages to the topic
+for (int i = 0; i < 10; i++){
+ Message msg = MessageBuilder().setContent("my-message").build();
+ Result res = producer.send(msg);
+ LOG_INFO("Message sent: " << res);
+}
+client.close();
+
+```
+
+## Enable authentication in connection URLs
+If you use TLS authentication when connecting to Pulsar, you need to add `ssl` in the connection URLs, and the default port is `6651`. The following is an example.
+
+```cpp
+
+ClientConfiguration config = ClientConfiguration();
+config.setUseTls(true);
+config.setTlsTrustCertsFilePath("/path/to/cacert.pem");
+config.setTlsAllowInsecureConnection(false);
+config.setAuth(pulsar::AuthTls::create(
+ "/path/to/client-cert.pem", "/path/to/client-key.pem"););
+
+Client client("pulsar+ssl://my-broker.com:6651", config);
+
+```
+
+For complete examples, refer to [C++ client examples](https://github.com/apache/pulsar/tree/master/pulsar-client-cpp/examples).
+
+## Schema
+
+This section describes some examples about schema. For more information about schema, see [Pulsar schema](schema-get-started).
+
+### Create producer with Avro schema
+
+The following example shows how to create a producer with an Avro schema.
+
+```cpp
+
+static const std::string exampleSchema =
+ "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\","
+ "\"fields\":[{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"int\"}]}";
+Producer producer;
+ProducerConfiguration producerConf;
+producerConf.setSchema(SchemaInfo(AVRO, "Avro", exampleSchema));
+client.createProducer("topic-avro", producerConf, producer);
+
+```
+
+### Create consumer with Avro schema
+
+The following example shows how to create a consumer with an Avro schema.
+
+```cpp
+
+static const std::string exampleSchema =
+ "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\","
+ "\"fields\":[{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"int\"}]}";
+ConsumerConfiguration consumerConf;
+Consumer consumer;
+consumerConf.setSchema(SchemaInfo(AVRO, "Avro", exampleSchema));
+client.subscribe("topic-avro", "sub-2", consumerConf, consumer)
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/client-libraries-go.md b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-go.md
new file mode 100644
index 0000000..7e797bb
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-go.md
@@ -0,0 +1,884 @@
+---
+id: client-libraries-go
+title: Pulsar Go client
+sidebar_label: "Go"
+original_id: client-libraries-go
+---
+
+> Tips: Currently, the CGo client will be deprecated, if you want to know more about the CGo client, please refer to [CGo client docs](client-libraries-cgo)
+
+You can use Pulsar [Go client](https://github.com/apache/pulsar-client-go) to create Pulsar [producers](#producers), [consumers](#consumers), and [readers](#readers) in Go (aka Golang).
+
+> **API docs available as well**
+> For standard API docs, consult the [Godoc](https://godoc.org/github.com/apache/pulsar-client-go/pulsar).
+
+
+## Installation
+
+### Install go package
+
+You can install the `pulsar` library locally using `go get`.
+
+```bash
+
+$ go get -u "github.com/apache/pulsar-client-go/pulsar"
+
+```
+
+Once installed locally, you can import it into your project:
+
+```go
+
+import "github.com/apache/pulsar-client-go/pulsar"
+
+```
+
+## Connection URLs
+
+To connect to Pulsar using client libraries, you need to specify a [Pulsar protocol](developing-binary-protocol) URL.
+
+Pulsar protocol URLs are assigned to specific clusters, use the `pulsar` scheme and have a default port of 6650. Here's an example for `localhost`:
+
+```http
+
+pulsar://localhost:6650
+
+```
+
+If you have multiple brokers, you can set the URL as below.
+
+```
+
+pulsar://localhost:6550,localhost:6651,localhost:6652
+
+```
+
+A URL for a production Pulsar cluster may look something like this:
+
+```http
+
+pulsar://pulsar.us-west.example.com:6650
+
+```
+
+If you're using [TLS](security-tls-authentication) authentication, the URL will look like something like this:
+
+```http
+
+pulsar+ssl://pulsar.us-west.example.com:6651
+
+```
+
+## Create a client
+
+In order to interact with Pulsar, you'll first need a `Client` object. You can create a client object using the `NewClient` function, passing in a `ClientOptions` object (more on configuration [below](#client-configuration)). Here's an example:
+
+```go
+
+import (
+ "log"
+ "time"
+
+ "github.com/apache/pulsar-client-go/pulsar"
+)
+
+func main() {
+ client, err := pulsar.NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650",
+ OperationTimeout: 30 * time.Second,
+ ConnectionTimeout: 30 * time.Second,
+ })
+ if err != nil {
+ log.Fatalf("Could not instantiate Pulsar client: %v", err)
+ }
+
+ defer client.Close()
+}
+
+```
+
+If you have multiple brokers, you can initiate a client object as below.
+
+```
+
+import (
+ "log"
+ "time"
+ "github.com/apache/pulsar-client-go/pulsar"
+)
+
+func main() {
+ client, err := pulsar.NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650,localhost:6651,localhost:6652",
+ OperationTimeout: 30 * time.Second,
+ ConnectionTimeout: 30 * time.Second,
+ })
+ if err != nil {
+ log.Fatalf("Could not instantiate Pulsar client: %v", err)
+ }
+
+ defer client.Close()
+}
+
+```
+
+The following configurable parameters are available for Pulsar clients:
+
+ Name | Description | Default
+| :-------- | :---------- |:---------- |
+| URL | Configure the service URL for the Pulsar service.<br /><br />If you have multiple brokers, you can set multiple Pulsar cluster addresses for a client. <br /><br />This parameter is **required**. |None |
+| ConnectionTimeout | Timeout for the establishment of a TCP connection | 30s |
+| OperationTimeout| Set the operation timeout. Producer-create, subscribe and unsubscribe operations will be retried until this interval, after which the operation will be marked as failed| 30s|
+| Authentication | Configure the authentication provider. Example: `Authentication: NewAuthenticationTLS("my-cert.pem", "my-key.pem")` | no authentication |
+| TLSTrustCertsFilePath | Set the path to the trusted TLS certificate file | |
+| TLSAllowInsecureConnection | Configure whether the Pulsar client accept untrusted TLS certificate from broker | false |
+| TLSValidateHostname | Configure whether the Pulsar client verify the validity of the host name from broker | false |
+| ListenerName | Configure the net model for VPC users to connect to the Pulsar broker | |
+| MaxConnectionsPerBroker | Max number of connections to a single broker that is kept in the pool | 1 |
+| CustomMetricsLabels | Add custom labels to all the metrics reported by this client instance | |
+| Logger | Configure the logger used by the client | logrus.StandardLogger |
+
+## Producers
+
+Pulsar producers publish messages to Pulsar topics. You can [configure](#producer-configuration) Go producers using a `ProducerOptions` object. Here's an example:
+
+```go
+
+producer, err := client.CreateProducer(pulsar.ProducerOptions{
+ Topic: "my-topic",
+})
+
+if err != nil {
+ log.Fatal(err)
+}
+
+_, err = producer.Send(context.Background(), &pulsar.ProducerMessage{
+ Payload: []byte("hello"),
+})
+
+defer producer.Close()
+
+if err != nil {
+ fmt.Println("Failed to publish message", err)
+}
+fmt.Println("Published message")
+
+```
+
+### Producer operations
+
+Pulsar Go producers have the following methods available:
+
+Method | Description | Return type
+:------|:------------|:-----------
+`Topic()` | Fetches the producer's [topic](reference-terminology.md#topic)| `string`
+`Name()` | Fetches the producer's name | `string`
+`Send(context.Context, *ProducerMessage)` | Publishes a [message](#messages) to the producer's topic. This call will block until the message is successfully acknowledged by the Pulsar broker, or an error will be thrown if the timeout set using the `SendTimeout` in the producer's [configuration](#producer-configuration) is exceeded. | (MessageID, error)
+`SendAsync(context.Context, *ProducerMessage, func(MessageID, *ProducerMessage, error))`| Send a message, this call will be blocking until is successfully acknowledged by the Pulsar broker. |
+`LastSequenceID()` | Get the last sequence id that was published by this producer. his represent either the automatically assigned or custom sequence id (set on the ProducerMessage) that was published and acknowledged by the broker. | int64
+`Flush()`| Flush all the messages buffered in the client and wait until all messages have been successfully persisted. | error
+`Close()` | Closes the producer and releases all resources allocated to it. If `Close()` is called then no more messages will be accepted from the publisher. This method will block until all pending publish requests have been persisted by Pulsar. If an error is thrown, no pending writes will be retried. |
+
+### Producer Example
+
+#### How to use message router in producer
+
+```go
+
+client, err := NewClient(pulsar.ClientOptions{
+ URL: serviceURL,
+})
+
+if err != nil {
+ log.Fatal(err)
+}
+defer client.Close()
+
+// Only subscribe on the specific partition
+consumer, err := client.Subscribe(pulsar.ConsumerOptions{
+ Topic: "my-partitioned-topic-partition-2",
+ SubscriptionName: "my-sub",
+})
+
+if err != nil {
+ log.Fatal(err)
+}
+defer consumer.Close()
+
+producer, err := client.CreateProducer(pulsar.ProducerOptions{
+ Topic: "my-partitioned-topic",
+ MessageRouter: func(msg *ProducerMessage, tm TopicMetadata) int {
+ fmt.Println("Routing message ", msg, " -- Partitions: ", tm.NumPartitions())
+ return 2
+ },
+})
+
+if err != nil {
+ log.Fatal(err)
+}
+defer producer.Close()
+
+```
+
+#### How to use schema interface in producer
+
+```go
+
+type testJSON struct {
+ ID int `json:"id"`
+ Name string `json:"name"`
+}
+
+```
+
+```go
+
+var (
+ exampleSchemaDef = "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\"," +
+ "\"fields\":[{\"name\":\"ID\",\"type\":\"int\"},{\"name\":\"Name\",\"type\":\"string\"}]}"
+)
+
+```
+
+```go
+
+client, err := NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650",
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer client.Close()
+
+properties := make(map[string]string)
+properties["pulsar"] = "hello"
+jsonSchemaWithProperties := NewJSONSchema(exampleSchemaDef, properties)
+producer, err := client.CreateProducer(ProducerOptions{
+ Topic: "jsonTopic",
+ Schema: jsonSchemaWithProperties,
+})
+assert.Nil(t, err)
+
+_, err = producer.Send(context.Background(), &ProducerMessage{
+ Value: &testJSON{
+ ID: 100,
+ Name: "pulsar",
+ },
+})
+if err != nil {
+ log.Fatal(err)
+}
+producer.Close()
+
+```
+
+#### How to use delay relative in producer
+
+```go
+
+client, err := NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650",
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer client.Close()
+
+topicName := newTopicName()
+producer, err := client.CreateProducer(pulsar.ProducerOptions{
+ Topic: topicName,
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer producer.Close()
+
+consumer, err := client.Subscribe(pulsar.ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "subName",
+ Type: Shared,
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer consumer.Close()
+
+ID, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
+ Payload: []byte(fmt.Sprintf("test")),
+ DeliverAfter: 3 * time.Second,
+})
+if err != nil {
+ log.Fatal(err)
+}
+fmt.Println(ID)
+
+ctx, canc := context.WithTimeout(context.Background(), 1*time.Second)
+msg, err := consumer.Receive(ctx)
+if err != nil {
+ log.Fatal(err)
+}
+fmt.Println(msg.Payload())
+canc()
+
+ctx, canc = context.WithTimeout(context.Background(), 5*time.Second)
+msg, err = consumer.Receive(ctx)
+if err != nil {
+ log.Fatal(err)
+}
+fmt.Println(msg.Payload())
+canc()
+
+```
+
+### Producer configuration
+
+ Name | Description | Default
+| :-------- | :---------- |:---------- |
+| Topic | Topic specify the topic this consumer will subscribe to. This argument is required when constructing the reader. | |
+| Name | Name specify a name for the producer. If not assigned, the system will generate a globally unique name which can be access with Producer.ProducerName(). | |
+| Properties | Properties attach a set of application defined properties to the producer This properties will be visible in the topic stats | |
+| SendTimeout | SendTimeout set the timeout for a message that is not acknowledged by the server | 30s |
+| DisableBlockIfQueueFull | DisableBlockIfQueueFull control whether Send and SendAsync block if producer's message queue is full | false |
+| MaxPendingMessages| MaxPendingMessages set the max size of the queue holding the messages pending to receive an acknowledgment from the broker. | |
+| HashingScheme | HashingScheme change the `HashingScheme` used to chose the partition on where to publish a particular message. | JavaStringHash |
+| CompressionType | CompressionType set the compression type for the producer. | not compressed |
+| CompressionLevel | Define the desired compression level. Options: Default, Faster and Better | Default |
+| MessageRouter | MessageRouter set a custom message routing policy by passing an implementation of MessageRouter | |
+| DisableBatching | DisableBatching control whether automatic batching of messages is enabled for the producer. | false |
+| BatchingMaxPublishDelay | BatchingMaxPublishDelay set the time period within which the messages sent will be batched | 1ms |
+| BatchingMaxMessages | BatchingMaxMessages set the maximum number of messages permitted in a batch. | 1000 |
+| BatchingMaxSize | BatchingMaxSize sets the maximum number of bytes permitted in a batch. | 128KB |
+| Schema | Schema set a custom schema type by passing an implementation of `Schema` | bytes[] |
+| Interceptors | A chain of interceptors. These interceptors are called at some points defined in the `ProducerInterceptor` interface. | None |
+| MaxReconnectToBroker | MaxReconnectToBroker set the maximum retry number of reconnectToBroker | ultimate |
+| BatcherBuilderType | BatcherBuilderType sets the batch builder type. This is used to create a batch container when batching is enabled. Options: DefaultBatchBuilder and KeyBasedBatchBuilder | DefaultBatchBuilder |
+
+## Consumers
+
+Pulsar consumers subscribe to one or more Pulsar topics and listen for incoming messages produced on that topic/those topics. You can [configure](#consumer-configuration) Go consumers using a `ConsumerOptions` object. Here's a basic example that uses channels:
+
+```go
+
+consumer, err := client.Subscribe(pulsar.ConsumerOptions{
+ Topic: "topic-1",
+ SubscriptionName: "my-sub",
+ Type: pulsar.Shared,
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer consumer.Close()
+
+for i := 0; i < 10; i++ {
+ msg, err := consumer.Receive(context.Background())
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
+ msg.ID(), string(msg.Payload()))
+
+ consumer.Ack(msg)
+}
+
+if err := consumer.Unsubscribe(); err != nil {
+ log.Fatal(err)
+}
+
+```
+
+### Consumer operations
+
+Pulsar Go consumers have the following methods available:
+
+Method | Description | Return type
+:------|:------------|:-----------
+`Subscription()` | Returns the consumer's subscription name | `string`
+`Unsubcribe()` | Unsubscribes the consumer from the assigned topic. Throws an error if the unsubscribe operation is somehow unsuccessful. | `error`
+`Receive(context.Context)` | Receives a single message from the topic. This method blocks until a message is available. | `(Message, error)`
+`Chan()` | Chan returns a channel from which to consume messages. | `<-chan ConsumerMessage`
+`Ack(Message)` | [Acknowledges](reference-terminology.md#acknowledgment-ack) a message to the Pulsar [broker](reference-terminology.md#broker) |
+`AckID(MessageID)` | [Acknowledges](reference-terminology.md#acknowledgment-ack) a message to the Pulsar [broker](reference-terminology.md#broker) by message ID |
+`ReconsumeLater(msg Message, delay time.Duration)` | ReconsumeLater mark a message for redelivery after custom delay |
+`Nack(Message)` | Acknowledge the failure to process a single message. |
+`NackID(MessageID)` | Acknowledge the failure to process a single message. |
+`Seek(msgID MessageID)` | Reset the subscription associated with this consumer to a specific message id. The message id can either be a specific message or represent the first or last messages in the topic. | `error`
+`SeekByTime(time time.Time)` | Reset the subscription associated with this consumer to a specific message publish time. | `error`
+`Close()` | Closes the consumer, disabling its ability to receive messages from the broker |
+`Name()` | Name returns the name of consumer | `string`
+
+### Receive example
+
+#### How to use regex consumer
+
+```go
+
+client, err := pulsar.NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650",
+})
+
+defer client.Close()
+
+p, err := client.CreateProducer(pulsar.ProducerOptions{
+ Topic: topicInRegex,
+ DisableBatching: true,
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer p.Close()
+
+topicsPattern := fmt.Sprintf("persistent://%s/foo.*", namespace)
+opts := pulsar.ConsumerOptions{
+ TopicsPattern: topicsPattern,
+ SubscriptionName: "regex-sub",
+}
+consumer, err := client.Subscribe(opts)
+if err != nil {
+ log.Fatal(err)
+}
+defer consumer.Close()
+
+```
+
+#### How to use multi topics Consumer
+
+```go
+
+func newTopicName() string {
+ return fmt.Sprintf("my-topic-%v", time.Now().Nanosecond())
+}
+
+
+topic1 := "topic-1"
+topic2 := "topic-2"
+
+client, err := NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650",
+})
+if err != nil {
+ log.Fatal(err)
+}
+topics := []string{topic1, topic2}
+consumer, err := client.Subscribe(pulsar.ConsumerOptions{
+ Topics: topics,
+ SubscriptionName: "multi-topic-sub",
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer consumer.Close()
+
+```
+
+#### How to use consumer listener
+
+```go
+
+import (
+ "fmt"
+ "log"
+
+ "github.com/apache/pulsar-client-go/pulsar"
+)
+
+func main() {
+ client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ defer client.Close()
+
+ channel := make(chan pulsar.ConsumerMessage, 100)
+
+ options := pulsar.ConsumerOptions{
+ Topic: "topic-1",
+ SubscriptionName: "my-subscription",
+ Type: pulsar.Shared,
+ }
+
+ options.MessageChannel = channel
+
+ consumer, err := client.Subscribe(options)
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ defer consumer.Close()
+
+ // Receive messages from channel. The channel returns a struct which contains message and the consumer from where
+ // the message was received. It's not necessary here since we have 1 single consumer, but the channel could be
+ // shared across multiple consumers as well
+ for cm := range channel {
+ msg := cm.Message
+ fmt.Printf("Received message msgId: %v -- content: '%s'\n",
+ msg.ID(), string(msg.Payload()))
+
+ consumer.Ack(msg)
+ }
+}
+
+```
+
+#### How to use consumer receive timeout
+
+```go
+
+client, err := NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650",
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer client.Close()
+
+topic := "test-topic-with-no-messages"
+ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+defer cancel()
+
+// create consumer
+consumer, err := client.Subscribe(pulsar.ConsumerOptions{
+ Topic: topic,
+ SubscriptionName: "my-sub1",
+ Type: Shared,
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer consumer.Close()
+
+msg, err := consumer.Receive(ctx)
+fmt.Println(msg.Payload())
+if err != nil {
+ log.Fatal(err)
+}
+
+```
+
+#### How to use schema in consumer
+
+```go
+
+type testJSON struct {
+ ID int `json:"id"`
+ Name string `json:"name"`
+}
+
+```
+
+```go
+
+var (
+ exampleSchemaDef = "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\"," +
+ "\"fields\":[{\"name\":\"ID\",\"type\":\"int\"},{\"name\":\"Name\",\"type\":\"string\"}]}"
+)
+
+```
+
+```go
+
+client, err := NewClient(pulsar.ClientOptions{
+ URL: "pulsar://localhost:6650",
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer client.Close()
+
+var s testJSON
+
+consumerJS := NewJSONSchema(exampleSchemaDef, nil)
+consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: "jsonTopic",
+ SubscriptionName: "sub-1",
+ Schema: consumerJS,
+ SubscriptionInitialPosition: SubscriptionPositionEarliest,
+})
+assert.Nil(t, err)
+msg, err := consumer.Receive(context.Background())
+assert.Nil(t, err)
+err = msg.GetSchemaValue(&s)
+if err != nil {
+ log.Fatal(err)
+}
+
+defer consumer.Close()
+
+```
+
+### Consumer configuration
+
+ Name | Description | Default
+| :-------- | :---------- |:---------- |
+| Topic | Topic specify the topic this consumer will subscribe to. This argument is required when constructing the reader. | |
+| Topics | Specify a list of topics this consumer will subscribe on. Either a topic, a list of topics or a topics pattern are required when subscribing| |
+| TopicsPattern | Specify a regular expression to subscribe to multiple topics under the same namespace. Either a topic, a list of topics or a topics pattern are required when subscribing | |
+| AutoDiscoveryPeriod | Specify the interval in which to poll for new partitions or new topics if using a TopicsPattern. | |
+| SubscriptionName | Specify the subscription name for this consumer. This argument is required when subscribing | |
+| Name | Set the consumer name | |
+| Properties | Properties attach a set of application defined properties to the producer This properties will be visible in the topic stats | |
+| Type | Select the subscription type to be used when subscribing to the topic. | Exclusive |
+| SubscriptionInitialPosition | InitialPosition at which the cursor will be set when subscribe | Latest |
+| DLQ | Configuration for Dead Letter Queue consumer policy. | no DLQ |
+| MessageChannel | Sets a `MessageChannel` for the consumer. When a message is received, it will be pushed to the channel for consumption | |
+| ReceiverQueueSize | Sets the size of the consumer receive queue. | 1000|
+| NackRedeliveryDelay | The delay after which to redeliver the messages that failed to be processed | 1min |
+| ReadCompacted | If enabled, the consumer will read messages from the compacted topic rather than reading the full message backlog of the topic | false |
+| ReplicateSubscriptionState | Mark the subscription as replicated to keep it in sync across clusters | false |
+| KeySharedPolicy | Configuration for Key Shared consumer policy. | |
+| RetryEnable | Auto retry send messages to default filled DLQPolicy topics | false |
+| Interceptors | A chain of interceptors. These interceptors are called at some points defined in the `ConsumerInterceptor` interface. | |
+| MaxReconnectToBroker | MaxReconnectToBroker set the maximum retry number of reconnectToBroker. | ultimate |
+| Schema | Schema set a custom schema type by passing an implementation of `Schema` | bytes[] |
+
+## Readers
+
+Pulsar readers process messages from Pulsar topics. Readers are different from consumers because with readers you need to explicitly specify which message in the stream you want to begin with (consumers, on the other hand, automatically begin with the most recent unacked message). You can [configure](#reader-configuration) Go readers using a `ReaderOptions` object. Here's an example:
+
+```go
+
+reader, err := client.CreateReader(pulsar.ReaderOptions{
+ Topic: "topic-1",
+ StartMessageID: pulsar.EarliestMessageID(),
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer reader.Close()
+
+```
+
+### Reader operations
+
+Pulsar Go readers have the following methods available:
+
+Method | Description | Return type
+:------|:------------|:-----------
+`Topic()` | Returns the reader's [topic](reference-terminology.md#topic) | `string`
+`Next(context.Context)` | Receives the next message on the topic (analogous to the `Receive` method for [consumers](#consumer-operations)). This method blocks until a message is available. | `(Message, error)`
+`HasNext()` | Check if there is any message available to read from the current position| (bool, error)
+`Close()` | Closes the reader, disabling its ability to receive messages from the broker | `error`
+`Seek(MessageID)` | Reset the subscription associated with this reader to a specific message ID | `error`
+`SeekByTime(time time.Time)` | Reset the subscription associated with this reader to a specific message publish time | `error`
+
+### Reader example
+
+#### How to use reader to read 'next' message
+
+Here's an example usage of a Go reader that uses the `Next()` method to process incoming messages:
+
+```go
+
+import (
+ "context"
+ "fmt"
+ "log"
+
+ "github.com/apache/pulsar-client-go/pulsar"
+)
+
+func main() {
+ client, err := pulsar.NewClient(pulsar.ClientOptions{URL: "pulsar://localhost:6650"})
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ defer client.Close()
+
+ reader, err := client.CreateReader(pulsar.ReaderOptions{
+ Topic: "topic-1",
+ StartMessageID: pulsar.EarliestMessageID(),
+ })
+ if err != nil {
+ log.Fatal(err)
+ }
+ defer reader.Close()
+
+ for reader.HasNext() {
+ msg, err := reader.Next(context.Background())
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
+ msg.ID(), string(msg.Payload()))
+ }
+}
+
+```
+
+In the example above, the reader begins reading from the earliest available message (specified by `pulsar.EarliestMessage`). The reader can also begin reading from the latest message (`pulsar.LatestMessage`) or some other message ID specified by bytes using the `DeserializeMessageID` function, which takes a byte array and returns a `MessageID` object. Here's an example:
+
+```go
+
+lastSavedId := // Read last saved message id from external store as byte[]
+
+reader, err := client.CreateReader(pulsar.ReaderOptions{
+ Topic: "my-golang-topic",
+ StartMessageID: pulsar.DeserializeMessageID(lastSavedId),
+})
+
+```
+
+#### How to use reader to read specific message
+
+```go
+
+client, err := NewClient(pulsar.ClientOptions{
+ URL: lookupURL,
+})
+
+if err != nil {
+ log.Fatal(err)
+}
+defer client.Close()
+
+topic := "topic-1"
+ctx := context.Background()
+
+// create producer
+producer, err := client.CreateProducer(pulsar.ProducerOptions{
+ Topic: topic,
+ DisableBatching: true,
+})
+if err != nil {
+ log.Fatal(err)
+}
+defer producer.Close()
+
+// send 10 messages
+msgIDs := [10]MessageID{}
+for i := 0; i < 10; i++ {
+ msgID, err := producer.Send(ctx, &pulsar.ProducerMessage{
+ Payload: []byte(fmt.Sprintf("hello-%d", i)),
+ })
+ assert.NoError(t, err)
+ assert.NotNil(t, msgID)
+ msgIDs[i] = msgID
+}
+
+// create reader on 5th message (not included)
+reader, err := client.CreateReader(pulsar.ReaderOptions{
+ Topic: topic,
+ StartMessageID: msgIDs[4],
+})
+
+if err != nil {
+ log.Fatal(err)
+}
+defer reader.Close()
+
+// receive the remaining 5 messages
+for i := 5; i < 10; i++ {
+ msg, err := reader.Next(context.Background())
+ if err != nil {
+ log.Fatal(err)
+}
+
+// create reader on 5th message (included)
+readerInclusive, err := client.CreateReader(pulsar.ReaderOptions{
+ Topic: topic,
+ StartMessageID: msgIDs[4],
+ StartMessageIDInclusive: true,
+})
+
+if err != nil {
+ log.Fatal(err)
+}
+defer readerInclusive.Close()
+
+```
+
+### Reader configuration
+
+ Name | Description | Default
+| :-------- | :---------- |:---------- |
+| Topic | Topic specify the topic this consumer will subscribe to. This argument is required when constructing the reader. | |
+| Name | Name set the reader name. | |
+| Properties | Attach a set of application defined properties to the reader. This properties will be visible in the topic stats | |
+| StartMessageID | StartMessageID initial reader positioning is done by specifying a message id. | |
+| StartMessageIDInclusive | If true, the reader will start at the `StartMessageID`, included. Default is `false` and the reader will start from the "next" message | false |
+| MessageChannel | MessageChannel sets a `MessageChannel` for the consumer When a message is received, it will be pushed to the channel for consumption| |
+| ReceiverQueueSize | ReceiverQueueSize sets the size of the consumer receive queue. | 1000 |
+| SubscriptionRolePrefix| SubscriptionRolePrefix set the subscription role prefix. | “reader” |
+| ReadCompacted | If enabled, the reader will read messages from the compacted topic rather than reading the full message backlog of the topic. ReadCompacted can only be enabled when reading from a persistent topic. | false|
+
+## Messages
+
+The Pulsar Go client provides a `ProducerMessage` interface that you can use to construct messages to producer on Pulsar topics. Here's an example message:
+
+```go
+
+msg := pulsar.ProducerMessage{
+ Payload: []byte("Here is some message data"),
+ Key: "message-key",
+ Properties: map[string]string{
+ "foo": "bar",
+ },
+ EventTime: time.Now(),
+ ReplicationClusters: []string{"cluster1", "cluster3"},
+}
+
+if _, err := producer.send(msg); err != nil {
+ log.Fatalf("Could not publish message due to: %v", err)
+}
+
+```
+
+The following methods parameters are available for `ProducerMessage` objects:
+
+Parameter | Description
+:---------|:-----------
+`Payload` | The actual data payload of the message
+`Value` | Value and payload is mutually exclusive, `Value interface{}` for schema message.
+`Key` | The optional key associated with the message (particularly useful for things like topic compaction)
+`OrderingKey` | OrderingKey sets the ordering key of the message.
+`Properties` | A key-value map (both keys and values must be strings) for any application-specific metadata attached to the message
+`EventTime` | The timestamp associated with the message
+`ReplicationClusters` | The clusters to which this message will be replicated. Pulsar brokers handle message replication automatically; you should only change this setting if you want to override the broker default.
+`SequenceID` | Set the sequence id to assign to the current message
+`DeliverAfter` | Request to deliver the message only after the specified relative delay
+`DeliverAt` | Deliver the message only at or after the specified absolute timestamp
+
+## TLS encryption and authentication
+
+In order to use [TLS encryption](security-tls-transport), you'll need to configure your client to do so:
+
+ * Use `pulsar+ssl` URL type
+ * Set `TLSTrustCertsFilePath` to the path to the TLS certs used by your client and the Pulsar broker
+ * Configure `Authentication` option
+
+Here's an example:
+
+```go
+
+opts := pulsar.ClientOptions{
+ URL: "pulsar+ssl://my-cluster.com:6651",
+ TLSTrustCertsFilePath: "/path/to/certs/my-cert.csr",
+ Authentication: NewAuthenticationTLS("my-cert.pem", "my-key.pem"),
+}
+
+```
+
+## OAuth2 authentication
+
+To use [OAuth2 authentication](security-oauth2), 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",
+ "audience": "https://dev-kt-aa9ne.us.auth0.com/api/v2/",
+ "privateKey": "/path/to/privateKey",
+ "clientId": "0Xx...Yyxeny",
+ })
+client, err := pulsar.NewClient(pulsar.ClientOptions{
+ URL: "pulsar://my-cluster:6650",
+ Authentication: oauth,
+})
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/client-libraries-java.md b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-java.md
new file mode 100644
index 0000000..cf0b0ea
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-java.md
@@ -0,0 +1,809 @@
+---
+id: client-libraries-java
+title: The Pulsar Java client
+sidebar_label: "Java"
+original_id: client-libraries-java
+---
+
+The Pulsar Java client can be used both to create Java producers, consumers, and [readers](#reader-interface) of messages and to perform [administrative tasks](admin-api-overview). The current version of the Java client is **@pulsar:version@**.
+
+Javadoc for the Pulsar client is divided up into two domains, by package:
+
+Package | Description | Maven Artifact
+:-------|:------------|:--------------
+[`org.apache.pulsar.client.api`](/api/client) | The producer and consumer API | [org.apache.pulsar:pulsar-client:@pulsar:version@](http://search.maven.org/#artifactdetails%7Corg.apache.pulsar%7Cpulsar-client%7C@pulsar:version@%7Cjar)
+[`org.apache.pulsar.client.admin`](/api/admin) | The Java [admin API](admin-api-overview) | [org.apache.pulsar:pulsar-client-admin:@pulsar:version@](http://search.maven.org/#artifactdetails%7Corg.apache.pulsar%7Cpulsar-client-admin%7C@pulsar:version@%7Cjar)
+
+This document will focus only on the client API for producing and consuming messages on Pulsar topics. For a guide to using the Java admin client, see [The Pulsar admin interface](admin-api-overview).
+
+## Installation
+
+The latest version of the Pulsar Java client library is available via [Maven Central](http://search.maven.org/#artifactdetails%7Corg.apache.pulsar%7Cpulsar-client%7C@pulsar:version@%7Cjar). To use the latest version, add the `pulsar-client` library to your build configuration.
+
+### Maven
+
+If you're using Maven, add this to your `pom.xml`:
+
+```xml
+
+<!-- in your <properties> block -->
+<pulsar.version>@pulsar:version@</pulsar.version>
+
+<!-- in your <dependencies> block -->
+<dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client</artifactId>
+ <version>${pulsar.version}</version>
+</dependency>
+
+```
+
+### Gradle
+
+If you're using Gradle, add this to your `build.gradle` file:
+
+```groovy
+
+def pulsarVersion = '@pulsar:version@'
+
+dependencies {
+ compile group: 'org.apache.pulsar', name: 'pulsar-client', version: pulsarVersion
+}
+
+```
+
+## Connection URLs
+
+To connect to Pulsar using client libraries, you need to specify a [Pulsar protocol](developing-binary-protocol) URL.
+
+Pulsar protocol URLs are assigned to specific clusters, use the `pulsar` scheme and have a default port of 6650. Here's an example for `localhost`:
+
+```http
+
+pulsar://localhost:6650
+
+```
+
+If you have more than one broker, the URL may look like this:
+
+```http
+
+pulsar://localhost:6550,localhost:6651,localhost:6652
+
+```
+
+A URL for a production Pulsar cluster may look something like this:
+
+```http
+
+pulsar://pulsar.us-west.example.com:6650
+
+```
+
+If you're using [TLS](security-tls-authentication) authentication, the URL will look like something like this:
+
+```http
+
+pulsar+ssl://pulsar.us-west.example.com:6651
+
+```
+
+## Client configuration
+
+You can instantiate a {@inject: javadoc:PulsarClient:/client/org/apache/pulsar/client/api/PulsarClient} object using just a URL for the target Pulsar [cluster](reference-terminology.md#cluster), like this:
+
+```java
+
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650")
+ .build();
+
+```
+
+If you have multiple brokers, you can initiate a PulsarClient like this:
+
+```java
+
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar://localhost:6650,localhost:6651,localhost:6652")
+ .build();
+
+```
+
+> #### Default broker URLs for standalone clusters
+> If you're running a cluster in [standalone mode](getting-started-standalone), the broker will be available at the `pulsar://localhost:6650` URL by default.
+
+If you create a client, you may use the `loadConf` configuration. Below are the available parameters used in `loadConf`.
+
+| Type | Name | Description | Default
+|---|---|---|---
+String | `serviceUrl` |Service URL provider for Pulsar service | None
+String | `authPluginClassName` | Name of the authentication plugin | None
+String | `authParams` | String represents parameters for the authentication plugin <br />**Example**<br /> key1:val1,key2:val2|None
+long|`operationTimeoutMs`|Operation timeout |30000
+long|`statsIntervalSeconds`|Interval between each stat info<br />Stats is activated with positive `statsInterval`<br />`statsIntervalSeconds` should be set to 1 second at least |60
+int|`numIoThreads`| Number of threads used for handling connections to brokers | 1
+int|`numListenerThreads`|Number of threads used for handling message listeners | 1
+boolean|`useTcpNoDelay`|Whether to use TCP no-delay flag on the connection to disable Nagle algorithm |true
+boolean |`useTls` |Whether to use TLS encryption on the connection| false
+string | `tlsTrustCertsFilePath` |Path to the trusted TLS certificate file|None
+boolean|`tlsAllowInsecureConnection`|Whether the Pulsar client accepts untrusted TLS certificate from broker | false
+boolean | `tlsHostnameVerificationEnable` | Whether to enable TLS hostname verification|false
+int|`concurrentLookupRequest`|Number of concurrent lookup requests allowed to send on each broker connection to prevent overload on broker|5000
+int|`maxLookupRequest`|Maximum number of lookup requests allowed on each broker connection to prevent overload on broker | 50000
+int|`maxNumberOfRejectedRequestPerConnection`|Maximum number of rejected requests of a broker in a certain time frame (30 seconds) after the current connection is closed and the client creates a new connection to connect to a different broker|50
+int|`keepAliveIntervalSeconds`|Seconds of keeping alive interval for each client broker connection|30
+int|`connectionTimeoutMs`|Duration of waiting for a connection to a broker to be established <br />If the duration passes without a response from a broker, the connection attempt is dropped|10000
+int|`requestTimeoutMs`|Maximum duration for completing a request |60000
+int|`defaultBackoffIntervalNanos`| Default duration for a backoff interval | TimeUnit.MILLISECONDS.toNanos(100);
+long|`maxBackoffIntervalNanos`|Maximum duration for a backoff interval|TimeUnit.SECONDS.toNanos(30)
+
+Check out the Javadoc for the {@inject: javadoc:PulsarClient:/client/org/apache/pulsar/client/api/PulsarClient} class for a full listing of configurable parameters.
+
+> In addition to client-level configuration, you can also apply [producer](#configuring-producers) and [consumer](#configuring-consumers) specific configuration, as you'll see in the sections below.
+
+## Producers
+
+In Pulsar, producers write messages to topics. Once you've instantiated a {@inject: javadoc:PulsarClient:/client/org/apache/pulsar/client/api/PulsarClient} object (as in the section [above](#client-configuration)), you can create a {@inject: javadoc:Producer:/client/org/apache/pulsar/client/api/Producer} for a specific Pulsar [topic](reference-terminology.md#topic).
+
+```java
+
+Producer<byte[]> producer = client.newProducer()
+ .topic("my-topic")
+ .create();
+
+// You can then send messages to the broker and topic you specified:
+producer.send("My message".getBytes());
+
+```
+
+By default, producers produce messages that consist of byte arrays. You can produce different types, however, by specifying a message [schema](#schemas).
+
+```java
+
+Producer<String> stringProducer = client.newProducer(Schema.STRING)
+ .topic("my-topic")
+ .create();
+stringProducer.send("My message");
+
+```
+
+> You should always make sure to close your producers, consumers, and clients when they are no longer needed:
+
+> ```java
+>
+> producer.close();
+> consumer.close();
+> client.close();
+>
+>
+> ```
+
+>
+> Close operations can also be asynchronous:
+
+> ```java
+>
+> producer.closeAsync()
+> .thenRun(() -> System.out.println("Producer closed"));
+> .exceptionally((ex) -> {
+> System.err.println("Failed to close producer: " + ex);
+> return ex;
+> });
+>
+>
+> ```
+
+
+### Configuring producers
+
+If you instantiate a `Producer` object specifying only a topic name, as in the example above, the producer will use the default configuration. To use a non-default configuration, there's a variety of configurable parameters that you can set.
+
+For a full listing, see the Javadoc for the {@inject: javadoc:ProducerBuilder:/client/org/apache/pulsar/client/api/ProducerBuilder} class. Here's an example:
+
+```java
+
+Producer<byte[]> producer = client.newProducer()
+ .topic("my-topic")
+ .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
+ .sendTimeout(10, TimeUnit.SECONDS)
+ .blockIfQueueFull(true)
+ .create();
+
+```
+
+### Message routing
+
+When using partitioned topics, you can specify the routing mode whenever you publish messages using a producer. For more on specifying a routing mode using the Java client, see the [Partitioned Topics](cookbooks-partitioned) cookbook.
+
+### Async send
+
+You can also publish messages [asynchronously](concepts-messaging.md#send-modes) using the Java client. With 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), the producer could be blocked or fail immediately when calling the API, depending on arguments passed to the producer.
+
+Here's an example async send operation:
+
+```java
+
+producer.sendAsync("my-async-message".getBytes()).thenAccept(msgId -> {
+ System.out.printf("Message with ID %s successfully sent", msgId);
+});
+
+```
+
+As you can see from the example above, async send operations return a {@inject: javadoc:MessageId:/client/org/apache/pulsar/client/api/MessageId} wrapped in a [`CompletableFuture`](http://www.baeldung.com/java-completablefuture).
+
+### Configuring messages
+
+In addition to a value, it's possible to set additional items on a given message:
+
+```java
+
+producer.newMessage()
+ .key("my-message-key")
+ .value("my-async-message".getBytes())
+ .property("my-key", "my-value")
+ .property("my-other-key", "my-other-value")
+ .send();
+
+```
+
+As for the previous case, it's also possible to terminate the builder chain with `sendAsync()` and
+get a future returned.
+
+## Consumers
+
+In Pulsar, consumers subscribe to topics and handle messages that producers publish to those topics. You can instantiate a new [consumer](reference-terminology.md#consumer) by first instantiating a {@inject: javadoc:PulsarClient:/client/org/apache/pulsar/client/api/PulsarClient} object and passing it a URL for a Pulsar broker (as [above](#client-configuration)).
+
+Once you've instantiated a {@inject: javadoc:PulsarClient:/client/org/apache/pulsar/client/api/PulsarClient} object, you can create a {@inject: javadoc:Consumer:/client/org/apache/pulsar/client/api/Consumer} by specifying a [topic](reference-terminology.md#topic) and a [subscription](concepts-messaging.md#subscription-modes).
+
+```java
+
+Consumer consumer = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscribe();
+
+```
+
+The `subscribe` method will automatically subscribe the consumer to the specified topic and subscription. One way to make the consumer listen on the topic is to set up a `while` loop. In this example loop, the consumer listens for messages, prints the contents of any message that's received, and then [acknowledges](reference-terminology.md#acknowledgment-ack) that the message has been processed. If the processing logic fails, we use [negative acknowledgement](reference-terminology.md#ack [...]
+to have the message redelivered at a later point in time.
+
+```java
+
+while (true) {
+ // Wait for a message
+ Message msg = consumer.receive();
+
+ try {
+ // Do something with the message
+ System.out.printf("Message received: %s", new String(msg.getData()));
+
+ // Acknowledge the message so that it can be deleted by the message broker
+ consumer.acknowledge(msg);
+ } catch (Exception e) {
+ // Message failed to process, redeliver later
+ consumer.negativeAcknowledge(msg);
+ }
+}
+
+```
+
+### Configuring consumers
+
+If you instantiate a `Consumer` object specifying only a topic and subscription name, as in the example above, the consumer will use the default configuration. To use a non-default configuration, there's a variety of configurable parameters that you can set. For a full listing, see the Javadoc for the {@inject: javadoc:ConsumerBuilder:/client/org/apache/pulsar/client/api/ConsumerBuilder} class. Here's an example:
+
+Here's an example configuration:
+
+```java
+
+Consumer consumer = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .ackTimeout(10, TimeUnit.SECONDS)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribe();
+
+```
+
+### Async receive
+
+The `receive` method will receive messages synchronously (the consumer process will be blocked until a message is available). You can also use [async receive](concepts-messaging.md#receive-modes), which will return immediately with a [`CompletableFuture`](http://www.baeldung.com/java-completablefuture) object that completes once a new message is available.
+
+Here's an example:
+
+```java
+
+CompletableFuture<Message> asyncMessage = consumer.receiveAsync();
+
+```
+
+Async receive operations return a {@inject: javadoc:Message:/client/org/apache/pulsar/client/api/Message} wrapped inside of a [`CompletableFuture`](http://www.baeldung.com/java-completablefuture).
+
+### Multi-topic subscriptions
+
+In addition to subscribing a consumer to a single Pulsar topic, you can also subscribe to multiple topics simultaneously using [multi-topic subscriptions](concepts-messaging.md#multi-topic-subscriptions). To use multi-topic subscriptions you can supply either a regular expression (regex) or a `List` of topics. If you select topics via regex, all topics must be within the same Pulsar namespace.
+
+Here are some examples:
+
+```java
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.PulsarClient;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.regex.Pattern;
+
+ConsumerBuilder consumerBuilder = pulsarClient.newConsumer()
+ .subscriptionName(subscription);
+
+// Subscribe to all topics in a namespace
+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("persistent://public/default/foo.*");
+Consumer allTopicsConsumer = consumerBuilder
+ .topicsPattern(someTopicsInNamespace)
+ .subscribe();
+
+```
+
+You can also subscribe to an explicit list of topics (across namespaces if you wish):
+
+```java
+
+List<String> topics = Arrays.asList(
+ "topic-1",
+ "topic-2",
+ "topic-3"
+);
+
+Consumer multiTopicConsumer = consumerBuilder
+ .topics(topics)
+ .subscribe();
+
+// Alternatively:
+Consumer multiTopicConsumer = consumerBuilder
+ .topics(
+ "topic-1",
+ "topic-2",
+ "topic-3"
+ )
+ .subscribe();
+
+```
+
+You can also subscribe to multiple topics asynchronously using the `subscribeAsync` method rather than the synchronous `subscribe` method. Here's an example:
+
+```java
+
+Pattern allTopicsInNamespace = Pattern.compile("persistent://public/default.*");
+consumerBuilder
+ .topics(topics)
+ .subscribeAsync()
+ .thenAccept(this::receiveMessageFromConsumer);
+
+private void receiveMessageFromConsumer(Consumer consumer) {
+ consumer.receiveAsync().thenAccept(message -> {
+ // Do something with the received message
+ receiveMessageFromConsumer(consumer);
+ });
+}
+
+```
+
+### Subscription types
+
+Pulsar has various [subscription types](concepts-messaging#subscription-types) to match different scenarios. A topic can have multiple subscriptions with different subscription types. However, a subscription can only have one subscription type at a time.
+
+A subscription is identified with the subscription name; a subscription name can specify only one subscription type at a time. To change the subscription type, you should first stop all consumers of this subscription.
+
+Different subscription types have different message distribution modes. This section describes the differences of subscription types and how to use them.
+
+In order to better describe their differences, assuming you have a topic named "my-topic", and the producer has published 10 messages.
+
+```java
+
+Producer<String> producer = client.newProducer(Schema.STRING)
+ .topic("my-topic")
+ .enableBatching(false)
+ .create();
+// 3 messages with "key-1", 3 messages with "key-2", 2 messages with "key-3" and 2 messages with "key-4"
+producer.newMessage().key("key-1").value("message-1-1").send();
+producer.newMessage().key("key-1").value("message-1-2").send();
+producer.newMessage().key("key-1").value("message-1-3").send();
+producer.newMessage().key("key-2").value("message-2-1").send();
+producer.newMessage().key("key-2").value("message-2-2").send();
+producer.newMessage().key("key-2").value("message-2-3").send();
+producer.newMessage().key("key-3").value("message-3-1").send();
+producer.newMessage().key("key-3").value("message-3-2").send();
+producer.newMessage().key("key-4").value("message-4-1").send();
+producer.newMessage().key("key-4").value("message-4-2").send();
+
+```
+
+#### Exclusive
+
+Create a new consumer and subscribe with the `Exclusive` subscription type.
+
+```java
+
+Consumer consumer = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscribe()
+
+```
+
+Only the first consumer is allowed to the subscription, other consumers receive an error. The first consumer receives all 10 messages, and the consuming order is the same as the producing order.
+
+:::note
+
+If topic is a partitioned topic, the first consumer subscribes to all partitioned topics, other consumers are not assigned with partitions and receive an error.
+
+:::
+
+#### Failover
+
+Create new consumers and subscribe with the`Failover` subscription type.
+
+```java
+
+Consumer consumer1 = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Failover)
+ .subscribe()
+Consumer consumer2 = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Failover)
+ .subscribe()
+//conumser1 is the active consumer, consumer2 is the standby consumer.
+//consumer1 receives 5 messages and then crashes, consumer2 takes over as an active consumer.
+
+```
+
+Multiple consumers can attach to the same subscription, yet only the first consumer is active, and others are standby. When the active consumer is disconnected, messages will be dispatched to one of standby consumers, and the standby consumer becomes active consumer.
+
+If the first active consumer receives 5 messages and is disconnected, the standby consumer becomes active consumer. Consumer1 will receive:
+
+```
+
+("key-1", "message-1-1")
+("key-1", "message-1-2")
+("key-1", "message-1-3")
+("key-2", "message-2-1")
+("key-2", "message-2-2")
+
+```
+
+consumer2 will receive:
+
+```
+
+("key-2", "message-2-3")
+("key-3", "message-3-1")
+("key-3", "message-3-2")
+("key-4", "message-4-1")
+("key-4", "message-4-2")
+
+```
+
+:::note
+
+If a topic is a partitioned topic, each partition only has one active consumer, messages of one partition only distributed to one consumer, messages of multiple partitions are distributed to multiple consumers.
+
+:::
+
+#### Shared
+
+Create new consumers and subscribe with `Shared` subscription type.
+
+```java
+
+Consumer consumer1 = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Shared)
+ .subscribe()
+
+Consumer consumer2 = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Shared)
+ .subscribe()
+//Both consumer1 and consumer 2 is active consumers.
+
+```
+
+In shared subscription type, multiple consumers can attach to the same subscription and message are delivered in a round robin distribution across consumers.
+
+If a broker dispatches only one message at a time, consumer1 will receive:
+
+```
+
+("key-1", "message-1-1")
+("key-1", "message-1-3")
+("key-2", "message-2-2")
+("key-3", "message-3-1")
+("key-4", "message-4-1")
+
+```
+
+consumer 2 will receive:
+
+```
+
+("key-1", "message-1-2")
+("key-2", "message-2-1")
+("key-2", "message-2-3")
+("key-3", "message-3-2")
+("key-4", "message-4-2")
+
+```
+
+`Shared` subscription is different from `Exclusive` and `Failover` subscription types. `Shared` subscription has better flexibility, but cannot provide order guarantee.
+
+#### Key_Shared
+
+This is a new subscription type since 2.4.0 release, create new consumers and subscribe with `Key_Shared` subscription type:
+
+```java
+
+Consumer consumer1 = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Key_Shared)
+ .subscribe()
+
+Consumer consumer2 = client.newConsumer()
+ .topic("my-topic")
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Key_Shared)
+ .subscribe()
+//Both consumer1 and consumer2 are active consumers.
+
+```
+
+`Key_Shared` subscription is like `Shared` subscription, all consumers can attach to the same subscription. But it is different from `Key_Shared` subscription, messages with the same key are delivered to only one consumer in order. The possible distribution of messages between different consumers(by default we do not know in advance which keys will be assigned to a consumer, but a key will only be assigned to a consumer at the same time. ) .
+
+consumer1 will receive:
+
+```
+
+("key-1", "message-1-1")
+("key-1", "message-1-2")
+("key-1", "message-1-3")
+("key-3", "message-3-1")
+("key-3", "message-3-2")
+
+```
+
+consumer 2 will receive:
+
+```
+
+("key-2", "message-2-1")
+("key-2", "message-2-2")
+("key-2", "message-2-3")
+("key-4", "message-4-1")
+("key-4", "message-4-2")
+
+```
+
+:::note
+
+If the message key is not specified, messages without key will be dispatched to one consumer in order by default.
+
+:::
+
+## Reader interface
+
+With the [reader interface](concepts-clients.md#reader-interface), Pulsar clients can "manually position" themselves within a topic, 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}.
+
+Here's an example:
+
+```java
+
+byte[] msgIdBytes = // Some message ID byte array
+MessageId id = MessageId.fromByteArray(msgIdBytes);
+Reader reader = pulsarClient.newReader()
+ .topic(topic)
+ .startMessageId(id)
+ .create();
+
+while (true) {
+ Message message = reader.readNext();
+ // Process message
+}
+
+```
+
+In the example above, a `Reader` object is instantiated for a specific topic and message (by ID); the reader then iterates over each message in the topic after the message identified by `msgIdBytes` (how that value is obtained depends on the application).
+
+The code sample above shows pointing the `Reader` object to a specific message (by ID), but you can also use `MessageId.earliest` to point to the earliest available message on the topic of `MessageId.latest` to point to the most recent available message.
+
+## Schemas
+
+In Pulsar, all message data consists of byte arrays "under the hood." [Message schemas](schema-get-started) enable you to use other types of data when constructing and handling messages (from simple types like strings to more complex, application-specific types). If you construct, say, a [producer](#producers) without specifying a schema, then the producer can only produce messages of type `byte[]`. Here's an example:
+
+```java
+
+Producer<byte[]> producer = client.newProducer()
+ .topic(topic)
+ .create();
+
+```
+
+The producer above is equivalent to a `Producer<byte[]>` (in fact, you should *always* explicitly specify the type). If you'd like to use a producer for a different type of data, you'll need to specify a **schema** that informs Pulsar which data type will be transmitted over the [topic](reference-terminology.md#topic).
+
+### Schema example
+
+Let's say that you have a `SensorReading` class that you'd like to transmit over a Pulsar topic:
+
+```java
+
+public class SensorReading {
+ public float temperature;
+
+ public SensorReading(float temperature) {
+ this.temperature = temperature;
+ }
+
+ // A no-arg constructor is required
+ public SensorReading() {
+ }
+
+ public float getTemperature() {
+ return temperature;
+ }
+
+ public void setTemperature(float temperature) {
+ this.temperature = temperature;
+ }
+}
+
+```
+
+You could then create a `Producer<SensorReading>` (or `Consumer<SensorReading>`) like so:
+
+```java
+
+Producer<SensorReading> producer = client.newProducer(JSONSchema.of(SensorReading.class))
+ .topic("sensor-readings")
+ .create();
+
+```
+
+The following schema formats are currently available for Java:
+
+* No schema or the byte array schema (which can be applied using `Schema.BYTES`):
+
+ ```java
+
+ Producer<byte[]> bytesProducer = client.newProducer(Schema.BYTES)
+ .topic("some-raw-bytes-topic")
+ .create();
+
+ ```
+
+ Or, equivalently:
+
+ ```java
+
+ Producer<byte[]> bytesProducer = client.newProducer()
+ .topic("some-raw-bytes-topic")
+ .create();
+
+ ```
+
+* `String` for normal UTF-8-encoded string data. This schema can be applied using `Schema.STRING`:
+
+ ```java
+
+ Producer<String> stringProducer = client.newProducer(Schema.STRING)
+ .topic("some-string-topic")
+ .create();
+
+ ```
+
+* JSON schemas can be created for POJOs using `Schema.JSON`. Here's an example:
+
+ ```java
+
+ Producer<MyPojo> pojoProducer = client.newProducer(Schema.JSON(MyPojo.class))
+ .topic("some-pojo-topic")
+ .create();
+
+ ```
+
+* Protobuf schemas can be generate using `Schema.PROTOBUF`. The following example shows how to create the Protobuf schema and use it to instantiate a new producer:
+
+ ```java
+
+ Producer<MyProtobuf> protobufProducer = client.newProducer(Schema.PROTOBUF(MyProtobuf.class))
+ .topic("some-protobuf-topic")
+ .create();
+
+ ```
+
+* Avro schemas can be defined with the help of `Schema.AVRO`. The next code snippet demonstrates the creation and usage of the Avro schema:
+
+ ```java
+
+ Producer<MyAvro> avroProducer = client.newProducer(Schema.AVRO(MyAvro.class))
+ .topic("some-avro-topic")
+ .create();
+
+ ```
+
+## Authentication
+
+Pulsar currently supports two authentication schemes: [TLS](security-tls-authentication.md) and [Athenz](security-athenz). The Pulsar Java client can be used with both.
+
+### TLS Authentication
+
+To use [TLS](security-tls-authentication), you need to set TLS to `true` using the `setUseTls` method, point your Pulsar client to a TLS cert path, and provide paths to cert and key files.
+
+Here's an example configuration:
+
+```java
+
+Map<String, String> authParams = new HashMap();
+authParams.put("tlsCertFile", "/path/to/client-cert.pem");
+authParams.put("tlsKeyFile", "/path/to/client-key.pem");
+
+Authentication tlsAuth = AuthenticationFactory
+ .create(AuthenticationTls.class.getName(), authParams);
+
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar+ssl://my-broker.com:6651")
+ .enableTls(true)
+ .tlsTrustCertsFilePath("/path/to/cacert.pem")
+ .authentication(tlsAuth)
+ .build();
+
+```
+
+### Athenz
+
+To use [Athenz](security-athenz) as an authentication provider, you need to [use TLS](#tls-authentication) and provide values for four parameters in a hash:
+
+* `tenantDomain`
+* `tenantService`
+* `providerDomain`
+* `privateKey`
+
+You can also set an optional `keyId`. Here's an example configuration:
+
+```java
+
+Map<String, String> authParams = new HashMap();
+authParams.put("tenantDomain", "shopping"); // Tenant domain name
+authParams.put("tenantService", "some_app"); // Tenant service name
+authParams.put("providerDomain", "pulsar"); // Provider domain name
+authParams.put("privateKey", "file:///path/to/private.pem"); // Tenant private key path
+authParams.put("keyId", "v1"); // Key id for the tenant private key (optional, default: "0")
+
+Authentication athenzAuth = AuthenticationFactory
+ .create(AuthenticationAthenz.class.getName(), authParams);
+
+PulsarClient client = PulsarClient.builder()
+ .serviceUrl("pulsar+ssl://my-broker.com:6651")
+ .enableTls(true)
+ .tlsTrustCertsFilePath("/path/to/cacert.pem")
+ .authentication(athenzAuth)
+ .build();
+
+```
+
+> #### Supported pattern formats
+> The `privateKey` parameter supports the following three pattern formats:
+> * `file:///path/to/file`
+> * `file:/path/to/file`
+> * `data:application/x-pem-file;base64,<base64-encoded value>`
diff --git a/site2/website-next/versioned_docs/version-2.4.1/client-libraries-python.md b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-python.md
new file mode 100644
index 0000000..1c22f72
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-python.md
@@ -0,0 +1,311 @@
+---
+id: client-libraries-python
+title: The Pulsar Python client
+sidebar_label: "Python"
+original_id: client-libraries-python
+---
+
+The Pulsar Python client library is a wrapper over the existing [C++ client library](client-libraries-cpp) and exposes all of the [same features](/api/cpp). You can find the code in the [`python` subdirectory](https://github.com/apache/pulsar/tree/master/pulsar-client-cpp/python) of the C++ client code.
+
+## Installation
+
+You can install the [`pulsar-client`](https://pypi.python.org/pypi/pulsar-client) library either via [PyPi](https://pypi.python.org/pypi), using [pip](#installation-using-pip), or by building the library from source.
+
+### Installation using pip
+
+To install the `pulsar-client` library as a pre-built package using the [pip](https://pip.pypa.io/en/stable/) package manager:
+
+```shell
+
+$ pip install pulsar-client==@pulsar:version_number@
+
+```
+
+Installation via PyPi is available for the following Python versions:
+
+Platform | Supported Python versions
+:--------|:-------------------------
+| MacOS<br /> 10.11 (El Capitan) — 10.12 (Sierra) — 10.13 (High Sierra) — 10.14 (Mojave)<br />| 2.7, 3.7
+|Linux | 2.7, 3.4, 3.5, 3.6, 3.7
+
+### Installing from source
+
+To install the `pulsar-client` library by building from source, follow [these instructions](client-libraries-cpp.md#compilation) and compile the Pulsar C++ client library. That will also build the Python binding for the library.
+
+To install the built Python bindings:
+
+```shell
+
+$ git clone https://github.com/apache/pulsar
+$ cd pulsar/pulsar-client-cpp/python
+$ sudo python setup.py install
+
+```
+
+## API Reference
+
+The complete Python API reference is available at [api/python](/api/python).
+
+## Examples
+
+Below you'll find a variety of Python code examples for the `pulsar-client` library.
+
+### Producer example
+
+This creates a Python producer for the `my-topic` topic and send 10 messages on that topic:
+
+```python
+
+import pulsar
+
+client = pulsar.Client('pulsar://localhost:6650')
+
+producer = client.create_producer('my-topic')
+
+for i in range(10):
+ producer.send(('Hello-%d' % i).encode('utf-8'))
+
+client.close()
+
+```
+
+### Consumer example
+
+This creates a consumer with the `my-subscription` subscription on the `my-topic` topic, listen for incoming messages, print the content and ID of messages that arrive, and acknowledge each message to the Pulsar broker:
+
+```python
+
+import pulsar
+
+client = pulsar.Client('pulsar://localhost:6650')
+
+consumer = client.subscribe('my-topic', 'my-subscription')
+
+while True:
+ msg = consumer.receive()
+ try:
+ print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
+ # Acknowledge successful processing of the message
+ consumer.acknowledge(msg)
+ except:
+ # Message failed to be processed
+ consumer.negative_acknowledge(msg)
+
+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:
+
+```python
+
+# MessageId taken from a previously fetched message
+msg_id = msg.message_id()
+
+reader = client.create_reader('my-topic', msg_id)
+
+while True:
+ msg = reader.read_next()
+ print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
+ # No acknowledgment
+
+```
+
+## Schema
+
+### Declaring and validating schema
+
+A schema can be declared by passing a class that inherits
+from `pulsar.schema.Record` and defines the fields as
+class variables. For example:
+
+```python
+
+from pulsar.schema import *
+
+class Example(Record):
+ a = String()
+ b = Integer()
+ c = Boolean()
+
+```
+
+With this simple schema definition we can then create producers,
+consumers and readers instances that will be referring to that.
+
+```python
+
+producer = client.create_producer(
+ topic='my-topic',
+ schema=AvroSchema(Example) )
+
+producer.send(Example(a='Hello', b=1))
+
+```
+
+When the producer is created, the Pulsar broker will validate that
+the existing topic schema is indeed of "Avro" type and that the
+format is compatible with the schema definition of the `Example`
+class.
+
+If there is a mismatch, the producer creation will raise an
+exception.
+
+Once a producer is created with a certain schema definition,
+it will only accept objects that are instances of the declared
+schema class.
+
+Similarly, for a consumer/reader, the consumer will return an
+object, instance of the schema record class, rather than the raw
+bytes:
+
+```python
+
+consumer = client.subscribe(
+ topic='my-topic',
+ subscription_name='my-subscription',
+ schema=AvroSchema(Example) )
+
+while True:
+ msg = consumer.receive()
+ ex = msg.value()
+ try:
+ print("Received message a={} b={} c={}".format(ex.a, ex.b, ex.c))
+ # Acknowledge successful processing of the message
+ consumer.acknowledge(msg)
+ except:
+ # Message failed to be processed
+ consumer.negative_acknowledge(msg)
+
+```
+
+### Supported schema types
+
+There are different builtin schema types that can be used in Pulsar.
+All the definitions are in the `pulsar.schema` package.
+
+| Schema | Notes |
+| ------ | ----- |
+| `BytesSchema` | Get the raw payload as a `bytes` object. No serialization/deserialization are performed. This is the default schema mode |
+| `StringSchema` | Encode/decode payload as a UTF-8 string. Uses `str` objects |
+| `JsonSchema` | Require record definition. Serializes the record into standard JSON payload |
+| `AvroSchema` | Require record definition. Serializes in AVRO format |
+
+### Schema definition reference
+
+The schema definition is done through a class that inherits from
+`pulsar.schema.Record`.
+
+This class can have a number of fields which can be of either
+`pulsar.schema.Field` type or even another nested `Record`. All the
+fields are also specified in the `pulsar.schema` package. The fields
+are matching the AVRO fields types.
+
+| Field Type | Python Type | Notes |
+| ---------- | ----------- | ----- |
+| `Boolean` | `bool` | |
+| `Integer` | `int` | |
+| `Long` | `int` | |
+| `Float` | `float` | |
+| `Double` | `float` | |
+| `Bytes` | `bytes` | |
+| `String` | `str` | |
+| `Array` | `list` | Need to specify record type for items |
+| `Map` | `dict` | Key is always `String`. Need to specify value type |
+
+Additionally, any Python `Enum` type can be used as a valid field
+type
+
+#### Fields parameters
+
+When adding a field these parameters can be used in the constructor:
+
+| Argument | Default | Notes |
+| ---------- | --------| ----- |
+| `default` | `None` | Set a default value for the field. Eg: `a = Integer(default=5)` |
+| `required` | `False` | Mark the field as "required". This will set it in the schema accordingly. |
+
+#### Schema definition examples
+
+##### Simple definition
+
+```python
+
+class Example(Record):
+ a = String()
+ b = Integer()
+ c = Array(String())
+ i = Map(String())
+
+```
+
+##### Using enums
+
+```python
+
+from enum import Enum
+
+class Color(Enum):
+ red = 1
+ green = 2
+ blue = 3
+
+class Example(Record):
+ name = String()
+ color = Color
+
+```
+
+##### Complex types
+
+```python
+
+class MySubRecord(Record):
+ x = Integer()
+ y = Long()
+ z = String()
+
+class Example(Record):
+ a = String()
+ sub = MySubRecord()
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/client-libraries-websocket.md b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-websocket.md
new file mode 100644
index 0000000..bc13b43
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/client-libraries-websocket.md
@@ -0,0 +1,621 @@
+---
+id: client-libraries-websocket
+title: Pulsar WebSocket API
+sidebar_label: "WebSocket"
+original_id: client-libraries-websocket
+---
+
+Pulsar [WebSocket](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API) API provides a simple way to interact with Pulsar using languages that do not have an official [client library](getting-started-clients). Through WebSocket, you can publish and consume messages and use features available on the [Client Features Matrix](https://github.com/apache/pulsar/wiki/Client-Features-Matrix) page.
+
+
+> You can use Pulsar WebSocket API with any WebSocket client library. See examples for Python and Node.js [below](#client-examples).
+
+## Running the WebSocket service
+
+The standalone variant of Pulsar that we recommend using for [local development](getting-started-standalone) already has the WebSocket service enabled.
+
+In non-standalone mode, there are two ways to deploy the WebSocket service:
+
+* [embedded](#embedded-with-a-pulsar-broker) with a Pulsar broker
+* as a [separate component](#as-a-separate-component)
+
+### Embedded with a Pulsar broker
+
+In this mode, the WebSocket service will run within the same HTTP service that's already running in the broker. To enable this mode, set the [`webSocketServiceEnabled`](reference-configuration.md#broker-webSocketServiceEnabled) parameter in the [`conf/broker.conf`](reference-configuration.md#broker) configuration file in your installation.
+
+```properties
+
+webSocketServiceEnabled=true
+
+```
+
+### As a separate component
+
+In this mode, the WebSocket service will be run from a Pulsar [broker](reference-terminology.md#broker) as a separate service. Configuration for this mode is handled in the [`conf/websocket.conf`](reference-configuration.md#websocket) configuration file. You'll need to set *at least* the following parameters:
+
+* [`configurationStoreServers`](reference-configuration.md#websocket-configurationStoreServers)
+* [`webServicePort`](reference-configuration.md#websocket-webServicePort)
+* [`clusterName`](reference-configuration.md#websocket-clusterName)
+
+Here's an example:
+
+```properties
+
+configurationStoreServers=zk1:2181,zk2:2181,zk3:2181
+webServicePort=8080
+clusterName=my-cluster
+
+```
+
+### Security settings
+
+To enable TLS encryption on WebSocket service:
+
+```properties
+
+tlsEnabled=true
+tlsAllowInsecureConnection=false
+tlsCertificateFilePath=/path/to/client-websocket.cert.pem
+tlsKeyFilePath=/path/to/client-websocket.key-pk8.pem
+tlsTrustCertsFilePath=/path/to/ca.cert.pem
+
+```
+
+### Starting the broker
+
+When the configuration is set, you can start the service using the [`pulsar-daemon`](reference-cli-tools.md#pulsar-daemon) tool:
+
+```shell
+
+$ bin/pulsar-daemon start websocket
+
+```
+
+## API Reference
+
+Pulsar's WebSocket API offers three endpoints for [producing](#producer-endpoint) messages, [consuming](#consumer-endpoint) messages and [reading](#reader-endpoint) messages.
+
+All exchanges via the WebSocket API use JSON.
+
+### Authentication
+
+#### Browser javascript WebSocket client
+
+Use the query param `token` transport the authentication token.
+
+```http
+
+ws://broker-service-url:8080/path?token=token
+
+```
+
+### Producer endpoint
+
+The producer endpoint requires you to specify a tenant, namespace, and topic in the URL:
+
+```http
+
+ws://broker-service-url:8080/ws/v2/producer/persistent/:tenant/:namespace/:topic
+
+```
+
+##### Query param
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`sendTimeoutMillis` | long | no | Send timeout (default: 30 secs)
+`batchingEnabled` | boolean | no | Enable batching of messages (default: false)
+`batchingMaxMessages` | int | no | Maximum number of messages permitted in a batch (default: 1000)
+`maxPendingMessages` | int | no | Set the max size of the internal-queue holding the messages (default: 1000)
+`batchingMaxPublishDelay` | long | no | Time period within which the messages will be batched (default: 10ms)
+`messageRoutingMode` | string | no | Message [routing mode](https://pulsar.apache.org/api/client/index.html?org/apache/pulsar/client/api/ProducerConfiguration.MessageRoutingMode.html) for the partitioned producer: `SinglePartition`, `RoundRobinPartition`
+`compressionType` | string | no | Compression [type](https://pulsar.apache.org/api/client/index.html?org/apache/pulsar/client/api/CompressionType.html): `LZ4`, `ZLIB`
+`producerName` | string | no | Specify the name for the producer. Pulsar will enforce only one producer with same name can be publishing on a topic
+`initialSequenceId` | long | no | Set the baseline for the sequence ids for messages published by the producer.
+`hashingScheme` | string | no | [Hashing function](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ProducerConfiguration.HashingScheme.html) to use when publishing on a partitioned topic: `JavaStringHash`, `Murmur3_32Hash`
+`token` | string | no | Authentication token, this is used for the browser javascript client
+
+
+#### Publishing a message
+
+```json
+
+{
+ "payload": "SGVsbG8gV29ybGQ=",
+ "properties": {"key1": "value1", "key2": "value2"},
+ "context": "1"
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`payload` | string | yes | Base-64 encoded payload
+`properties` | key-value pairs | no | Application-defined properties
+`context` | string | no | Application-defined request identifier
+`key` | string | no | For partitioned topics, decides which partition to use
+`replicationClusters` | array | no | Restrict replication to this list of [clusters](reference-terminology.md#cluster), specified by name
+
+
+##### Example success response
+
+```json
+
+{
+ "result": "ok",
+ "messageId": "CAAQAw==",
+ "context": "1"
+ }
+
+```
+
+##### Example failure response
+
+```json
+
+ {
+ "result": "send-error:3",
+ "errorMsg": "Failed to de-serialize from JSON",
+ "context": "1"
+ }
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`result` | string | yes | `ok` if successful or an error message if unsuccessful
+`messageId` | string | yes | Message ID assigned to the published message
+`context` | string | no | Application-defined request identifier
+
+
+### Consumer endpoint
+
+The consumer endpoint requires you to specify a tenant, namespace, and topic, as well as a subscription, in the URL:
+
+```http
+
+ws://broker-service-url:8080/ws/v2/consumer/persistent/:tenant/:namespace/:topic/:subscription
+
+```
+
+##### Query param
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`ackTimeoutMillis` | long | no | Set the timeout for unacked messages (default: 0)
+`subscriptionType` | string | no | [Subscription type](https://pulsar.apache.org/api/client/index.html?org/apache/pulsar/client/api/SubscriptionType.html): `Exclusive`, `Failover`, `Shared`, `Key_Shared`
+`receiverQueueSize` | int | no | Size of the consumer receive queue (default: 1000)
+`consumerName` | string | no | Consumer name
+`priorityLevel` | int | no | Define a [priority](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerConfiguration.html#setPriorityLevel-int-) for the consumer
+`maxRedeliverCount` | int | no | Define a [maxRedeliverCount](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.html#deadLetterPolicy-org.apache.pulsar.client.api.DeadLetterPolicy-) for the consumer (default: 0). Activates [Dead Letter Topic](https://github.com/apache/pulsar/wiki/PIP-22%3A-Pulsar-Dead-Letter-Topic) feature.
+`deadLetterTopic` | string | no | Define a [deadLetterTopic](http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/ConsumerBuilder.html#deadLetterPolicy-org.apache.pulsar.client.api.DeadLetterPolicy-) for the consumer (default: {topic}-{subscription}-DLQ). Activates [Dead Letter Topic](https://github.com/apache/pulsar/wiki/PIP-22%3A-Pulsar-Dead-Letter-Topic) feature.
+`pullMode` | boolean | no | Enable pull mode (default: false). See "Flow Control" below.
+`negativeAckRedeliveryDelay` | int | no | When a message is negatively acknowledged, it will be redelivered to the DLQ.
+`token` | string | no | Authentication token, this is used for the browser javascript client
+
+NB: these parameter (except `pullMode`) apply to the internal consumer of the WebSocket service.
+So messages will be subject to the redelivery settings as soon as the get into the receive queue,
+even if the client doesn't consume on the WebSocket.
+
+##### Receiving messages
+
+Server will push messages on the WebSocket session:
+
+```json
+
+{
+ "messageId": "CAAQAw==",
+ "payload": "SGVsbG8gV29ybGQ=",
+ "properties": {"key1": "value1", "key2": "value2"},
+ "publishTime": "2016-08-30 16:45:57.785",
+ "redeliveryCount": 4
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`messageId` | string | yes | Message ID
+`payload` | string | yes | Base-64 encoded payload
+`publishTime` | string | yes | Publish timestamp
+`redeliveryCount` | number | yes | Number of times this message was already delivered
+`properties` | key-value pairs | no | Application-defined properties
+`key` | string | no | Original routing key set by producer
+
+#### Acknowledging the message
+
+Consumer needs to acknowledge the successful processing of the message to
+have the Pulsar broker delete it.
+
+```json
+
+{
+ "messageId": "CAAQAw=="
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`messageId`| string | yes | Message ID of the processed message
+
+#### Negatively acknowledging messages
+
+```json
+
+{
+ "type": "negativeAcknowledge",
+ "messageId": "CAAQAw=="
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`messageId`| string | yes | Message ID of the processed message
+
+#### Flow control
+
+##### Push Mode
+
+By default (`pullMode=false`), the consumer endpoint will use the `receiverQueueSize` parameter both to size its
+internal receive queue and to limit the number of unacknowledged messages that are passed to the WebSocket client.
+In this mode, if you don't send acknowledgements, the Pulsar WebSocket service will stop sending messages after reaching
+`receiverQueueSize` unacked messages sent to the WebSocket client.
+
+##### Pull Mode
+
+If you set `pullMode` to `true`, the WebSocket client will need to send `permit` commands to permit the
+Pulsar WebSocket service to send more messages.
+
+```json
+
+{
+ "type": "permit",
+ "permitMessages": 100
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`type`| string | yes | Type of command. Must be `permit`
+`permitMessages`| int | yes | Number of messages to permit
+
+NB: in this mode it's possible to acknowledge messages in a different connection.
+
+#### Check if reach end of topic
+
+Consumer can check if it has reached end of topic by sending `isEndOfTopic` request.
+
+**Request**
+
+```json
+
+{
+ "type": "isEndOfTopic"
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`type`| string | yes | Type of command. Must be `isEndOfTopic`
+
+**Response**
+
+```json
+
+{
+ "endOfTopic": "true/false"
+ }
+
+```
+
+### Reader endpoint
+
+The reader endpoint requires you to specify a tenant, namespace, and topic in the URL:
+
+```http
+
+ws://broker-service-url:8080/ws/v2/reader/persistent/:tenant/:namespace/:topic
+
+```
+
+##### Query param
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`readerName` | string | no | Reader name
+`receiverQueueSize` | int | no | Size of the consumer receive queue (default: 1000)
+`messageId` | int or enum | no | Message ID to start from, `earliest` or `latest` (default: `latest`)
+`token` | string | no | Authentication token, this is used for the browser javascript client
+
+##### Receiving messages
+
+Server will push messages on the WebSocket session:
+
+```json
+
+{
+ "messageId": "CAAQAw==",
+ "payload": "SGVsbG8gV29ybGQ=",
+ "properties": {"key1": "value1", "key2": "value2"},
+ "publishTime": "2016-08-30 16:45:57.785",
+ "redeliveryCount": 4
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`messageId` | string | yes | Message ID
+`payload` | string | yes | Base-64 encoded payload
+`publishTime` | string | yes | Publish timestamp
+`redeliveryCount` | number | yes | Number of times this message was already delivered
+`properties` | key-value pairs | no | Application-defined properties
+`key` | string | no | Original routing key set by producer
+
+#### Acknowledging the message
+
+**In WebSocket**, Reader needs to acknowledge the successful processing of the message to
+have the Pulsar WebSocket service update the number of pending messages.
+If you don't send acknowledgements, Pulsar WebSocket service will stop sending messages after reaching the pendingMessages limit.
+
+```json
+
+{
+ "messageId": "CAAQAw=="
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`messageId`| string | yes | Message ID of the processed message
+
+#### Check if reach end of topic
+
+Consumer can check if it has reached end of topic by sending `isEndOfTopic` request.
+
+**Request**
+
+```json
+
+{
+ "type": "isEndOfTopic"
+}
+
+```
+
+Key | Type | Required? | Explanation
+:---|:-----|:----------|:-----------
+`type`| string | yes | Type of command. Must be `isEndOfTopic`
+
+**Response**
+
+```json
+
+{
+ "endOfTopic": "true/false"
+ }
+
+```
+
+### Error codes
+
+In case of error the server will close the WebSocket session using the
+following error codes:
+
+Error Code | Error Message
+:----------|:-------------
+1 | Failed to create producer
+2 | Failed to subscribe
+3 | Failed to deserialize from JSON
+4 | Failed to serialize to JSON
+5 | Failed to authenticate client
+6 | Client is not authorized
+7 | Invalid payload encoding
+8 | Unknown error
+
+> The application is responsible for re-establishing a new WebSocket session after a backoff period.
+
+## Client examples
+
+Below you'll find code examples for the Pulsar WebSocket API in [Python](#python) and [Node.js](#nodejs).
+
+### Python
+
+This example uses the [`websocket-client`](https://pypi.python.org/pypi/websocket-client) package. You can install it using [pip](https://pypi.python.org/pypi/pip):
+
+```shell
+
+$ pip install websocket-client
+
+```
+
+You can also download it from [PyPI](https://pypi.python.org/pypi/websocket-client).
+
+#### Python producer
+
+Here's an example Python producer that sends a simple message to a Pulsar [topic](reference-terminology.md#topic):
+
+```python
+
+import websocket, base64, json
+
+# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
+enable_TLS = False
+scheme = 'ws'
+if enable_TLS:
+ scheme = 'wss'
+
+TOPIC = scheme + '://localhost:8080/ws/v2/producer/persistent/public/default/my-topic'
+
+ws = websocket.create_connection(TOPIC)
+
+# Send one message as JSON
+ws.send(json.dumps({
+ 'payload' : base64.b64encode('Hello World'),
+ 'properties': {
+ 'key1' : 'value1',
+ 'key2' : 'value2'
+ },
+ 'context' : 5
+}))
+
+response = json.loads(ws.recv())
+if response['result'] == 'ok':
+ print 'Message published successfully'
+else:
+ print 'Failed to publish message:', response
+ws.close()
+
+```
+
+#### Python consumer
+
+Here's an example Python consumer that listens on a Pulsar topic and prints the message ID whenever a message arrives:
+
+```python
+
+import websocket, base64, json
+
+# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
+enable_TLS = False
+scheme = 'ws'
+if enable_TLS:
+ scheme = 'wss'
+
+TOPIC = scheme + '://localhost:8080/ws/v2/consumer/persistent/public/default/my-topic/my-sub'
+
+ws = websocket.create_connection(TOPIC)
+
+while True:
+ msg = json.loads(ws.recv())
+ if not msg: break
+
+ print "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload']))
+
+ # Acknowledge successful processing
+ ws.send(json.dumps({'messageId' : msg['messageId']}))
+
+ws.close()
+
+```
+
+#### Python reader
+
+Here's an example Python reader that listens on a Pulsar topic and prints the message ID whenever a message arrives:
+
+```python
+
+import websocket, base64, json
+
+# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
+enable_TLS = False
+scheme = 'ws'
+if enable_TLS:
+ scheme = 'wss'
+
+TOPIC = scheme + '://localhost:8080/ws/v2/reader/persistent/public/default/my-topic'
+ws = websocket.create_connection(TOPIC)
+
+while True:
+ msg = json.loads(ws.recv())
+ if not msg: break
+
+ print "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload']))
+
+ # Acknowledge successful processing
+ ws.send(json.dumps({'messageId' : msg['messageId']}))
+
+ws.close()
+
+```
+
+### Node.js
+
+This example uses the [`ws`](https://websockets.github.io/ws/) package. You can install it using [npm](https://www.npmjs.com/):
+
+```shell
+
+$ npm install ws
+
+```
+
+#### Node.js producer
+
+Here's an example Node.js producer that sends a simple message to a Pulsar topic:
+
+```javascript
+
+const WebSocket = require('ws');
+
+// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
+const enableTLS = false;
+const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/producer/persistent/public/default/my-topic`;
+const ws = new WebSocket(topic);
+
+var message = {
+ "payload" : new Buffer("Hello World").toString('base64'),
+ "properties": {
+ "key1" : "value1",
+ "key2" : "value2"
+ },
+ "context" : "1"
+};
+
+ws.on('open', function() {
+ // Send one message
+ ws.send(JSON.stringify(message));
+});
+
+ws.on('message', function(message) {
+ console.log('received ack: %s', message);
+});
+
+```
+
+#### Node.js consumer
+
+Here's an example Node.js consumer that listens on the same topic used by the producer above:
+
+```javascript
+
+const WebSocket = require('ws');
+
+// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
+const enableTLS = false;
+const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/consumer/persistent/public/default/my-topic/my-sub`;
+const ws = new WebSocket(topic);
+
+ws.on('message', function(message) {
+ var receiveMsg = JSON.parse(message);
+ console.log('Received: %s - payload: %s', message, new Buffer(receiveMsg.payload, 'base64').toString());
+ var ackMsg = {"messageId" : receiveMsg.messageId};
+ ws.send(JSON.stringify(ackMsg));
+});
+
+```
+
+#### NodeJS reader
+
+```javascript
+
+const WebSocket = require('ws');
+
+// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
+const enableTLS = false;
+const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/reader/persistent/public/default/my-topic`;
+const ws = new WebSocket(topic);
+
+ws.on('message', function(message) {
+ var receiveMsg = JSON.parse(message);
+ console.log('Received: %s - payload: %s', message, new Buffer(receiveMsg.payload, 'base64').toString());
+ var ackMsg = {"messageId" : receiveMsg.messageId};
+ ws.send(JSON.stringify(ackMsg));
+});
+
+```
+
diff --git a/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json b/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json
index a353b96..3d4b328 100644
--- a/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json
+++ b/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json
@@ -297,6 +297,100 @@
"id": "version-2.4.1/security-extending"
}
]
+ },
+ {
+ "type": "category",
+ "label": "Client Libraries",
+ "items": [
+ {
+ "type": "doc",
+ "id": "version-2.4.1/client-libraries-java"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/client-libraries-go"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/client-libraries-python"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/client-libraries-cpp"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/client-libraries-websocket"
+ }
+ ]
+ },
+ {
+ "type": "category",
+ "label": "Admin API",
+ "items": [
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-overview"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-clusters"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-tenants"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-brokers"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-namespaces"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-permissions"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-persistent-topics"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-non-persistent-topics"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-partitioned-topics"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-schemas"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/admin-api-functions"
+ }
+ ]
+ },
+ {
+ "type": "category",
+ "label": "Adaptors",
+ "items": [
+ {
+ "type": "doc",
+ "id": "version-2.4.1/adaptors-kafka"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/adaptors-spark"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/adaptors-storm"
+ }
+ ]
}
]
}
\ No newline at end of file