[GitHub] [flink] AHeise commented on a change in pull request #17188: [FLINK-23864][docs] Add pulsar connector document

[GitBox <gi...@apache.org>]

AHeise commented on a change in pull request #17188:
URL: https://github.com/apache/flink/pull/17188#discussion_r704385012



##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.

Review comment:
       ```suggestion
   You can use the connector with Pulsar 2.7.0 or later, but since connector uses Pulsar [transactions](https://pulsar.apache.org/docs/en/txn-what/), the minimum recommend version is 2.8.0, where Pulsar transactions became stable.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.

Review comment:
       ```suggestion
   The connector is compatible from Pulsar 2.8.0 or later.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.

Review comment:
       ```suggestion
   The subscription name is required for consuming message. If you don't provide subscription type,
   the connector would use ```SubscriptionType.Shared``` as the default subscription type.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.

Review comment:
       ```suggestion
   If only the raw payload of the message (message data in bytes) is needed, you can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.

Review comment:
       Isn't this already subsumed in the previous paragraph?

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.
+
+Pulsar connector would use a ```UniformRangeGenerator``` which would divide the range by the flink source parallelism
+if no ```RangeGenerator``` was provided in ```Key_Shared``` subscription.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by ```setStartCursor(StartCursor)```.
+Built-in start cursor include:
+
+```java
+// Start from the earliest available message in the topic.
+StartCursor.earliest()
+
+// Start from the latest available message in the topic
+StartCursor.latest()
+
+// Start from a specified message between the earliest and the latest.
+// Pulsar connector would consume from the latest available message if the message id doesn't exist.
+// The start message is included in consuming result. 
+StartCursor.fromMessageId(MessageId)
+
+// Include or exclude the start message by using the second boolean parameter.
+StartCursor.fromMessageId(MessageId, boolean)
+
+// Start from the specified message time (Message<byte[]>#getEventTime()).
+StartCursor.fromMessageTime(long)
+```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (```MessageId```) of the message is ordered in that sequence.
+MessageId contains some extra information on how the message was stored (the ledger, entry, partition),
+you can create a MessageId by using ```DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)```
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source is designed to support both streaming and batch running mode.
+By default, the PulsarSource is set to run in streaming manner, thus never stops until Flink job fails or is cancelled.
+You can use ```setBoundedStopCursor(StopCursor)``` to specify stopping position and set the source running in batch mode.
+When all partitions have reached their stopping position, the source will exit.
+
+You can also set PulsarSource running in streaming mode, but still stop at the stopping position by using ```setUnboundedStopCursor(StopCursor)```.
+The source will exit when all partitions reach their specified stopping position.
+
+Built-in stop cursor include:
+
+```java
+// Connector will never stop consuming
+StopCursor.never()
+
+// Stop at the latest available message in Pulsar when connector start consuming.
+StopCursor.latest()
+
+// Stop when connector meet a given message, or stop at a message which is produced after this given message.
+StopCursor.atMessageId(MessageId)
+
+// Stop but include the given message in consuming result.
+StopCursor.afterMessageId(MessageId)
+
+// Stop at the specified message time (Message<byte[]>#getEventTime()).
+StopCursor.atEventTime(long)
+```
+
+### Configurable Options
+
+In addition to config options described above, you can set arbitrary options for PulsarSource, PulsarClient,
+PulsarAdmin and Pulsar Consumer by using ```setConfig(ConfigOption<T>, T)``` and ```setConfig(Configuration)```.
+
+#### Pulsar Client Options
+
+Pulsar connector use both [client API](https://pulsar.apache.org/docs/en/client-libraries-java/)
+for creating the consuming instance. Pulsar connector extract most parts of Pulsar's ```ClientConfigurationData```,
+which is required for creating a Pulsar client, as Flink config options in ```PulsarOptions```.
+
+{{< generated/pulsar_client_configuration >}} 
+
+#### Pulsar Admin Options
+
+[Admin API](https://pulsar.apache.org/docs/en/admin-api-overview/) is used for querying topic metadata
+and filter the desired topics when connector use topic pattern subscription. It would share most part of the
+config options with client API. The config options listed here is only used in admin API.
+They are also defined in ```PulsarOptions```.
+
+{{< generated/pulsar_admin_configuration >}}
+
+#### Pulsar Consumer Options
+
+Pulsar provide Reader API and Consumer API for consuming messages in different scenarios.
+Pulsar connector use Consumer API. Pulsar connector extract most parts of Pulsar's ```ConsumerConfigurationData``` as Flink config options in ```PulsarSourceOptions```.

Review comment:
       ```suggestion
   In general, Pulsar provides Reader API and Consumer API for consuming messages in different scenarios.
   Flink's Pulsar connector uses Consumer API. Pulsar connector extract most parts of Pulsar's ```ConsumerConfigurationData``` as Flink config options in ```PulsarSourceOptions```.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.

Review comment:
       ```suggestion
   Flink provides an [Apache Pulsar](https://pulsar.apache.org) source for reading data from Pulsar topics with exactly-once guarantees.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.

Review comment:
       I'd remove this as it didn't provide much information to me.

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.

Review comment:
       ```suggestion
   Pulsar source provides a builder class for constructing an instance of PulsarSource. The code snippet below shows
   how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
   with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as strings.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.
+
+Pulsar connector would use a ```UniformRangeGenerator``` which would divide the range by the flink source parallelism
+if no ```RangeGenerator``` was provided in ```Key_Shared``` subscription.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by ```setStartCursor(StartCursor)```.
+Built-in start cursor include:
+
+```java
+// Start from the earliest available message in the topic.
+StartCursor.earliest()
+
+// Start from the latest available message in the topic
+StartCursor.latest()
+
+// Start from a specified message between the earliest and the latest.
+// Pulsar connector would consume from the latest available message if the message id doesn't exist.
+// The start message is included in consuming result. 
+StartCursor.fromMessageId(MessageId)
+
+// Include or exclude the start message by using the second boolean parameter.
+StartCursor.fromMessageId(MessageId, boolean)
+
+// Start from the specified message time (Message<byte[]>#getEventTime()).
+StartCursor.fromMessageTime(long)
+```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (```MessageId```) of the message is ordered in that sequence.
+MessageId contains some extra information on how the message was stored (the ledger, entry, partition),
+you can create a MessageId by using ```DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)```
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source is designed to support both streaming and batch running mode.
+By default, the PulsarSource is set to run in streaming manner, thus never stops until Flink job fails or is cancelled.
+You can use ```setBoundedStopCursor(StopCursor)``` to specify stopping position and set the source running in batch mode.
+When all partitions have reached their stopping position, the source will exit.

Review comment:
       ```suggestion
   By default, the ```PulsarSource``` is set to run in a streaming manner, thus it never stops until Flink job fails or is cancelled.
   You can use ```setBoundedStopCursor(StopCursor)``` to specify stop position such that the source can run in batch mode.
   When all partitions have reached their stopping position, the source will finish.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.

Review comment:
       ```suggestion
   If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure that you also provide a ```RangeGenerator``` implementation.
   ```RangeGenerator``` would generate a set of key hash ranges such that
   a respective reader subtask will only dispatch messages where the hash of the message key is contained in the specified range.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}

Review comment:
       This should be transitively included automatically. I see that this is also written in Kafka, so I guess we should leave it in for now and then do a general cleanup.

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.

Review comment:
       ```suggestion
   Pulsar divides internally represents partitioned topic as a set of non-partitioned topics according to the partition size.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.

Review comment:
       ```suggestion
   Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
   For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
   The topic type would be ```persistent``` if you don't provide the topic type in the regular expression.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.
+
+Pulsar connector would use a ```UniformRangeGenerator``` which would divide the range by the flink source parallelism
+if no ```RangeGenerator``` was provided in ```Key_Shared``` subscription.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by ```setStartCursor(StartCursor)```.
+Built-in start cursor include:
+
+```java
+// Start from the earliest available message in the topic.
+StartCursor.earliest()
+
+// Start from the latest available message in the topic
+StartCursor.latest()
+
+// Start from a specified message between the earliest and the latest.
+// Pulsar connector would consume from the latest available message if the message id doesn't exist.
+// The start message is included in consuming result. 
+StartCursor.fromMessageId(MessageId)
+
+// Include or exclude the start message by using the second boolean parameter.
+StartCursor.fromMessageId(MessageId, boolean)
+
+// Start from the specified message time (Message<byte[]>#getEventTime()).
+StartCursor.fromMessageTime(long)
+```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (```MessageId```) of the message is ordered in that sequence.
+MessageId contains some extra information on how the message was stored (the ledger, entry, partition),
+you can create a MessageId by using ```DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)```
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source is designed to support both streaming and batch running mode.
+By default, the PulsarSource is set to run in streaming manner, thus never stops until Flink job fails or is cancelled.
+You can use ```setBoundedStopCursor(StopCursor)``` to specify stopping position and set the source running in batch mode.
+When all partitions have reached their stopping position, the source will exit.
+
+You can also set PulsarSource running in streaming mode, but still stop at the stopping position by using ```setUnboundedStopCursor(StopCursor)```.
+The source will exit when all partitions reach their specified stopping position.
+
+Built-in stop cursor include:
+
+```java
+// Connector will never stop consuming
+StopCursor.never()
+
+// Stop at the latest available message in Pulsar when connector start consuming.
+StopCursor.latest()
+
+// Stop when connector meet a given message, or stop at a message which is produced after this given message.
+StopCursor.atMessageId(MessageId)
+
+// Stop but include the given message in consuming result.
+StopCursor.afterMessageId(MessageId)
+
+// Stop at the specified message time (Message<byte[]>#getEventTime()).
+StopCursor.atEventTime(long)
+```
+
+### Configurable Options
+
+In addition to config options described above, you can set arbitrary options for PulsarSource, PulsarClient,
+PulsarAdmin and Pulsar Consumer by using ```setConfig(ConfigOption<T>, T)``` and ```setConfig(Configuration)```.
+
+#### Pulsar Client Options
+
+Pulsar connector use both [client API](https://pulsar.apache.org/docs/en/client-libraries-java/)
+for creating the consuming instance. Pulsar connector extract most parts of Pulsar's ```ClientConfigurationData```,
+which is required for creating a Pulsar client, as Flink config options in ```PulsarOptions```.
+
+{{< generated/pulsar_client_configuration >}} 
+
+#### Pulsar Admin Options
+
+[Admin API](https://pulsar.apache.org/docs/en/admin-api-overview/) is used for querying topic metadata
+and filter the desired topics when connector use topic pattern subscription. It would share most part of the

Review comment:
       ```suggestion
   [Admin API](https://pulsar.apache.org/docs/en/admin-api-overview/) is used for querying topic metadata
   and for discovering the desired topics when connector use topic pattern subscription. It would share most part of the
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.
+
+Pulsar connector would use a ```UniformRangeGenerator``` which would divide the range by the flink source parallelism
+if no ```RangeGenerator``` was provided in ```Key_Shared``` subscription.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by ```setStartCursor(StartCursor)```.
+Built-in start cursor include:
+
+```java
+// Start from the earliest available message in the topic.
+StartCursor.earliest()
+
+// Start from the latest available message in the topic
+StartCursor.latest()
+
+// Start from a specified message between the earliest and the latest.
+// Pulsar connector would consume from the latest available message if the message id doesn't exist.
+// The start message is included in consuming result. 
+StartCursor.fromMessageId(MessageId)
+
+// Include or exclude the start message by using the second boolean parameter.
+StartCursor.fromMessageId(MessageId, boolean)
+
+// Start from the specified message time (Message<byte[]>#getEventTime()).
+StartCursor.fromMessageTime(long)
+```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (```MessageId```) of the message is ordered in that sequence.
+MessageId contains some extra information on how the message was stored (the ledger, entry, partition),
+you can create a MessageId by using ```DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)```
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source is designed to support both streaming and batch running mode.
+By default, the PulsarSource is set to run in streaming manner, thus never stops until Flink job fails or is cancelled.
+You can use ```setBoundedStopCursor(StopCursor)``` to specify stopping position and set the source running in batch mode.
+When all partitions have reached their stopping position, the source will exit.
+
+You can also set PulsarSource running in streaming mode, but still stop at the stopping position by using ```setUnboundedStopCursor(StopCursor)```.
+The source will exit when all partitions reach their specified stopping position.

Review comment:
       ```suggestion
   The source will finish when all partitions reach their specified stopping position.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.

Review comment:
       Not too sure on this one
   ```suggestion
   When a Flink reader crashes, all (non-acknowledged and subsequent) messages are redelivered to the available Flink readers.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.
+
+Pulsar connector would use a ```UniformRangeGenerator``` which would divide the range by the flink source parallelism
+if no ```RangeGenerator``` was provided in ```Key_Shared``` subscription.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by ```setStartCursor(StartCursor)```.
+Built-in start cursor include:
+
+```java
+// Start from the earliest available message in the topic.
+StartCursor.earliest()
+
+// Start from the latest available message in the topic
+StartCursor.latest()
+
+// Start from a specified message between the earliest and the latest.
+// Pulsar connector would consume from the latest available message if the message id doesn't exist.
+// The start message is included in consuming result. 
+StartCursor.fromMessageId(MessageId)
+
+// Include or exclude the start message by using the second boolean parameter.
+StartCursor.fromMessageId(MessageId, boolean)
+
+// Start from the specified message time (Message<byte[]>#getEventTime()).
+StartCursor.fromMessageTime(long)
+```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (```MessageId```) of the message is ordered in that sequence.
+MessageId contains some extra information on how the message was stored (the ledger, entry, partition),
+you can create a MessageId by using ```DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)```
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source is designed to support both streaming and batch running mode.
+By default, the PulsarSource is set to run in streaming manner, thus never stops until Flink job fails or is cancelled.
+You can use ```setBoundedStopCursor(StopCursor)``` to specify stopping position and set the source running in batch mode.
+When all partitions have reached their stopping position, the source will exit.
+
+You can also set PulsarSource running in streaming mode, but still stop at the stopping position by using ```setUnboundedStopCursor(StopCursor)```.
+The source will exit when all partitions reach their specified stopping position.
+
+Built-in stop cursor include:
+
+```java
+// Connector will never stop consuming
+StopCursor.never()
+
+// Stop at the latest available message in Pulsar when connector start consuming.
+StopCursor.latest()
+
+// Stop when connector meet a given message, or stop at a message which is produced after this given message.
+StopCursor.atMessageId(MessageId)
+
+// Stop but include the given message in consuming result.
+StopCursor.afterMessageId(MessageId)
+
+// Stop at the specified message time (Message<byte[]>#getEventTime()).
+StopCursor.atEventTime(long)
+```
+
+### Configurable Options
+
+In addition to config options described above, you can set arbitrary options for PulsarSource, PulsarClient,
+PulsarAdmin and Pulsar Consumer by using ```setConfig(ConfigOption<T>, T)``` and ```setConfig(Configuration)```.
+
+#### Pulsar Client Options
+
+Pulsar connector use both [client API](https://pulsar.apache.org/docs/en/client-libraries-java/)
+for creating the consuming instance. Pulsar connector extract most parts of Pulsar's ```ClientConfigurationData```,
+which is required for creating a Pulsar client, as Flink config options in ```PulsarOptions```.
+
+{{< generated/pulsar_client_configuration >}} 
+
+#### Pulsar Admin Options
+
+[Admin API](https://pulsar.apache.org/docs/en/admin-api-overview/) is used for querying topic metadata
+and filter the desired topics when connector use topic pattern subscription. It would share most part of the
+config options with client API. The config options listed here is only used in admin API.
+They are also defined in ```PulsarOptions```.
+
+{{< generated/pulsar_admin_configuration >}}
+
+#### Pulsar Consumer Options
+
+Pulsar provide Reader API and Consumer API for consuming messages in different scenarios.
+Pulsar connector use Consumer API. Pulsar connector extract most parts of Pulsar's ```ConsumerConfigurationData``` as Flink config options in ```PulsarSourceOptions```.
+
+{{< generated/pulsar_consumer_configuration >}}
+
+#### Pulsar Source Connector Options
+
+The config options below is mainly used for customizing the performance and message acknowledge behavior.
+You can just leave them alone if you don't meet any performance issues.
+
+{{< generated/pulsar_source_configuration >}}
+
+### Dynamic Partition Discovery
+
+In order to handle scenarios like topic scaling-out or topic creation without restarting the Flink
+job, Pulsar source can be configured to periodically discover new partitions under provided
+topic-partition subscribing pattern. To enable partition discovery, set a non-negative value for
+option ```PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS```:
+
+```java
+// discover new partitions per 10 seconds
+PulsarSource.builder()
+        .setConfig(PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS, 10000);
+```
+
+{{< hint warning >}}
+Partition discovery is **enabled** by default, Pulsar connector would query the topic metadata every 30 seconds.
+You need to explicitly set the partition discovery interval to a negative number to disable this feature.
+
+The partition discovery would be disabled in batch mode even if you set this option with a non-negative value.
+{{< /hint >}}
+
+### Event Time and Watermarks
+
+By default, the message will use the timestamp embedded in Pulsar ```Message<byte[]>``` as the event time.
+You can define your own ```WatermarkStrategy``` for extract event time from the message itself,
+and emit watermark downstream:
+
+```java
+env.fromSource(pulsarSource, new CustomWatermarkStrategy(), "Pulsar Source With Custom Watermark Strategy")
+```
+
+[This documentation]({{< ref "docs/dev/datastream/event-time/generating_watermarks.md" >}}) describes
+details about how to define a ```WatermarkStrategy```.
+
+### Message Acknowledgement
+
+When a subscription is created, Pulsar [retains](https://pulsar.apache.org/docs/en/concepts-architecture-overview/#persistent-storage) all messages, even if the consumer is disconnected.
+The retained messages are discarded only when the connector acknowledges that all these messages are processed successfully.
+Pulsar connector supports four types of subscription, this makes the acknowledge behaviors variety among different subscriptions.
+
+#### Acknowledgement on Exclusive and Failover
+
+```Exclusive``` and ```Failover``` supports cumulative acknowledgment, we only need to acknowledge
+the latest successfully consumed message. All the message before the given message would be marked
+with consumed status.

Review comment:
       ```suggestion
   ```Exclusive``` and ```Failover``` supports cumulative acknowledgment: Flink only needs to acknowledge
   the latest successfully consumed message. All the message before the given message would be marked
   with consumed status.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.
+
+Pulsar connector would use a ```UniformRangeGenerator``` which would divide the range by the flink source parallelism
+if no ```RangeGenerator``` was provided in ```Key_Shared``` subscription.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by ```setStartCursor(StartCursor)```.
+Built-in start cursor include:
+
+```java
+// Start from the earliest available message in the topic.
+StartCursor.earliest()
+
+// Start from the latest available message in the topic
+StartCursor.latest()
+
+// Start from a specified message between the earliest and the latest.
+// Pulsar connector would consume from the latest available message if the message id doesn't exist.
+// The start message is included in consuming result. 
+StartCursor.fromMessageId(MessageId)
+
+// Include or exclude the start message by using the second boolean parameter.
+StartCursor.fromMessageId(MessageId, boolean)
+
+// Start from the specified message time (Message<byte[]>#getEventTime()).
+StartCursor.fromMessageTime(long)
+```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (```MessageId```) of the message is ordered in that sequence.
+MessageId contains some extra information on how the message was stored (the ledger, entry, partition),
+you can create a MessageId by using ```DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)```
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source is designed to support both streaming and batch running mode.
+By default, the PulsarSource is set to run in streaming manner, thus never stops until Flink job fails or is cancelled.
+You can use ```setBoundedStopCursor(StopCursor)``` to specify stopping position and set the source running in batch mode.
+When all partitions have reached their stopping position, the source will exit.
+
+You can also set PulsarSource running in streaming mode, but still stop at the stopping position by using ```setUnboundedStopCursor(StopCursor)```.
+The source will exit when all partitions reach their specified stopping position.
+
+Built-in stop cursor include:
+
+```java
+// Connector will never stop consuming
+StopCursor.never()
+
+// Stop at the latest available message in Pulsar when connector start consuming.
+StopCursor.latest()
+
+// Stop when connector meet a given message, or stop at a message which is produced after this given message.
+StopCursor.atMessageId(MessageId)
+
+// Stop but include the given message in consuming result.
+StopCursor.afterMessageId(MessageId)
+
+// Stop at the specified message time (Message<byte[]>#getEventTime()).
+StopCursor.atEventTime(long)
+```
+
+### Configurable Options
+
+In addition to config options described above, you can set arbitrary options for PulsarSource, PulsarClient,
+PulsarAdmin and Pulsar Consumer by using ```setConfig(ConfigOption<T>, T)``` and ```setConfig(Configuration)```.
+
+#### Pulsar Client Options
+
+Pulsar connector use both [client API](https://pulsar.apache.org/docs/en/client-libraries-java/)
+for creating the consuming instance. Pulsar connector extract most parts of Pulsar's ```ClientConfigurationData```,
+which is required for creating a Pulsar client, as Flink config options in ```PulsarOptions```.
+
+{{< generated/pulsar_client_configuration >}} 
+
+#### Pulsar Admin Options
+
+[Admin API](https://pulsar.apache.org/docs/en/admin-api-overview/) is used for querying topic metadata
+and filter the desired topics when connector use topic pattern subscription. It would share most part of the
+config options with client API. The config options listed here is only used in admin API.
+They are also defined in ```PulsarOptions```.
+
+{{< generated/pulsar_admin_configuration >}}
+
+#### Pulsar Consumer Options
+
+Pulsar provide Reader API and Consumer API for consuming messages in different scenarios.
+Pulsar connector use Consumer API. Pulsar connector extract most parts of Pulsar's ```ConsumerConfigurationData``` as Flink config options in ```PulsarSourceOptions```.
+
+{{< generated/pulsar_consumer_configuration >}}
+
+#### Pulsar Source Connector Options
+
+The config options below is mainly used for customizing the performance and message acknowledge behavior.
+You can just leave them alone if you don't meet any performance issues.
+
+{{< generated/pulsar_source_configuration >}}
+
+### Dynamic Partition Discovery
+
+In order to handle scenarios like topic scaling-out or topic creation without restarting the Flink
+job, Pulsar source can be configured to periodically discover new partitions under provided
+topic-partition subscribing pattern. To enable partition discovery, set a non-negative value for
+option ```PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS```:
+
+```java
+// discover new partitions per 10 seconds
+PulsarSource.builder()
+        .setConfig(PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS, 10000);
+```
+
+{{< hint warning >}}
+Partition discovery is **enabled** by default, Pulsar connector would query the topic metadata every 30 seconds.
+You need to explicitly set the partition discovery interval to a negative number to disable this feature.
+
+The partition discovery would be disabled in batch mode even if you set this option with a non-negative value.
+{{< /hint >}}
+
+### Event Time and Watermarks
+
+By default, the message will use the timestamp embedded in Pulsar ```Message<byte[]>``` as the event time.
+You can define your own ```WatermarkStrategy``` for extract event time from the message itself,
+and emit watermark downstream:
+
+```java
+env.fromSource(pulsarSource, new CustomWatermarkStrategy(), "Pulsar Source With Custom Watermark Strategy")
+```
+
+[This documentation]({{< ref "docs/dev/datastream/event-time/generating_watermarks.md" >}}) describes
+details about how to define a ```WatermarkStrategy```.
+
+### Message Acknowledgement
+
+When a subscription is created, Pulsar [retains](https://pulsar.apache.org/docs/en/concepts-architecture-overview/#persistent-storage) all messages, even if the consumer is disconnected.
+The retained messages are discarded only when the connector acknowledges that all these messages are processed successfully.
+Pulsar connector supports four types of subscription, this makes the acknowledge behaviors variety among different subscriptions.
+
+#### Acknowledgement on Exclusive and Failover
+
+```Exclusive``` and ```Failover``` supports cumulative acknowledgment, we only need to acknowledge
+the latest successfully consumed message. All the message before the given message would be marked
+with consumed status.
+
+Pulsar source acknowledge the current consuming message when when checkpoints are **completed**,
+for ensuring the consistency between Flink's checkpoint state and committed position on Pulsar brokers.

Review comment:
       ```suggestion
   Pulsar source acknowledge the current consuming message when checkpoints are **completed**,
   for ensuring the consistency between Flink's checkpoint state and committed position on Pulsar brokers.
   ```

##########
File path: docs/content/docs/connectors/datastream/pulsar.md
##########
@@ -0,0 +1,413 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+Apache Flink ships with a Pulsar connector which attempts to track the latest version of the Pulsar client.
+The version of the client it uses may change between Flink releases.
+Pulsar clients are backwards compatible with Pulsar 2.7.0 or later,
+but since connector need to use Pulsar [transaction](https://pulsar.apache.org/docs/en/txn-what/), it is stable after Pulsar 2.8.0 release.
+The connector are compatible from Pulsar 2.8.0 or later.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+if you are using Pulsar source, ```flink-connector-base``` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy ```SourceFunction``` or on Flink below 1.14.0, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink)
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription ```my-subscription``` and deserialize the raw payload of the messages as string.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+    .setServiceUrl(serviceUrl)
+    .setAdminUrl(adminUrl)
+    .setStartCursor(StartCursor.earliest())
+    .setTopics("my-topic")
+    .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+    .setSubscriptionName("my-subscription")
+    .setSubscriptionType(SubscriptionType.Exclusive)
+    .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by ```setServiceUrl(String)```
+- Pulsar service http url (aka. admin url), configured by ```setAdminUrl(String)```
+- Pulsar subscription name, configured by ```setSubscriptionName(String)```
+- Topics / partitions to subscribe, see the following
+  <a href="#topic-partition-subscription">Topic-partition subscription</a> for more details.
+- Deserializer to parse Pulsar messages, see the following
+  <a href="#deserializer">Deserializer</a> for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide 2 ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+  ```java
+  PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+  // Partition 0 and 2 of topic "topic-a"
+  PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+  ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+  ```java
+  PulsarSource.builder().setTopicPattern("topic-*")
+  ```
+
+#### Flexible Topic Naming
+
+All topic names after Pulsar 2.0 internally have the form ```{persistent|non-persistent}://tenant/namespace/topic```,
+but you can now use shorthand names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace
+
+Topic aspect | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+The table below shows some example topic name translations that use implicit defaults:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics you'll need to continue to specify the entire topic name,
+as the default-based rules for persistent topic names don't apply.
+Thus you cannot use a shorthand name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead
+{{< /hint >}}
+
+#### Subscribe Pulsar Topic Partition
+
+Pulsar divide partitioned topic into non-partitioned topics.
+All the partitioned topic would internally a set of non-partitioned topics by the partition size.
+
+For example, if we create a ```simple-string``` topic with 3 partitions under the tenant ```sample``` with ```flink``` namespace.
+
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+```persistent://sample/flink/simple-string``` | Y
+```persistent://sample/flink/simple-string-partition-0``` | N
+```persistent://sample/flink/simple-string-partition-1``` | N
+```persistent://sample/flink/simple-string-partition-2``` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use ```PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")``` would consume the partition 1 and 2 of topic ```sample/flink/simple-string```.
+
+#### RegexSubscriptionMode for Topic Pattern 
+
+Pulsar connector extract the topic type (```persistent``` or ```non-persistent```) from the given topic pattern.
+For example, ```PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")``` would be ```non-persistent```.
+The topic type would be ```persistent``` if you don't provide the topic type in regular expression.
+
+In order to consume both ```persistent``` and ```non-persistent``` topics by topic pattern,
+you can just use ```setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)```.
+Pulsar connector would filter the available topics by the ```RegexSubscriptionMode```.
+
+### Deserializer
+
+A deserializer is required for parsing Pulsar messages. Deserializer (Deserialization schema) can be
+configured by ```setDeserializationSchema(PulsarDeserializationSchema)```.
+The ```PulsarDeserializationSchema``` defines how to deserialize a Pulsar ```Message<byte[]>```.
+
+If only the raw payload of the message (message data in bytes) is needed.
+You can use the predefined ```PulsarDeserializationSchema```. Pulsar connector provided three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+  ```java
+  // Primitive types
+  PulsarDeserializationSchema.pulsarSchema(Schema)
+
+  // Struct types (JSON, Protobuf, Avro, etc.)
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+  // KeyValue type
+  PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+  ```
+- Decode the message by using Flink's ```DeserializationSchema```
+  ```java
+  PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+  ```
+- Decode the message by using Flink's ```TypeInformation```
+  ```java
+  PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+  ```
+
+Pulsar ```Message<byte[]>``` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by ```Message<byte[]>``` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement ```PulsarDeserializationSchema```.
+And make sure the ```TypeInformation``` from the ```PulsarDeserializationSchema.getProducedType()``` must be correct.
+Flink would use this ```TypeInformation``` for passing the messages to downstream.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+Four subscription types are available in Pulsar connector:
+[exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive),
+[shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared),
+[failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover),
+and [key_shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared).
+
+There is no difference between ```exclusive``` and ```failover``` in Flink connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are delivered to the available Flink readers.
+
+The subscription name is required for consuming message. If you don't provide subscription type,
+connector would the ```SubscriptionType.Shared``` as the default subscription type.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use ```Key_Shared``` subscription on Pulsar connector. Make sure you have provide a ```RangeGenerator``` implementation.
+```RangeGenerator``` would generate a set of key hash range that
+Pulsar will only dispatch messages which hash of the message key contains by the specified range.
+
+Pulsar connector would use a ```UniformRangeGenerator``` which would divide the range by the flink source parallelism
+if no ```RangeGenerator``` was provided in ```Key_Shared``` subscription.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by ```setStartCursor(StartCursor)```.
+Built-in start cursor include:
+
+```java
+// Start from the earliest available message in the topic.
+StartCursor.earliest()
+
+// Start from the latest available message in the topic
+StartCursor.latest()
+
+// Start from a specified message between the earliest and the latest.
+// Pulsar connector would consume from the latest available message if the message id doesn't exist.
+// The start message is included in consuming result. 
+StartCursor.fromMessageId(MessageId)
+
+// Include or exclude the start message by using the second boolean parameter.
+StartCursor.fromMessageId(MessageId, boolean)
+
+// Start from the specified message time (Message<byte[]>#getEventTime()).
+StartCursor.fromMessageTime(long)
+```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (```MessageId```) of the message is ordered in that sequence.
+MessageId contains some extra information on how the message was stored (the ledger, entry, partition),
+you can create a MessageId by using ```DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)```
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source is designed to support both streaming and batch running mode.
+By default, the PulsarSource is set to run in streaming manner, thus never stops until Flink job fails or is cancelled.
+You can use ```setBoundedStopCursor(StopCursor)``` to specify stopping position and set the source running in batch mode.
+When all partitions have reached their stopping position, the source will exit.
+
+You can also set PulsarSource running in streaming mode, but still stop at the stopping position by using ```setUnboundedStopCursor(StopCursor)```.
+The source will exit when all partitions reach their specified stopping position.
+
+Built-in stop cursor include:
+
+```java
+// Connector will never stop consuming
+StopCursor.never()
+
+// Stop at the latest available message in Pulsar when connector start consuming.
+StopCursor.latest()
+
+// Stop when connector meet a given message, or stop at a message which is produced after this given message.
+StopCursor.atMessageId(MessageId)
+
+// Stop but include the given message in consuming result.
+StopCursor.afterMessageId(MessageId)
+
+// Stop at the specified message time (Message<byte[]>#getEventTime()).
+StopCursor.atEventTime(long)
+```
+
+### Configurable Options
+
+In addition to config options described above, you can set arbitrary options for PulsarSource, PulsarClient,
+PulsarAdmin and Pulsar Consumer by using ```setConfig(ConfigOption<T>, T)``` and ```setConfig(Configuration)```.
+
+#### Pulsar Client Options
+
+Pulsar connector use both [client API](https://pulsar.apache.org/docs/en/client-libraries-java/)
+for creating the consuming instance. Pulsar connector extract most parts of Pulsar's ```ClientConfigurationData```,
+which is required for creating a Pulsar client, as Flink config options in ```PulsarOptions```.
+
+{{< generated/pulsar_client_configuration >}} 
+
+#### Pulsar Admin Options
+
+[Admin API](https://pulsar.apache.org/docs/en/admin-api-overview/) is used for querying topic metadata
+and filter the desired topics when connector use topic pattern subscription. It would share most part of the
+config options with client API. The config options listed here is only used in admin API.
+They are also defined in ```PulsarOptions```.
+
+{{< generated/pulsar_admin_configuration >}}
+
+#### Pulsar Consumer Options
+
+Pulsar provide Reader API and Consumer API for consuming messages in different scenarios.
+Pulsar connector use Consumer API. Pulsar connector extract most parts of Pulsar's ```ConsumerConfigurationData``` as Flink config options in ```PulsarSourceOptions```.
+
+{{< generated/pulsar_consumer_configuration >}}
+
+#### Pulsar Source Connector Options
+
+The config options below is mainly used for customizing the performance and message acknowledge behavior.
+You can just leave them alone if you don't meet any performance issues.
+
+{{< generated/pulsar_source_configuration >}}
+
+### Dynamic Partition Discovery
+
+In order to handle scenarios like topic scaling-out or topic creation without restarting the Flink
+job, Pulsar source can be configured to periodically discover new partitions under provided
+topic-partition subscribing pattern. To enable partition discovery, set a non-negative value for
+option ```PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS```:
+
+```java
+// discover new partitions per 10 seconds
+PulsarSource.builder()
+        .setConfig(PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS, 10000);
+```
+
+{{< hint warning >}}
+Partition discovery is **enabled** by default, Pulsar connector would query the topic metadata every 30 seconds.
+You need to explicitly set the partition discovery interval to a negative number to disable this feature.
+
+The partition discovery would be disabled in batch mode even if you set this option with a non-negative value.
+{{< /hint >}}
+
+### Event Time and Watermarks
+
+By default, the message will use the timestamp embedded in Pulsar ```Message<byte[]>``` as the event time.
+You can define your own ```WatermarkStrategy``` for extract event time from the message itself,
+and emit watermark downstream:
+
+```java
+env.fromSource(pulsarSource, new CustomWatermarkStrategy(), "Pulsar Source With Custom Watermark Strategy")
+```
+
+[This documentation]({{< ref "docs/dev/datastream/event-time/generating_watermarks.md" >}}) describes
+details about how to define a ```WatermarkStrategy```.
+
+### Message Acknowledgement
+
+When a subscription is created, Pulsar [retains](https://pulsar.apache.org/docs/en/concepts-architecture-overview/#persistent-storage) all messages, even if the consumer is disconnected.
+The retained messages are discarded only when the connector acknowledges that all these messages are processed successfully.
+Pulsar connector supports four types of subscription, this makes the acknowledge behaviors variety among different subscriptions.
+
+#### Acknowledgement on Exclusive and Failover
+
+```Exclusive``` and ```Failover``` supports cumulative acknowledgment, we only need to acknowledge
+the latest successfully consumed message. All the message before the given message would be marked
+with consumed status.
+
+Pulsar source acknowledge the current consuming message when when checkpoints are **completed**,
+for ensuring the consistency between Flink's checkpoint state and committed position on Pulsar brokers.
+
+If checkpointing is not enabled, Pulsar source would periodic acknowledge message.
+It's configured by ```PulsarSourceOptions.PULSAR_AUTO_COMMIT_CURSOR_INTERVAL```.
+
+Note that Pulsar source does **NOT** rely on committed positions for fault tolerance.
+Acknowledge message is only for exposing the progress of consumer and monitoring in these two subscriptions.
+
+#### Acknowledgement on Shared and Key_Shared
+
+```Shared``` and ```Key_Shared``` need to acknowledge message one by one. We will acknowledge
+message in a transaction and commit it to Pulsar.
+
+You should enable transaction in Pulsar ```borker.conf``` file when using these two subscriptions in connector:
+
+```text
+transactionCoordinatorEnabled=true
+```
+
+Pulsar transaction would be created with 3 hours as the timeout. Make sure the timeout is bigger than the
+checkpoint interval. You can change the transaction timeout by ```PulsarSourceOptions.PULSAR_TRANSACTION_TIMEOUT_MILLIS```

Review comment:
       ```suggestion
   Pulsar transaction would be created with 3 hours as the timeout. Make sure that ```timeout >>
   checkpoint interval + maximum recovery time```. You can change the transaction timeout by ```PulsarSourceOptions.PULSAR_TRANSACTION_TIMEOUT_MILLIS```
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org