You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by li...@apache.org on 2021/12/07 04:53:27 UTC
[pulsar] branch master updated: [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / functions/io/sql (#13143)
This is an automated email from the ASF dual-hosted git repository.
liuyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 286e636 [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / functions/io/sql (#13143)
286e636 is described below
commit 286e636fca4123347b6463049bf2e5e430857b9b
Author: Li Li <ur...@gmail.com>
AuthorDate: Tue Dec 7 12:52:23 2021 +0800
[website][upgrade]feat: website upgrade / docs migration - 2.4.1 / functions/io/sql (#13143)
* [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / get started/concepts/schema
Signed-off-by: LiLi <ur...@gmail.com>
* [website][upgrade]feat: website upgrade / docs migration - 2.4.1 / functions/io/sql
Signed-off-by: LiLi <ur...@gmail.com>
Co-authored-by: Anonymitaet <50...@users.noreply.github.com>
---
.../versioned_docs/version-2.4.1/functions-cli.md | 196 +++
.../version-2.4.1/functions-debug.md | 514 ++++++
.../version-2.4.1/functions-deploy.md | 241 +++
.../version-2.4.1/functions-develop.md | 1144 +++++++++++++
.../version-2.4.1/functions-overview.md | 205 +++
.../version-2.4.1/functions-runtime.md | 157 ++
.../version-2.4.1/functions-worker.md | 280 ++++
.../versioned_docs/version-2.4.1/io-cdc.md | 26 +
.../versioned_docs/version-2.4.1/io-connectors.md | 31 +
.../versioned_docs/version-2.4.1/io-debug.md | 407 +++++
.../versioned_docs/version-2.4.1/io-develop.md | 205 +++
.../versioned_docs/version-2.4.1/io-managing.md | 193 +++
.../versioned_docs/version-2.4.1/io-overview.md | 158 ++
.../versioned_docs/version-2.4.1/io-quickstart.md | 892 ++++++++++
.../versioned_docs/version-2.4.1/io-use.md | 1745 ++++++++++++++++++++
.../version-2.4.1/sql-deployment-configurations.md | 189 +++
.../version-2.4.1/sql-getting-started.md | 187 +++
.../versioned_docs/version-2.4.1/sql-overview.md | 18 +
.../version-2.4.1/window-functions-context.md | 581 +++++++
.../versioned_sidebars/version-2.4.1-sidebars.json | 94 ++
20 files changed, 7463 insertions(+)
diff --git a/site2/website-next/versioned_docs/version-2.4.1/functions-cli.md b/site2/website-next/versioned_docs/version-2.4.1/functions-cli.md
new file mode 100644
index 0000000..e3ce803
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/functions-cli.md
@@ -0,0 +1,196 @@
+---
+id: functions-cli
+title: Pulsar Functions command line tool
+sidebar_label: "Reference: CLI"
+original_id: functions-cli
+---
+
+The following tables list Pulsar Functions command-line tools. You can learn Pulsar Functions modes, commands, and parameters.
+
+## localrun
+
+Run Pulsar Functions locally, rather than deploying it to the Pulsar cluster.
+
+Name | Description | Default
+---|---|---
+auto-ack | Whether or not the framework acknowledges messages automatically. | false |
+broker-service-url | The URL for the Pulsar broker. | |
+classname | The class name of a Pulsar Function.| |
+client-auth-params | Client authentication parameter. | |
+client-auth-plugin | Client authentication plugin using which function-process can connect to broker. | |
+CPU | The CPU in cores that need to be allocated per function instance (applicable only to docker runtime).| |
+custom-schema-inputs | The map of input topics to Schema class names (as a JSON string). | |
+custom-serde-inputs | The map of input topics to SerDe class names (as a JSON string). | |
+dead-letter-topic | The topic where all messages that were not processed successfully are sent. This parameter is not supported in Python Functions. | |
+disk | The disk in bytes that need to be allocated per function instance (applicable only to docker runtime). | |
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+function-config-file | The path to a YAML config file specifying the configuration of a Pulsar Function. | |
+go | Path to the main Go executable binary for the function (if the function is written in Go). | |
+hostname-verification-enabled | Enable hostname verification. | false
+inputs | The input topic or topics of a Pulsar Function (multiple topics can be specified as a comma-separated list). | |
+jar | Path to the jar file for the function (if the function is written in Java). It also supports URL-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package. | |
+instance-id-offset | Start the instanceIds from this offset. | 0
+log-topic | The topic to which the logs a Pulsar Function are produced. | |
+max-message-retries | How many times should we try to process a message before giving up. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+output | The output topic of a Pulsar Function (If none is specified, no output is written). | |
+output-serde-classname | The SerDe class to be used for messages output by the function. | |
+parallelism | The parallelism factor of a Pulsar Function (i.e. the number of function instances to run). | |
+processing-guarantees | The processing guarantees (delivery semantics) applied to the function. Available values: [ATLEAST_ONCE, ATMOST_ONCE, EFFECTIVELY_ONCE]. | ATLEAST_ONCE
+py | Path to the main Python file/Python Wheel file for the function (if the function is written in Python). | |
+ram | The ram in bytes that need to be allocated per function instance (applicable only to process/docker runtime). | |
+retain-ordering | Function consumes and processes messages in order. | |
+schema-type | The builtin schema type or custom schema class name to be used for messages output by the function. | |
+sliding-interval-count | The number of messages after which the window slides. | |
+sliding-interval-duration-ms | The time duration after which the window slides. | |
+subs-name | Pulsar source subscription name if user wants a specific subscription-name for the input-topic consumer. | |
+tenant | The tenant of a Pulsar Function. | |
+timeout-ms | The message timeout in milliseconds. | |
+tls-allow-insecure | Allow insecure tls connection. | false
+tls-trust-cert-path | tls trust cert file path. | |
+topics-pattern | The topic pattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe class name for a pattern in --custom-serde-inputs (only supported in Java Function). | |
+use-tls | Use tls connection. | false
+user-config | User-defined config key/values. | |
+window-length-count | The number of messages per window. | |
+window-length-duration-ms | The time duration of the window in milliseconds. | |
+
+
+## create
+
+Create and deploy a Pulsar Function in cluster mode.
+
+Name | Description | Default
+---|---|---
+auto-ack | Whether or not the framework acknowledges messages automatically. | false |
+classname | The class name of a Pulsar Function. | |
+CPU | The CPU in cores that need to be allocated per function instance (applicable only to docker runtime).| |
+custom-schema-inputs | The map of input topics to Schema class names (as a JSON string). | |
+custom-serde-inputs | The map of input topics to SerDe class names (as a JSON string). | |
+dead-letter-topic | The topic where all messages that were not processed successfully are sent. This parameter is not supported in Python Functions. | |
+disk | The disk in bytes that need to be allocated per function instance (applicable only to docker runtime). | |
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+function-config-file | The path to a YAML config file specifying the configuration of a Pulsar Function. | |
+go | Path to the main Go executable binary for the function (if the function is written in Go). | |
+inputs | The input topic or topics of a Pulsar Function (multiple topics can be specified as a comma-separated list). | |
+jar | Path to the jar file for the function (if the function is written in Java). It also supports URL-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package. | |
+log-topic | The topic to which the logs of a Pulsar Function are produced. | |
+max-message-retries | How many times should we try to process a message before giving up. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+output | The output topic of a Pulsar Function (If none is specified, no output is written). | |
+output-serde-classname | The SerDe class to be used for messages output by the function. | |
+parallelism | The parallelism factor of a Pulsar Function (i.e. the number of function instances to run). | |
+processing-guarantees | The processing guarantees (delivery semantics) applied to the function. Available values: [ATLEAST_ONCE, ATMOST_ONCE, EFFECTIVELY_ONCE]. | ATLEAST_ONCE
+py | Path to the main Python file/Python Wheel file for the function (if the function is written in Python). | |
+ram | The ram in bytes that need to be allocated per function instance (applicable only to process/docker runtime). | |
+retain-ordering | Function consumes and processes messages in order. | |
+schema-type | The builtin schema type or custom schema class name to be used for messages output by the function. | |
+sliding-interval-count | The number of messages after which the window slides. | |
+sliding-interval-duration-ms | The time duration after which the window slides. | |
+subs-name | Pulsar source subscription name if user wants a specific subscription-name for the input-topic consumer. | |
+tenant | The tenant of a Pulsar Function. | |
+timeout-ms | The message timeout in milliseconds. | |
+topics-pattern | The topic pattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe class name for a pattern in --custom-serde-inputs (only supported in Java Function). | |
+user-config | User-defined config key/values. | |
+window-length-count | The number of messages per window. | |
+window-length-duration-ms | The time duration of the window in milliseconds. | |
+
+## delete
+
+Delete a Pulsar Function that is running on a Pulsar cluster.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## update
+
+Update a Pulsar Function that has been deployed to a Pulsar cluster.
+
+Name | Description | Default
+---|---|---
+auto-ack | Whether or not the framework acknowledges messages automatically. | false
+classname | The class name of a Pulsar Function. | |
+CPU | The CPU in cores that need to be allocated per function instance (applicable only to docker runtime). | |
+custom-schema-inputs | The map of input topics to Schema class names (as a JSON string). | |
+custom-serde-inputs | The map of input topics to SerDe class names (as a JSON string). | |
+dead-letter-topic | The topic where all messages that were not processed successfully are sent. This parameter is not supported in Python Functions. | |
+disk | The disk in bytes that need to be allocated per function instance (applicable only to docker runtime). | |
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+function-config-file | The path to a YAML config file specifying the configuration of a Pulsar Function. | |
+go | Path to the main Go executable binary for the function (if the function is written in Go). | |
+inputs | The input topic or topics of a Pulsar Function (multiple topics can be specified as a comma-separated list). | |
+jar | Path to the jar file for the function (if the function is written in Java). It also supports URL-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package. | |
+log-topic | The topic to which the logs of a Pulsar Function are produced. | |
+max-message-retries | How many times should we try to process a message before giving up. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+output | The output topic of a Pulsar Function (If none is specified, no output is written). | |
+output-serde-classname | The SerDe class to be used for messages output by the function. | |
+parallelism | The parallelism factor of a Pulsar Function (i.e. the number of function instances to run). | |
+processing-guarantees | The processing guarantees (delivery semantics) applied to the function. Available values: [ATLEAST_ONCE, ATMOST_ONCE, EFFECTIVELY_ONCE]. | ATLEAST_ONCE
+py | Path to the main Python file/Python Wheel file for the function (if the function is written in Python). | |
+ram | The ram in bytes that need to be allocated per function instance (applicable only to process/docker runtime). | |
+retain-ordering | Function consumes and processes messages in order. | |
+schema-type | The builtin schema type or custom schema class name to be used for messages output by the function. | |
+sliding-interval-count | The number of messages after which the window slides. | |
+sliding-interval-duration-ms | The time duration after which the window slides. | |
+subs-name | Pulsar source subscription name if user wants a specific subscription-name for the input-topic consumer. | |
+tenant | The tenant of a Pulsar Function. | |
+timeout-ms | The message timeout in milliseconds. | |
+topics-pattern | The topic pattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe class name for a pattern in --custom-serde-inputs (only supported in Java Function). | |
+update-auth-data | Whether or not to update the auth data. | false
+user-config | User-defined config key/values. | |
+window-length-count | The number of messages per window. | |
+window-length-duration-ms | The time duration of the window in milliseconds. | |
+
+## get
+
+Fetch information about a Pulsar Function.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## restart
+
+Restart function instance.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+instance-id | The function instanceId (restart all instances if instance-id is not provided. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## stop
+
+Stops function instance.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+instance-id | The function instanceId (restart all instances if instance-id is not provided. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## start
+
+Starts a stopped function instance.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+instance-id | The function instanceId (restart all instances if instance-id is not provided. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
diff --git a/site2/website-next/versioned_docs/version-2.4.1/functions-debug.md b/site2/website-next/versioned_docs/version-2.4.1/functions-debug.md
new file mode 100644
index 0000000..8f619f5
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/functions-debug.md
@@ -0,0 +1,514 @@
+---
+id: functions-debug
+title: Debug Pulsar Functions
+sidebar_label: "How-to: Debug"
+original_id: functions-debug
+---
+
+You can use the following methods to debug Pulsar Functions:
+
+* [Use unit test](functions-debug.md#use-unit-test)
+* [Debug with localrun mode](functions-debug.md#debug-with-localrun-mode)
+* [Use log topic](functions-debug.md#use-log-topic)
+* [Use Functions CLI](functions-debug.md#use-functions-cli)
+
+## Use unit test
+
+A Pulsar Function is a function with inputs and outputs, you can test a Pulsar Function in a similar way as you test any function.
+
+For example, if you have the following Pulsar Function:
+
+```java
+
+import java.util.function.Function;
+
+public class JavaNativeExclamationFunction implements Function<String, String> {
+ @Override
+ public String apply(String input) {
+ return String.format("%s!", input);
+ }
+}
+
+```
+
+You can write a simple unit test to test Pulsar Function.
+
+```java
+
+@Test
+public void testJavaNativeExclamationFunction() {
+ JavaNativeExclamationFunction exclamation = new JavaNativeExclamationFunction();
+ String output = exclamation.apply("foo");
+ Assert.assertEquals(output, "foo!");
+}
+
+```
+
+The following Pulsar Function implements the `org.apache.pulsar.functions.api.Function` interface.
+
+```java
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+
+public class ExclamationFunction implements Function<String, String> {
+ @Override
+ public String process(String input, Context context) {
+ return String.format("%s!", input);
+ }
+}
+
+```
+
+In this situation, you can write a unit test for this function as well. Remember to mock the `Context` parameter. The following is an example.
+
+```java
+
+@Test
+public void testExclamationFunction() {
+ ExclamationFunction exclamation = new ExclamationFunction();
+ String output = exclamation.process("foo", mock(Context.class));
+ Assert.assertEquals(output, "foo!");
+}
+
+```
+
+## Debug with localrun mode
+When you run a Pulsar Function in localrun mode, it launches an instance of the Function on your local machine as a thread.
+
+In this mode, a Pulsar Function consumes and produces actual data to a Pulsar cluster, and mirrors how the function actually runs in a Pulsar cluster.
+
+:::note
+
+Currently, debugging with localrun mode is only supported by Pulsar Functions written in Java. You need Pulsar version 2.4.0 or later to do the following. Even though localrun is available in versions earlier than Pulsar 2.4.0, you cannot debug with localrun mode programmatically or run Functions as threads.
+
+:::
+
+You can launch your function in the following manner.
+
+```java
+
+FunctionConfig functionConfig = new FunctionConfig();
+functionConfig.setName(functionName);
+functionConfig.setInputs(Collections.singleton(sourceTopic));
+functionConfig.setClassName(ExclamationFunction.class.getName());
+functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+functionConfig.setOutput(sinkTopic);
+
+LocalRunner localRunner = LocalRunner.builder().functionConfig(functionConfig).build();
+localRunner.start(true);
+
+```
+
+So you can debug functions using an IDE easily. Set breakpoints and manually step through a function to debug with real data.
+
+The following example illustrates how to programmatically launch a function in localrun mode.
+
+```java
+
+public class ExclamationFunction implements Function<String, String> {
+
+ @Override
+ public String process(String s, Context context) throws Exception {
+ return s + "!";
+ }
+
+public static void main(String[] args) throws Exception {
+ FunctionConfig functionConfig = new FunctionConfig();
+ functionConfig.setName("exclamation");
+ functionConfig.setInputs(Collections.singleton("input"));
+ functionConfig.setClassName(ExclamationFunction.class.getName());
+ functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
+ functionConfig.setOutput("output");
+
+ LocalRunner localRunner = LocalRunner.builder().functionConfig(functionConfig).build();
+ localRunner.start(false);
+}
+
+```
+
+To use localrun mode programmatically, add the following dependency.
+
+```xml
+
+<dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-local-runner</artifactId>
+ <version>${pulsar.version}</version>
+</dependency>
+
+```
+
+For complete code samples, see [here](https://github.com/jerrypeng/pulsar-functions-demos/tree/master/debugging).
+
+:::note
+
+Debugging with localrun mode for Pulsar Functions written in other languages will be supported soon.
+
+:::
+
+## Use log topic
+
+In Pulsar Functions, you can generate log information defined in functions to a specified log topic. You can configure consumers to consume messages from a specified log topic to check the log information.
+
+
+
+**Example**
+
+```text
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+import org.slf4j.Logger;
+
+public class LoggingFunction implements Function<String, Void> {
+ @Override
+ public void apply(String input, Context context) {
+ Logger LOG = context.getLogger();
+ String messageId = new String(context.getMessageId());
+
+ if (input.contains("danger")) {
+ LOG.warn("A warning was received in message {}", messageId);
+ } else {
+ LOG.info("Message {} received\nContent: {}", messageId, input);
+ }
+
+ return null;
+ }
+}
+
+```
+
+As shown in the example above, you can get the logger via `context.getLogger()` and assign the logger to the `LOG` variable of `slf4j`, so you can define your desired log information in a function using the `LOG` variable. Meanwhile, you need to specify the topic to which the log information is produced.
+
+**Example**
+
+```
+
+$ bin/pulsar-admin functions create \
+ --log-topic persistent://public/default/logging-function-logs \
+ # Other function configs
+
+```
+
+## Use Functions CLI
+
+With [Pulsar Functions CLI](reference-pulsar-admin.md#functions), you can debug Pulsar Functions with the following subcommands:
+
+* `get`
+* `status`
+* `stats`
+* `list`
+* `trigger`
+
+:::tip
+
+For complete commands of **Pulsar Functions CLI**, see [here](reference-pulsar-admin.md#functions)。
+
+:::
+
+### `get`
+
+Get information about a Pulsar Function.
+
+**Usage**
+
+```text
+
+$ pulsar-admin functions get options
+
+```
+
+**Options**
+
+|Flag|Description
+|---|---
+|`--fqfn`|The Fully Qualified Function Name (FQFN) of a Pulsar Function.
+|`--name`|The name of a Pulsar Function.
+|`--namespace`|The namespace of a Pulsar Function.
+|`--tenant`|The tenant of a Pulsar Function.
+
+:::tip
+
+`--fqfn` consists of `--name`, `--namespace` and `--tenant`, so you can specify either `--fqfn` or `--name`, `--namespace` and `--tenant`.
+
+:::
+
+**Example**
+
+You can specify `--fqfn` to get information about a Pulsar Function.
+
+```text
+
+$ ./bin/pulsar-admin functions get public/default/ExclamationFunctio6
+
+```
+
+Optionally, you can specify `--name`, `--namespace` and `--tenant` to get information about a Pulsar Function.
+
+```text
+
+$ ./bin/pulsar-admin functions get \
+ --tenant public \
+ --namespace default \
+ --name ExclamationFunctio6
+
+```
+
+As shown below, the `get` command shows input, output, runtime, and other information about the _ExclamationFunctio6_ function.
+
+```text
+
+{
+ "tenant": "public",
+ "namespace": "default",
+ "name": "ExclamationFunctio6",
+ "className": "org.example.test.ExclamationFunction",
+ "inputSpecs": {
+ "persistent://public/default/my-topic-1": {
+ "isRegexPattern": false
+ }
+ },
+ "output": "persistent://public/default/test-1",
+ "processingGuarantees": "ATLEAST_ONCE",
+ "retainOrdering": false,
+ "userConfig": {},
+ "runtime": "JAVA",
+ "autoAck": true,
+ "parallelism": 1
+}
+
+```
+
+### `status`
+
+Check the current status of a Pulsar Function.
+
+**Usage**
+
+```text
+
+$ pulsar-admin functions status options
+
+```
+
+**Options**
+
+|Flag|Description
+|---|---
+|`--fqfn`|The Fully Qualified Function Name (FQFN) of a Pulsar Function.
+|`--instance-id`|The instance ID of a Pulsar Function <br />If the `--instance-id` is not specified, it gets the IDs of all instances.<br />
+|`--name`|The name of a Pulsar Function.
+|`--namespace`|The namespace of a Pulsar Function.
+|`--tenant`|The tenant of a Pulsar Function.
+
+**Example**
+
+```text
+
+$ ./bin/pulsar-admin functions status \
+ --tenant public \
+ --namespace default \
+ --name ExclamationFunctio6 \
+
+```
+
+As shown below, the `status` command shows the number of instances, running instances, the instance running under the _ExclamationFunctio6_ function, received messages, successfully processed messages, system exceptions, the average latency and so on.
+
+```text
+
+{
+ "numInstances" : 1,
+ "numRunning" : 1,
+ "instances" : [ {
+ "instanceId" : 0,
+ "status" : {
+ "running" : true,
+ "error" : "",
+ "numRestarts" : 0,
+ "numReceived" : 1,
+ "numSuccessfullyProcessed" : 1,
+ "numUserExceptions" : 0,
+ "latestUserExceptions" : [ ],
+ "numSystemExceptions" : 0,
+ "latestSystemExceptions" : [ ],
+ "averageLatency" : 0.8385,
+ "lastInvocationTime" : 1557734137987,
+ "workerId" : "c-standalone-fw-23ccc88ef29b-8080"
+ }
+ } ]
+}
+
+```
+
+### `stats`
+
+Get the current stats of a Pulsar Function.
+
+**Usage**
+
+```text
+
+$ pulsar-admin functions stats options
+
+```
+
+**Options**
+
+|Flag|Description
+|---|---
+|`--fqfn`|The Fully Qualified Function Name (FQFN) of a Pulsar Function.
+|`--instance-id`|The instance ID of a Pulsar Function. <br />If the `--instance-id` is not specified, it gets the IDs of all instances.<br />
+|`--name`|The name of a Pulsar Function.
+|`--namespace`|The namespace of a Pulsar Function.
+|`--tenant`|The tenant of a Pulsar Function.
+
+**Example**
+
+```text
+
+$ ./bin/pulsar-admin functions stats \
+ --tenant public \
+ --namespace default \
+ --name ExclamationFunctio6 \
+
+```
+
+The output is shown as follows:
+
+```text
+
+{
+ "receivedTotal" : 1,
+ "processedSuccessfullyTotal" : 1,
+ "systemExceptionsTotal" : 0,
+ "userExceptionsTotal" : 0,
+ "avgProcessLatency" : 0.8385,
+ "1min" : {
+ "receivedTotal" : 0,
+ "processedSuccessfullyTotal" : 0,
+ "systemExceptionsTotal" : 0,
+ "userExceptionsTotal" : 0,
+ "avgProcessLatency" : null
+ },
+ "lastInvocation" : 1557734137987,
+ "instances" : [ {
+ "instanceId" : 0,
+ "metrics" : {
+ "receivedTotal" : 1,
+ "processedSuccessfullyTotal" : 1,
+ "systemExceptionsTotal" : 0,
+ "userExceptionsTotal" : 0,
+ "avgProcessLatency" : 0.8385,
+ "1min" : {
+ "receivedTotal" : 0,
+ "processedSuccessfullyTotal" : 0,
+ "systemExceptionsTotal" : 0,
+ "userExceptionsTotal" : 0,
+ "avgProcessLatency" : null
+ },
+ "lastInvocation" : 1557734137987,
+ "userMetrics" : { }
+ }
+ } ]
+}
+
+```
+
+### `list`
+
+List all Pulsar Functions running under a specific tenant and namespace.
+
+**Usage**
+
+```text
+
+$ pulsar-admin functions list options
+
+```
+
+**Options**
+
+|Flag|Description
+|---|---
+|`--namespace`|The namespace of a Pulsar Function.
+|`--tenant`|The tenant of a Pulsar Function.
+
+**Example**
+
+```text
+
+$ ./bin/pulsar-admin functions list \
+ --tenant public \
+ --namespace default
+
+```
+
+As shown below, the `list` command returns three functions running under the _public_ tenant and the _default_ namespace.
+
+```text
+
+ExclamationFunctio1
+ExclamationFunctio2
+ExclamationFunctio3
+
+```
+
+### `trigger`
+
+Trigger a specified Pulsar Function with a supplied value. This command simulates the execution process of a Pulsar Function and verifies it.
+
+**Usage**
+
+```text
+
+$ pulsar-admin functions trigger options
+
+```
+
+**Options**
+
+|Flag|Description
+|---|---
+|`--fqfn`|The Fully Qualified Function Name (FQFN) of a Pulsar Function.
+|`--name`|The name of a Pulsar Function.
+|`--namespace`|The namespace of a Pulsar Function.
+|`--tenant`|The tenant of a Pulsar Function.
+|`--topic`|The topic name that a Pulsar Function consumes from.
+|`--trigger-file`|The path to a file that contains the data to trigger a Pulsar Function.
+|`--trigger-value`|The value to trigger a Pulsar Function.
+
+**Example**
+
+```text
+
+$ ./bin/pulsar-admin functions trigger \
+ --tenant public \
+ --namespace default \
+ --name ExclamationFunctio6 \
+ --topic persistent://public/default/my-topic-1 \
+ --trigger-value "hello pulsar functions"
+
+```
+
+As shown below, the `trigger` command returns the following result:
+
+```text
+
+This is my function!
+
+```
+
+:::note
+
+You must specify the [entire topic name](getting-started-pulsar.md#topic-names) when using the `--topic` option. Otherwise, the following error occurs.
+
+```text
+
+Function in trigger function has unidentified topic
+Reason: Function in trigger function has unidentified topic
+
+```
+
+:::
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/functions-deploy.md b/site2/website-next/versioned_docs/version-2.4.1/functions-deploy.md
new file mode 100644
index 0000000..8ae4be9
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/functions-deploy.md
@@ -0,0 +1,241 @@
+---
+id: functions-deploy
+title: Deploy Pulsar Functions
+sidebar_label: "How-to: Deploy"
+original_id: functions-deploy
+---
+
+## Requirements
+
+To deploy and manage Pulsar Functions, you need to have a Pulsar cluster running. There are several options for this:
+
+* You can run a [standalone cluster](getting-started-standalone) locally on your own machine.
+* You can deploy a Pulsar cluster on [Kubernetes](deploy-kubernetes.md), [Amazon Web Services](deploy-aws.md), [bare metal](deploy-bare-metal.md), [DC/OS](deploy-dcos), and more.
+
+If you run a non-[standalone](reference-terminology.md#standalone) cluster, you need to obtain the service URL for the cluster. How you obtain the service URL depends on how you deploy your Pulsar cluster.
+
+If you want to deploy and trigger Python user-defined functions, you need to install [the pulsar python client](http://pulsar.apache.org/docs/en/client-libraries-python/) on all the machines running [functions workers](functions-worker).
+
+## Command-line interface
+
+Pulsar Functions are deployed and managed using the [`pulsar-admin functions`](reference-pulsar-admin.md#functions) interface, which contains commands such as [`create`](reference-pulsar-admin.md#functions-create) for deploying functions in [cluster mode](#cluster-mode), [`trigger`](reference-pulsar-admin.md#trigger) for [triggering](#triggering-pulsar-functions) functions, [`list`](reference-pulsar-admin.md#list-2) for listing deployed functions.
+
+To learn more commands, refer to [`pulsar-admin functions`](reference-pulsar-admin.md#functions).
+
+### Default arguments
+
+When managing Pulsar Functions, you need to specify a variety of information about functions, including tenant, namespace, input and output topics, and so on. However, some parameters have default values if you do not specify values for them. The following table lists the default values.
+
+Parameter | Default
+:---------|:-------
+Function name | You can specify any value for the class name (except org, library, or similar class names). For example, when you specify the flag `--classname org.example.MyFunction`, the function name is `MyFunction`.
+Tenant | Derived from names of the input topics. If the input topics are under the `marketing` tenant, which means the topic names have the form `persistent://marketing/{namespace}/{topicName}`, the tenant is `marketing`.
+Namespace | Derived from names of the input topics. If the input topics are under the `asia` namespace under the `marketing` tenant, which means the topic names have the form `persistent://marketing/asia/{topicName}`, then the namespace is `asia`.
+Output topic | `{input topic}-{function name}-output`. For example, if an input topic name of a function is `incoming`, and the function name is `exclamation`, then the name of the output topic is `incoming-exclamation-output`.
+Subscription type | For `at-least-once` and `at-most-once` [processing guarantees](functions-overview.md#processing-guarantees), the [`SHARED`](concepts-messaging.md#shared) mode is applied by default; for `effectively-once` guarantees, the [`FAILOVER`](concepts-messaging.md#failover) mode is applied.
+Processing guarantees | [`ATLEAST_ONCE`](functions-overview.md#processing-guarantees)
+Pulsar service URL | `pulsar://localhost:6650`
+
+### Example of default arguments
+
+Take the `create` command as an example.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --jar my-pulsar-functions.jar \
+ --classname org.example.MyFunction \
+ --inputs my-function-input-topic1,my-function-input-topic2
+
+```
+
+The function has default values for the function name (`MyFunction`), tenant (`public`), namespace (`default`), subscription type (`SHARED`), processing guarantees (`ATLEAST_ONCE`), and Pulsar service URL (`pulsar://localhost:6650`).
+
+## Local run mode
+
+If you run a Pulsar Function in **local run** mode, it runs on the machine from which you enter the commands (on your laptop, an [AWS EC2](https://aws.amazon.com/ec2/) instance, and so on). The following is a [`localrun`](reference-pulsar-admin.md#localrun) command example.
+
+```bash
+
+$ bin/pulsar-admin functions localrun \
+ --py myfunc.py \
+ --classname myfunc.SomeFunction \
+ --inputs persistent://public/default/input-1 \
+ --output persistent://public/default/output-1
+
+```
+
+By default, the function connects to a Pulsar cluster running on the same machine, via a local [broker](reference-terminology.md#broker) service URL of `pulsar://localhost:6650`. If you use local run mode to run a function but connect it to a non-local Pulsar cluster, you can specify a different broker URL using the `--brokerServiceUrl` flag. The following is an example.
+
+```bash
+
+$ bin/pulsar-admin functions localrun \
+ --broker-service-url pulsar://my-cluster-host:6650 \
+ # Other function parameters
+
+```
+
+## Cluster mode
+
+When you run a Pulsar Function in **cluster** mode, the function code is uploaded to a Pulsar broker and runs *alongside the broker* rather than in your [local environment](#local-run-mode). You can run a function in cluster mode using the [`create`](reference-pulsar-admin.md#create-1) command.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --py myfunc.py \
+ --classname myfunc.SomeFunction \
+ --inputs persistent://public/default/input-1 \
+ --output persistent://public/default/output-1
+
+```
+
+### Update functions in cluster mode
+
+You can use the [`update`](reference-pulsar-admin.md#update-1) command to update a Pulsar Function running in cluster mode. The following command updates the function created in the [cluster mode](#cluster-mode) section.
+
+```bash
+
+$ bin/pulsar-admin functions update \
+ --py myfunc.py \
+ --classname myfunc.SomeFunction \
+ --inputs persistent://public/default/new-input-topic \
+ --output persistent://public/default/new-output-topic
+
+```
+
+### Parallelism
+
+Pulsar Functions run as processes or threads, which are called **instances**. When you run a Pulsar Function, it runs as a single instance by default. With one localrun command, you can only run a single instance of a function. If you want to run multiple instances, you can use localrun command multiple times.
+
+When you create a function, you can specify the *parallelism* of a function (the number of instances to run). You can set the parallelism factor using the `--parallelism` flag of the [`create`](reference-pulsar-admin.md#functions-create) command.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --parallelism 3 \
+ # Other function info
+
+```
+
+You can adjust the parallelism of an already created function using the [`update`](reference-pulsar-admin.md#update-1) interface.
+
+```bash
+
+$ bin/pulsar-admin functions update \
+ --parallelism 5 \
+ # Other function
+
+```
+
+If you specify a function configuration via YAML, use the `parallelism` parameter. The following is a config file example.
+
+```yaml
+
+# function-config.yaml
+parallelism: 3
+inputs:
+- persistent://public/default/input-1
+output: persistent://public/default/output-1
+# other parameters
+
+```
+
+The following is corresponding update command.
+
+```bash
+
+$ bin/pulsar-admin functions update \
+ --function-config-file function-config.yaml
+
+```
+
+### Function instance resources
+
+When you run Pulsar Functions in [cluster mode](#cluster-mode), you can specify the resources that are assigned to each function [instance](#parallelism).
+
+Resource | Specified as | Runtimes
+:--------|:----------------|:--------
+CPU | The number of cores | Kubernetes
+RAM | The number of bytes | Process, Docker
+Disk space | The number of bytes | Docker
+
+The following function creation command allocates 8 cores, 8 GB of RAM, and 10 GB of disk space to a function.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --jar target/my-functions.jar \
+ --classname org.example.functions.MyFunction \
+ --cpu 8 \
+ --ram 8589934592 \
+ --disk 10737418240
+
+```
+
+> #### Resources are *per instance*
+> The resources that you apply to a given Pulsar Function are applied to each instance of the function. For example, if you apply 8 GB of RAM to a function with a parallelism of 5, you are applying 40 GB of RAM for the function in total. Make sure that you take the parallelism (the number of instances) factor into your resource calculations.
+
+## Trigger Pulsar Functions
+
+If a Pulsar Function is running in [cluster mode](#cluster-mode), you can **trigger** it at any time using the command line. Triggering a function means that you send a message with a specific value to the function and get the function output (if any) via the command line.
+
+> Triggering a function is to invoke a function by producing a message on one of the input topics. With the [`pulsar-admin functions trigger`](reference-pulsar-admin.md#trigger) command, you can send messages to functions without using the [`pulsar-client`](reference-cli-tools.md#pulsar-client) tool or a language-specific client library.
+
+To learn how to trigger a function, you can start with Python function that returns a simple string based on the input.
+
+```python
+
+# myfunc.py
+def process(input):
+ return "This function has been triggered with a value of {0}".format(input)
+
+```
+
+You can run the function in [local run mode](functions-deploy.md#local-run-mode).
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --tenant public \
+ --namespace default \
+ --name myfunc \
+ --py myfunc.py \
+ --classname myfunc \
+ --inputs persistent://public/default/in \
+ --output persistent://public/default/out
+
+```
+
+Then assign a consumer to listen on the output topic for messages from the `myfunc` function with the [`pulsar-client consume`](reference-cli-tools.md#consume) command.
+
+```bash
+
+$ bin/pulsar-client consume persistent://public/default/out \
+ --subscription-name my-subscription
+ --num-messages 0 # Listen indefinitely
+
+```
+
+And then you can trigger the function.
+
+```bash
+
+$ bin/pulsar-admin functions trigger \
+ --tenant public \
+ --namespace default \
+ --name myfunc \
+ --trigger-value "hello world"
+
+```
+
+The consumer listening on the output topic produces something as follows in the log.
+
+```
+
+----- got message -----
+This function has been triggered with a value of hello world
+
+```
+
+> #### Topic info is not required
+> In the `trigger` command, you only need to specify basic information about the function (tenant, namespace, and name). To trigger the function, you do not need to know the function input topics.
diff --git a/site2/website-next/versioned_docs/version-2.4.1/functions-develop.md b/site2/website-next/versioned_docs/version-2.4.1/functions-develop.md
new file mode 100644
index 0000000..c2f5a53
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/functions-develop.md
@@ -0,0 +1,1144 @@
+---
+id: functions-develop
+title: Develop Pulsar Functions
+sidebar_label: "How-to: Develop"
+original_id: functions-develop
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+This tutorial walks you through how to develop Pulsar Functions.
+
+## Available APIs
+In Java and Python, you have two options to write Pulsar Functions. In Go, you can use Pulsar Functions SDK for Go.
+
+Interface | Description | Use cases
+:---------|:------------|:---------
+Language-native interface | No Pulsar-specific libraries or special dependencies required (only core libraries from Java/Python). | Functions that do not require access to the function [context](#context).
+Pulsar Function SDK for Java/Python/Go | Pulsar-specific libraries that provide a range of functionality not provided by "native" interfaces. | Functions that require access to the function [context](#context).
+
+The language-native function, which adds an exclamation point to all incoming strings and publishes the resulting string to a topic, has no external dependencies. The following example is language-native function.
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"}]}>
+<TabItem value="Java">
+
+```Java
+
+public class JavaNativeExclamationFunction implements Function<String, String> {
+ @Override
+ public String apply(String input) {
+ return String.format("%s!", input);
+ }
+}
+
+```
+
+For complete code, see [here](https://github.com/apache/pulsar/blob/master/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/JavaNativeExclamationFunction.java).
+
+</TabItem>
+<TabItem value="Python">
+
+```python
+
+def process(input):
+ return "{}!".format(input)
+
+```
+
+For complete code, see [here](https://github.com/apache/pulsar/blob/master/pulsar-functions/python-examples/native_exclamation_function.py).
+
+:::note
+
+You can write Pulsar Functions in python2 or python3. However, Pulsar only looks for `python` as the interpreter.
+If you're running Pulsar Functions on an Ubuntu system that only supports python3, you might fail to start the functions. In this case, you can create a symlink. Your system will fail if you subsequently install any other package that depends on Python 2.x. A solution is under development in [Issue 5518](https://github.com/apache/pulsar/issues/5518).
+
+```bash
+
+sudo update-alternatives --install /usr/bin/python python /usr/bin/python3 10
+
+```
+
+:::
+
+</TabItem>
+
+</Tabs>
+
+The following example uses Pulsar Functions SDK.
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"},{"label":"Go","value":"Go"}]}>
+<TabItem value="Java">
+
+```Java
+
+public class ExclamationFunction implements Function<String, String> {
+ @Override
+ public String process(String input, Context context) {
+ return String.format("%s!", input);
+ }
+}
+
+```
+
+For complete code, see [here](https://github.com/apache/pulsar/blob/master/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/ExclamationFunction.java).
+
+</TabItem>
+<TabItem value="Python">
+
+```python
+
+from pulsar import Function
+
+class ExclamationFunction(Function):
+ def __init__(self):
+ pass
+
+ def process(self, input, context):
+ return input + '!'
+
+```
+
+For complete code, see [here](https://github.com/apache/pulsar/blob/master/pulsar-functions/python-examples/exclamation_function.py).
+
+</TabItem>
+<TabItem value="Go">
+
+```Go
+
+package main
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/apache/pulsar/pulsar-function-go/pf"
+)
+
+func HandleRequest(ctx context.Context, in []byte) error{
+ fmt.Println(string(in) + "!")
+ return nil
+}
+
+func main() {
+ pf.Start(HandleRequest)
+}
+
+```
+
+For complete code, see [here](https://github.com/apache/pulsar/blob/77cf09eafa4f1626a53a1fe2e65dd25f377c1127/pulsar-function-go/examples/inputFunc/inputFunc.go#L20-L36).
+
+</TabItem>
+
+</Tabs>
+
+## Schema registry
+Pulsar has a built in schema registry and comes bundled with a variety of popular schema types(avro, json and protobuf). Pulsar Functions can leverage existing schema information from input topics and derive the input type. The schema registry applies for output topic as well.
+
+## SerDe
+SerDe stands for **Ser**ialization and **De**serialization. Pulsar Functions uses SerDe when publishing data to and consuming data from Pulsar topics. How SerDe works by default depends on the language you use for a particular function.
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"}]}>
+<TabItem value="Java">
+
+When you write Pulsar Functions in Java, the following basic Java types are built in and supported by default:
+
+* `String`
+* `Double`
+* `Integer`
+* `Float`
+* `Long`
+* `Short`
+* `Byte`
+
+To customize Java types, you need to implement the following interface.
+
+```java
+
+public interface SerDe<T> {
+ T deserialize(byte[] input);
+ byte[] serialize(T input);
+}
+
+```
+
+</TabItem>
+<TabItem value="Python">
+
+In Python, the default SerDe is identity, meaning that the type is serialized as whatever type the producer function returns.
+
+You can specify the SerDe when [creating](functions-deploy.md#cluster-mode) or [running](functions-deploy.md#local-run-mode) functions.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --tenant public \
+ --namespace default \
+ --name my_function \
+ --py my_function.py \
+ --classname my_function.MyFunction \
+ --custom-serde-inputs '{"input-topic-1":"Serde1","input-topic-2":"Serde2"}' \
+ --output-serde-classname Serde3 \
+ --output output-topic-1
+
+```
+
+This case contains two input topics: `input-topic-1` and `input-topic-2`, each of which is mapped to a different SerDe class (the map must be specified as a JSON string). The output topic, `output-topic-1`, uses the `Serde3` class for SerDe. At the moment, all Pulsar Functions logic, include processing function and SerDe classes, must be contained within a single Python file.
+
+When using Pulsar Functions for Python, you have three SerDe options:
+
+1. You can use the [`IdentitySerde`](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/python/pulsar/functions/serde.py#L70), which leaves the data unchanged. The `IdentitySerDe` is the **default**. Creating or running a function without explicitly specifying SerDe means that this option is used.
+2. You can use the [`PickleSerDe`](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/python/pulsar/functions/serde.py#L62), which uses Python [`pickle`](https://docs.python.org/3/library/pickle.html) for SerDe.
+3. You can create a custom SerDe class by implementing the baseline [`SerDe`](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/python/pulsar/functions/serde.py#L50) class, which has just two methods: [`serialize`](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/python/pulsar/functions/serde.py#L53) for converting the object into bytes, and [`deserialize`](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/python/pulsar/functions/serde.py#L58) for [...]
+
+The table below shows when you should use each SerDe.
+
+SerDe option | When to use
+:------------|:-----------
+`IdentitySerde` | When you work with simple types like strings, Booleans, integers.
+`PickleSerDe` | When you work with complex, application-specific types and are comfortable with the "best effort" approach of `pickle`.
+Custom SerDe | When you require explicit control over SerDe, potentially for performance or data compatibility purposes.
+
+</TabItem>
+
+</Tabs>
+
+### Example
+Imagine that you're writing Pulsar Functions that are processing tweet objects, you can refer to the following example of `Tweet` class.
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"}]}>
+<TabItem value="Java">
+
+```java
+
+public class Tweet {
+ private String username;
+ private String tweetContent;
+
+ public Tweet(String username, String tweetContent) {
+ this.username = username;
+ this.tweetContent = tweetContent;
+ }
+
+ // Standard setters and getters
+}
+
+```
+
+To pass `Tweet` objects directly between Pulsar Functions, you need to provide a custom SerDe class. In the example below, `Tweet` objects are basically strings in which the username and tweet content are separated by a `|`.
+
+```java
+
+package com.example.serde;
+
+import org.apache.pulsar.functions.api.SerDe;
+
+import java.util.regex.Pattern;
+
+public class TweetSerde implements SerDe<Tweet> {
+ public Tweet deserialize(byte[] input) {
+ String s = new String(input);
+ String[] fields = s.split(Pattern.quote("|"));
+ return new Tweet(fields[0], fields[1]);
+ }
+
+ public byte[] serialize(Tweet input) {
+ return "%s|%s".format(input.getUsername(), input.getTweetContent()).getBytes();
+ }
+}
+
+```
+
+To apply this customized SerDe to a particular Pulsar Function, you need to:
+
+* Package the `Tweet` and `TweetSerde` classes into a JAR.
+* Specify a path to the JAR and SerDe class name when deploying the function.
+
+The following is an example of [`create`](reference-pulsar-admin.md#create-1) operation.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --jar /path/to/your.jar \
+ --output-serde-classname com.example.serde.TweetSerde \
+ # Other function attributes
+
+```
+
+> #### Custom SerDe classes must be packaged with your function JARs
+> Pulsar does not store your custom SerDe classes separately from your Pulsar Functions. So you need to include your SerDe classes in your function JARs. If not, Pulsar returns an error.
+
+</TabItem>
+<TabItem value="Python">
+
+```python
+
+class Tweet(object):
+ def __init__(self, username, tweet_content):
+ self.username = username
+ self.tweet_content = tweet_content
+
+```
+
+In order to use this class in Pulsar Functions, you have two options:
+
+1. You can specify `PickleSerDe`, which applies the [`pickle`](https://docs.python.org/3/library/pickle.html) library SerDe.
+2. You can create your own SerDe class. The following is an example.
+
+ ```python
+
+ from pulsar import SerDe
+
+ class TweetSerDe(SerDe):
+
+ def serialize(self, input):
+ return bytes("{0}|{1}".format(input.username, input.tweet_content))
+
+ def deserialize(self, input_bytes):
+ tweet_components = str(input_bytes).split('|')
+ return Tweet(tweet_components[0], tweet_componentsp[1])
+
+ ```
+
+For complete code, see [here](https://github.com/apache/pulsar/blob/master/pulsar-functions/python-examples/custom_object_function.py).
+
+</TabItem>
+
+</Tabs>
+
+In both languages, however, you can write custom SerDe logic for more complex, application-specific types.
+
+## Context
+Java, Python and Go SDKs provide access to a **context object** that can be used by a function. This context object provides a wide variety of information and functionality to the function.
+
+* The name and ID of a Pulsar Function.
+* The message ID of each message. Each Pulsar message is automatically assigned with an ID.
+* The key, event time, properties and partition key of each message.
+* The name of the topic to which the message is sent.
+* The names of all input topics as well as the output topic associated with the function.
+* The name of the class used for [SerDe](#serde).
+* The [tenant](reference-terminology.md#tenant) and namespace associated with the function.
+* The ID of the Pulsar Functions instance running the function.
+* The version of the function.
+* The [logger object](functions-develop.md#logger) used by the function, which can be used to create function log messages.
+* Access to arbitrary [user configuration](#user-config) values supplied via the CLI.
+* An interface for recording [metrics](#metrics).
+* An interface for storing and retrieving state in [state storage](#state-storage).
+* A function to publish new messages onto arbitrary topics.
+* A function to ack the message being processed (if auto-ack is disabled).
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"},{"label":"Go","value":"Go"}]}>
+<TabItem value="Java">
+
+The [Context](https://github.com/apache/pulsar/blob/master/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java) interface provides a number of methods that you can use to access the function [context](#context). The various method signatures for the `Context` interface are listed as follows.
+
+```java
+
+public interface Context {
+ Record<?> getCurrentRecord();
+ Collection<String> getInputTopics();
+ String getOutputTopic();
+ String getOutputSchemaType();
+ String getTenant();
+ String getNamespace();
+ String getFunctionName();
+ String getFunctionId();
+ String getInstanceId();
+ String getFunctionVersion();
+ Logger getLogger();
+ void incrCounter(String key, long amount);
+ void incrCounterAsync(String key, long amount);
+ long getCounter(String key);
+ long getCounterAsync(String key);
+ void putState(String key, ByteBuffer value);
+ void putStateAsync(String key, ByteBuffer value);
+ void deleteState(String key);
+ ByteBuffer getState(String key);
+ ByteBuffer getStateAsync(String key);
+ Map<String, Object> getUserConfigMap();
+ Optional<Object> getUserConfigValue(String key);
+ Object getUserConfigValueOrDefault(String key, Object defaultValue);
+ void recordMetric(String metricName, double value);
+ <O> CompletableFuture<Void> publish(String topicName, O object, String schemaOrSerdeClassName);
+ <O> CompletableFuture<Void> publish(String topicName, O object);
+ <O> TypedMessageBuilder<O> newOutputMessage(String topicName, Schema<O> schema) throws PulsarClientException;
+}
+
+```
+
+The following example uses several methods available via the `Context` object.
+
+```java
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+import org.slf4j.Logger;
+
+import java.util.stream.Collectors;
+
+public class ContextFunction implements Function<String, Void> {
+ public Void process(String input, Context context) {
+ Logger LOG = context.getLogger();
+ String inputTopics = context.getInputTopics().stream().collect(Collectors.joining(", "));
+ String functionName = context.getFunctionName();
+
+ String logMessage = String.format("A message with a value of \"%s\" has arrived on one of the following topics: %s\n",
+ input,
+ inputTopics);
+
+ LOG.info(logMessage);
+
+ String metricName = String.format("function-%s-messages-received", functionName);
+ context.recordMetric(metricName, 1);
+
+ return null;
+ }
+}
+
+```
+
+</TabItem>
+<TabItem value="Python">
+
+```
+
+class ContextImpl(pulsar.Context):
+ def get_message_id(self):
+ ...
+ def get_message_key(self):
+ ...
+ def get_message_eventtime(self):
+ ...
+ def get_message_properties(self):
+ ...
+ def get_current_message_topic_name(self):
+ ...
+ def get_partition_key(self):
+ ...
+ def get_function_name(self):
+ ...
+ def get_function_tenant(self):
+ ...
+ def get_function_namespace(self):
+ ...
+ def get_function_id(self):
+ ...
+ def get_instance_id(self):
+ ...
+ def get_function_version(self):
+ ...
+ def get_logger(self):
+ ...
+ def get_user_config_value(self, key):
+ ...
+ def get_user_config_map(self):
+ ...
+ def record_metric(self, metric_name, metric_value):
+ ...
+ def get_input_topics(self):
+ ...
+ def get_output_topic(self):
+ ...
+ def get_output_serde_class_name(self):
+ ...
+ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe",
+ properties=None, compression_type=None, callback=None, message_conf=None):
+ ...
+ def ack(self, msgid, topic):
+ ...
+ def get_and_reset_metrics(self):
+ ...
+ def reset_metrics(self):
+ ...
+ def get_metrics(self):
+ ...
+ def incr_counter(self, key, amount):
+ ...
+ def get_counter(self, key):
+ ...
+ def del_counter(self, key):
+ ...
+ def put_state(self, key, value):
+ ...
+ def get_state(self, key):
+ ...
+
+```
+
+</TabItem>
+<TabItem value="Go">
+
+```
+
+func (c *FunctionContext) GetInstanceID() int {
+ return c.instanceConf.instanceID
+}
+
+func (c *FunctionContext) GetInputTopics() []string {
+ return c.inputTopics
+}
+
+func (c *FunctionContext) GetOutputTopic() string {
+ return c.instanceConf.funcDetails.GetSink().Topic
+}
+
+func (c *FunctionContext) GetFuncTenant() string {
+ return c.instanceConf.funcDetails.Tenant
+}
+
+func (c *FunctionContext) GetFuncName() string {
+ return c.instanceConf.funcDetails.Name
+}
+
+func (c *FunctionContext) GetFuncNamespace() string {
+ return c.instanceConf.funcDetails.Namespace
+}
+
+func (c *FunctionContext) GetFuncID() string {
+ return c.instanceConf.funcID
+}
+
+func (c *FunctionContext) GetFuncVersion() string {
+ return c.instanceConf.funcVersion
+}
+
+func (c *FunctionContext) GetUserConfValue(key string) interface{} {
+ return c.userConfigs[key]
+}
+
+func (c *FunctionContext) GetUserConfMap() map[string]interface{} {
+ return c.userConfigs
+}
+
+```
+
+The following example uses several methods available via the `Context` object.
+
+```
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/apache/pulsar/pulsar-function-go/pf"
+)
+
+func contextFunc(ctx context.Context) {
+ if fc, ok := pf.FromContext(ctx); ok {
+ fmt.Printf("function ID is:%s, ", fc.GetFuncID())
+ fmt.Printf("function version is:%s\n", fc.GetFuncVersion())
+ }
+}
+
+```
+
+For complete code, see [here](https://github.com/apache/pulsar/blob/77cf09eafa4f1626a53a1fe2e65dd25f377c1127/pulsar-function-go/examples/contextFunc/contextFunc.go#L29-L34).
+
+</TabItem>
+
+</Tabs>
+
+### User config
+When you run or update Pulsar Functions created using SDK, you can pass arbitrary key/values to them with the command line with the `--userConfig` flag. Key/values must be specified as JSON. The following function creation command passes a user configured key/value to a function.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --name word-filter \
+ # Other function configs
+ --user-config '{"forbidden-word":"rosebud"}'
+
+```
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"}]}>
+<TabItem value="Java">
+
+The Java SDK [`Context`](#context) object enables you to access key/value pairs provided to Pulsar Functions via the command line (as JSON). The following example passes a key/value pair.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ # Other function configs
+ --user-config '{"word-of-the-day":"verdure"}'
+
+```
+
+To access that value in a Java function:
+
+```java
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+import org.slf4j.Logger;
+
+import java.util.Optional;
+
+public class UserConfigFunction implements Function<String, Void> {
+ @Override
+ public void apply(String input, Context context) {
+ Logger LOG = context.getLogger();
+ Optional<String> wotd = context.getUserConfigValue("word-of-the-day");
+ if (wotd.isPresent()) {
+ LOG.info("The word of the day is {}", wotd);
+ } else {
+ LOG.warn("No word of the day provided");
+ }
+ return null;
+ }
+}
+
+```
+
+The `UserConfigFunction` function will log the string `"The word of the day is verdure"` every time the function is invoked (which means every time a message arrives). The `word-of-the-day` user config will be changed only when the function is updated with a new config value via the command line.
+
+You can also access the entire user config map or set a default value in case no value is present:
+
+```java
+
+// Get the whole config map
+Map<String, String> allConfigs = context.getUserConfigMap();
+
+// Get value or resort to default
+String wotd = context.getUserConfigValueOrDefault("word-of-the-day", "perspicacious");
+
+```
+
+> For all key/value pairs passed to Java functions, both the key *and* the value are `String`. To set the value to be a different type, you need to deserialize from the `String` type.
+
+</TabItem>
+<TabItem value="Python">
+
+In Python function, you can access the configuration value like this.
+
+```python
+
+from pulsar import Function
+
+class WordFilter(Function):
+ def process(self, context, input):
+ forbidden_word = context.user_config()["forbidden-word"]
+
+ # Don't publish the message if it contains the user-supplied
+ # forbidden word
+ if forbidden_word in input:
+ pass
+ # Otherwise publish the message
+ else:
+ return input
+
+```
+
+The Python SDK [`Context`](#context) object enables you to access key/value pairs provided to Pulsar Functions via the command line (as JSON). The following example passes a key/value pair.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ # Other function configs \
+ --user-config '{"word-of-the-day":"verdure"}'
+
+```
+
+To access that value in a Python function:
+
+```python
+
+from pulsar import Function
+
+class UserConfigFunction(Function):
+ def process(self, input, context):
+ logger = context.get_logger()
+ wotd = context.get_user_config_value('word-of-the-day')
+ if wotd is None:
+ logger.warn('No word of the day provided')
+ else:
+ logger.info("The word of the day is {0}".format(wotd))
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Logger
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"},{"label":"Go","value":"Go"}]}>
+<TabItem value="Java">
+
+Pulsar Functions that use the Java SDK have access to an [SLF4j](https://www.slf4j.org/) [`Logger`](https://www.slf4j.org/api/org/apache/log4j/Logger.html) object that can be used to produce logs at the chosen log level. The following example logs either a `WARNING`- or `INFO`-level log based on whether the incoming string contains the word `danger`.
+
+```java
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+import org.slf4j.Logger;
+
+public class LoggingFunction implements Function<String, Void> {
+ @Override
+ public void apply(String input, Context context) {
+ Logger LOG = context.getLogger();
+ String messageId = new String(context.getMessageId());
+
+ if (input.contains("danger")) {
+ LOG.warn("A warning was received in message {}", messageId);
+ } else {
+ LOG.info("Message {} received\nContent: {}", messageId, input);
+ }
+
+ return null;
+ }
+}
+
+```
+
+If you want your function to produce logs, you need to specify a log topic when creating or running the function. The following is an example.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --jar my-functions.jar \
+ --classname my.package.LoggingFunction \
+ --log-topic persistent://public/default/logging-function-logs \
+ # Other function configs
+
+```
+
+All logs produced by `LoggingFunction` above can be accessed via the `persistent://public/default/logging-function-logs` topic.
+
+</TabItem>
+<TabItem value="Python">
+
+Pulsar Functions that use the Python SDK have access to a logging object that can be used to produce logs at the chosen log level. The following example function that logs either a `WARNING`- or `INFO`-level log based on whether the incoming string contains the word `danger`.
+
+```python
+
+from pulsar import Function
+
+class LoggingFunction(Function):
+ def process(self, input, context):
+ logger = context.get_logger()
+ msg_id = context.get_message_id()
+ if 'danger' in input:
+ logger.warn("A warning was received in message {0}".format(context.get_message_id()))
+ else:
+ logger.info("Message {0} received\nContent: {1}".format(msg_id, input))
+
+```
+
+If you want your function to produce logs on a Pulsar topic, you need to specify a **log topic** when creating or running the function. The following is an example.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --py logging_function.py \
+ --classname logging_function.LoggingFunction \
+ --log-topic logging-function-logs \
+ # Other function configs
+
+```
+
+All logs produced by `LoggingFunction` above can be accessed via the `logging-function-logs` topic.
+
+</TabItem>
+<TabItem value="Go">
+
+The following Go Function example shows different log levels based on the function input.
+
+```
+
+import (
+ "context"
+
+ "github.com/apache/pulsar/pulsar-function-go/log"
+ "github.com/apache/pulsar/pulsar-function-go/pf"
+)
+
+func loggerFunc(ctx context.Context, input []byte) {
+ if len(input) <= 100 {
+ log.Infof("This input has a length of: %d", len(input))
+ } else {
+ log.Warnf("This input is getting too long! It has {%d} characters", len(input))
+ }
+}
+
+func main() {
+ pf.Start(loggerFunc)
+}
+
+```
+
+When you use `logTopic` related functionalities in Go Function, import `github.com/apache/pulsar/pulsar-function-go/log`, and you do not have to use the `getLogger()` context object.
+
+</TabItem>
+
+</Tabs>
+
+## Metrics
+Pulsar Functions can publish arbitrary metrics to the metrics interface which can be queried.
+
+> If a Pulsar Function uses the language-native interface for Java or Python, that function is not able to publish metrics and stats to Pulsar.
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"}]}>
+<TabItem value="Java">
+
+You can record metrics using the [`Context`](#context) object on a per-key basis. For example, you can set a metric for the `process-count` key and a different metric for the `elevens-count` key every time the function processes a message.
+
+```java
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+
+public class MetricRecorderFunction implements Function<Integer, Void> {
+ @Override
+ public void apply(Integer input, Context context) {
+ // Records the metric 1 every time a message arrives
+ context.recordMetric("hit-count", 1);
+
+ // Records the metric only if the arriving number equals 11
+ if (input == 11) {
+ context.recordMetric("elevens-count", 1);
+ }
+
+ return null;
+ }
+}
+
+```
+
+> For instructions on reading and using metrics, see the [Monitoring](deploy-monitoring) guide.
+
+</TabItem>
+<TabItem value="Python">
+
+You can record metrics using the [`Context`](#context) object on a per-key basis. For example, you can set a metric for the `process-count` key and a different metric for the `elevens-count` key every time the function processes a message. The following is an example.
+
+```python
+
+from pulsar import Function
+
+class MetricRecorderFunction(Function):
+ def process(self, input, context):
+ context.record_metric('hit-count', 1)
+
+ if input == 11:
+ context.record_metric('elevens-count', 1)
+
+```
+
+</TabItem>
+
+</Tabs>
+
+### Access metrics
+To access metrics created by Pulsar Functions, refer to [Monitoring](deploy-monitoring) in Pulsar.
+
+## State storage
+Pulsar Functions use [Apache BookKeeper](https://bookkeeper.apache.org) as a state storage interface. Pulsar installation, including the local standalone installation, includes deployment of BookKeeper bookies.
+
+Since Pulsar 2.1.0 release, Pulsar integrates with Apache BookKeeper [table service](https://docs.google.com/document/d/155xAwWv5IdOitHh1NVMEwCMGgB28M3FyMiQSxEpjE-Y/edit#heading=h.56rbh52koe3f) to store the `State` for functions. For example, a `WordCount` function can store its `counters` state into BookKeeper table service via Pulsar Functions State API.
+
+States are key-value pairs, where the key is a string and the value is arbitrary binary data - counters are stored as 64-bit big-endian binary values. Keys are scoped to an individual Pulsar Function, and shared between instances of that function.
+
+You can access states within Pulsar Java Functions using the `putState`, `putStateAsync`, `getState`, `getStateAsync`, `incrCounter`, `incrCounterAsync`, `getCounter`, `getCounterAsync` and `deleteState` calls on the context object. You can access states within Pulsar Python Functions using the `putState`, `getState`, `incrCounter`, `getCounter` and `deleteState` calls on the context object. You can also manage states using the [querystate](#query-state) and [putstate](#putstate) option [...]
+
+### API
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"}]}>
+<TabItem value="Java">
+
+Currently Pulsar Functions expose the following APIs for mutating and accessing State. These APIs are available in the [Context](functions-develop.md#context) object when you are using Java SDK functions.
+
+#### incrCounter
+
+```java
+
+ /**
+ * Increment the builtin distributed counter referred by key
+ * @param key The name of the key
+ * @param amount The amount to be incremented
+ */
+ void incrCounter(String key, long amount);
+
+```
+
+The application can use `incrCounter` to change the counter of a given `key` by the given `amount`.
+
+#### incrCounterAsync
+
+```java
+
+ /**
+ * Increment the builtin distributed counter referred by key
+ * but dont wait for the completion of the increment operation
+ *
+ * @param key The name of the key
+ * @param amount The amount to be incremented
+ */
+ CompletableFuture<Void> incrCounterAsync(String key, long amount);
+
+```
+
+The application can use `incrCounterAsync` to asynchronously change the counter of a given `key` by the given `amount`.
+
+#### getCounter
+
+```java
+
+ /**
+ * Retrieve the counter value for the key.
+ *
+ * @param key name of the key
+ * @return the amount of the counter value for this key
+ */
+ long getCounter(String key);
+
+```
+
+The pplication can use `getCounter` to retrieve the counter of a given `key` mutated by `incrCounter`.
+
+Except the `counter` API, Pulsar also exposes a general key/value API for functions to store
+general key/value state.
+
+#### getCounterAsync
+
+```java
+
+ /**
+ * Retrieve the counter value for the key, but don't wait
+ * for the operation to be completed
+ *
+ * @param key name of the key
+ * @return the amount of the counter value for this key
+ */
+ CompletableFuture<Long> getCounterAsync(String key);
+
+```
+
+The application can use `getCounterAsync` to asynchronously retrieve the counter of a given `key` mutated by `incrCounterAsync`.
+
+#### putState
+
+```java
+
+ /**
+ * Update the state value for the key.
+ *
+ * @param key name of the key
+ * @param value state value of the key
+ */
+ void putState(String key, ByteBuffer value);
+
+```
+
+#### putStateAsync
+
+```java
+
+ /**
+ * Update the state value for the key, but don't wait for the operation to be completed
+ *
+ * @param key name of the key
+ * @param value state value of the key
+ */
+ CompletableFuture<Void> putStateAsync(String key, ByteBuffer value);
+
+```
+
+The application can use `putStateAsync` to asynchronously update the state of a given `key`.
+
+#### getState
+
+```java
+
+ /**
+ * Retrieve the state value for the key.
+ *
+ * @param key name of the key
+ * @return the state value for the key.
+ */
+ ByteBuffer getState(String key);
+
+```
+
+#### getStateAsync
+
+```java
+
+ /**
+ * Retrieve the state value for the key, but don't wait for the operation to be completed
+ *
+ * @param key name of the key
+ * @return the state value for the key.
+ */
+ CompletableFuture<ByteBuffer> getStateAsync(String key);
+
+```
+
+The application can use `getStateAsync` to asynchronously retrieve the state of a given `key`.
+
+#### deleteState
+
+```java
+
+ /**
+ * Delete the state value for the key.
+ *
+ * @param key name of the key
+ */
+
+```
+
+Counters and binary values share the same keyspace, so this deletes either type.
+
+</TabItem>
+<TabItem value="Python">
+
+Currently Pulsar Functions expose the following APIs for mutating and accessing State. These APIs are available in the [Context](#context) object when you are using Python SDK functions.
+
+#### incr_counter
+
+```python
+
+ def incr_counter(self, key, amount):
+ """incr the counter of a given key in the managed state"""
+
+```
+
+Application can use `incr_counter` to change the counter of a given `key` by the given `amount`.
+If the `key` does not exist, a new key is created.
+
+#### get_counter
+
+```python
+
+ def get_counter(self, key):
+ """get the counter of a given key in the managed state"""
+
+```
+
+Application can use `get_counter` to retrieve the counter of a given `key` mutated by `incrCounter`.
+
+Except the `counter` API, Pulsar also exposes a general key/value API for functions to store
+general key/value state.
+
+#### put_state
+
+```python
+
+ def put_state(self, key, value):
+ """update the value of a given key in the managed state"""
+
+```
+
+The key is a string, and the value is arbitrary binary data.
+
+#### get_state
+
+```python
+
+ def get_state(self, key):
+ """get the value of a given key in the managed state"""
+
+```
+
+#### del_counter
+
+```python
+
+ def del_counter(self, key):
+ """delete the counter of a given key in the managed state"""
+
+```
+
+Counters and binary values share the same keyspace, so this deletes either type.
+
+</TabItem>
+
+</Tabs>
+
+### Query State
+
+A Pulsar Function can use the [State API](#api) for storing state into Pulsar's state storage
+and retrieving state back from Pulsar's state storage. Additionally Pulsar also provides
+CLI commands for querying its state.
+
+```shell
+
+$ bin/pulsar-admin functions querystate \
+ --tenant <tenant> \
+ --namespace <namespace> \
+ --name <function-name> \
+ --state-storage-url <bookkeeper-service-url> \
+ --key <state-key> \
+ [---watch]
+
+```
+
+If `--watch` is specified, the CLI will watch the value of the provided `state-key`.
+
+### Example
+
+<Tabs
+ defaultValue="Java"
+ values={[{"label":"Java","value":"Java"},{"label":"Python","value":"Python"}]}>
+<TabItem value="Java">
+
+{@inject: github:WordCountFunction:/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/WordCountFunction.java} is a very good example
+demonstrating on how Application can easily store `state` in Pulsar Functions.
+
+```java
+
+public class WordCountFunction implements Function<String, Void> {
+ @Override
+ public Void process(String input, Context context) throws Exception {
+ Arrays.asList(input.split("\\.")).forEach(word -> context.incrCounter(word, 1));
+ return null;
+ }
+}
+
+```
+
+The logic of this `WordCount` function is pretty simple and straightforward:
+
+1. The function first splits the received `String` into multiple words using regex `\\.`.
+2. For each `word`, the function increments the corresponding `counter` by 1 (via `incrCounter(key, amount)`).
+
+</TabItem>
+<TabItem value="Python">
+
+```python
+
+from pulsar import Function
+
+class WordCount(Function):
+ def process(self, item, context):
+ for word in item.split():
+ context.incr_counter(word, 1)
+
+```
+
+The logic of this `WordCount` function is pretty simple and straightforward:
+
+1. The function first splits the received string into multiple words on space.
+2. For each `word`, the function increments the corresponding `counter` by 1 (via `incr_counter(key, amount)`).
+
+</TabItem>
+
+</Tabs>
diff --git a/site2/website-next/versioned_docs/version-2.4.1/functions-overview.md b/site2/website-next/versioned_docs/version-2.4.1/functions-overview.md
new file mode 100644
index 0000000..f222cbe
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/functions-overview.md
@@ -0,0 +1,205 @@
+---
+id: functions-overview
+title: Pulsar Functions overview
+sidebar_label: "Overview"
+original_id: functions-overview
+---
+
+**Pulsar Functions** are lightweight compute processes that
+
+* consume messages from one or more Pulsar topics,
+* apply a user-supplied processing logic to each message,
+* publish the results of the computation to another topic.
+
+
+## Goals
+With Pulsar Functions, you can create complex processing logic without deploying a separate neighboring system (such as [Apache Storm](http://storm.apache.org/), [Apache Heron](https://apache.github.io/incubator-heron), [Apache Flink](https://flink.apache.org/)). Pulsar Functions are computing infrastructure of Pulsar messaging system. The core goal is tied to a series of other goals:
+
+* Developer productivity (language-native vs Pulsar Functions SDK functions)
+* Easy troubleshooting
+* Operational simplicity (no need for an external processing system)
+
+## Inspirations
+Pulsar Functions are inspired by (and take cues from) several systems and paradigms:
+
+* Stream processing engines such as [Apache Storm](http://storm.apache.org/), [Apache Heron](https://apache.github.io/incubator-heron), and [Apache Flink](https://flink.apache.org)
+* "Serverless" and "Function as a Service" (FaaS) cloud platforms like [Amazon Web Services Lambda](https://aws.amazon.com/lambda/), [Google Cloud Functions](https://cloud.google.com/functions/), and [Azure Cloud Functions](https://azure.microsoft.com/en-us/services/functions/)
+
+Pulsar Functions can be described as
+
+* [Lambda](https://aws.amazon.com/lambda/)-style functions that are
+* specifically designed to use Pulsar as a message bus.
+
+## Programming model
+Pulsar Functions provide a wide range of functionality, and the core programming model is simple. Functions receive messages from one or more **input [topics](reference-terminology.md#topic)**. Each time a message is received, the function will complete the following tasks.
+
+ * Apply some processing logic to the input and write output to:
+ * An **output topic** in Pulsar
+ * [Apache BookKeeper](functions-develop.md#state-storage)
+ * Write logs to a **log topic** (potentially for debugging purposes)
+ * Increment a [counter](#word-count-example)
+
+
+
+You can use Pulsar Functions to set up the following processing chain:
+
+* A Python function listens for the `raw-sentences` topic and "sanitizes" incoming strings (removing extraneous whitespace and converting all characters to lowercase) and then publishes the results to a `sanitized-sentences` topic.
+* A Java function listens for the `sanitized-sentences` topic, counts the number of times each word appears within a specified time window, and publishes the results to a `results` topic
+* Finally, a Python function listens for the `results` topic and writes the results to a MySQL table.
+
+
+### Word count example
+
+If you implement the classic word count example using Pulsar Functions, it looks something like this:
+
+
+
+To write the function in Java with [Pulsar Functions SDK for Java](functions-develop.md#available-apis), you can write the function as follows.
+
+```java
+
+package org.example.functions;
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+
+import java.util.Arrays;
+
+public class WordCountFunction implements Function<String, Void> {
+ // This function is invoked every time a message is published to the input topic
+ @Override
+ public Void process(String input, Context context) throws Exception {
+ Arrays.asList(input.split(" ")).forEach(word -> {
+ String counterKey = word.toLowerCase();
+ context.incrCounter(counterKey, 1);
+ });
+ return null;
+ }
+}
+
+```
+
+Bundle and build the JAR file to be deployed, and then deploy it in your Pulsar cluster using the [command line](functions-deploy.md#command-line-interface) as follows.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --jar target/my-jar-with-dependencies.jar \
+ --classname org.example.functions.WordCountFunction \
+ --tenant public \
+ --namespace default \
+ --name word-count \
+ --inputs persistent://public/default/sentences \
+ --output persistent://public/default/count
+
+```
+
+### Content-based routing example
+
+Pulsar Functions are used in many cases. The following is a sophisticated example that involves content-based routing.
+
+For example, a function takes items (strings) as input and publishes them to either a `fruits` or `vegetables` topic, depending on the item. Or, if an item is neither fruit nor vegetable, a warning is logged to a [log topic](functions-develop.md#logger). The following is a visual representation.
+
+
+
+If you implement this routing functionality in Python, it looks something like this:
+
+```python
+
+from pulsar import Function
+
+class RoutingFunction(Function):
+ def __init__(self):
+ self.fruits_topic = "persistent://public/default/fruits"
+ self.vegetables_topic = "persistent://public/default/vegetables"
+
+ @staticmethod
+ def is_fruit(item):
+ return item in ["apple", "orange", "pear", "other fruits..."]
+
+ @staticmethod
+ def is_vegetable(item):
+ return item in ["carrot", "lettuce", "radish", "other vegetables..."]
+
+ def process(self, item, context):
+ if self.is_fruit(item):
+ context.publish(self.fruits_topic, item)
+ elif self.is_vegetable(item):
+ context.publish(self.vegetables_topic, item)
+ else:
+ warning = "The item {0} is neither a fruit nor a vegetable".format(item)
+ context.get_logger().warn(warning)
+
+```
+
+### Functions, messages and message types
+Pulsar Functions take byte arrays as inputs and spit out byte arrays as output. However in languages that support typed interfaces(Java), you can write typed Functions, and bind messages to types in the following ways.
+* [Schema Registry](functions-develop.md#schema-registry)
+* [SerDe](functions-develop.md#serde)
+
+
+## Fully Qualified Function Name (FQFN)
+Each Pulsar Function has a **Fully Qualified Function Name** (FQFN) that consists of three elements: the function tenant, namespace, and function name. FQFN looks like this:
+
+```http
+
+tenant/namespace/name
+
+```
+
+FQFNs enable you to create multiple functions with the same name provided that they are in different namespaces.
+
+## Supported languages
+Currently, you can write Pulsar Functions in Java, Python, and Go. For details, refer to [Develop Pulsar Functions](functions-develop).
+
+## Processing guarantees
+Pulsar Functions provide three different messaging semantics that you can apply to any function.
+
+Delivery semantics | Description
+:------------------|:-------
+**At-most-once** delivery | Each message sent to the function is likely to be processed, or not to be processed (hence "at most").
+**At-least-once** delivery | Each message sent to the function can be processed more than once (hence the "at least").
+**Effectively-once** delivery | Each message sent to the function will have one output associated with it.
+
+
+### Apply processing guarantees to a function
+You can set the processing guarantees for a Pulsar Function when you create the Function. The [`pulsar-function create`](reference-pulsar-admin.md#create-1) command applies effectively-once guarantees to the Function.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --processing-guarantees EFFECTIVELY_ONCE \
+ # Other function configs
+
+```
+
+The available options are:
+
+* `ATMOST_ONCE`
+* `ATLEAST_ONCE`
+* `EFFECTIVELY_ONCE`
+
+The following command runs a function in the cluster mode with effectively-once guarantees applied.
+
+```bash
+
+$ bin/pulsar-admin functions create \
+ --name my-effectively-once-function \
+ --processing-guarantees EFFECTIVELY_ONCE \
+ # Other function configs
+
+```
+
+> By default, Pulsar Functions provide at-least-once delivery guarantees. So if you create a function without supplying a value for the `--processingGuarantees` flag, the function provides at-least-once guarantees.
+
+### Update the processing guarantees of a function
+You can change the processing guarantees applied to a function using the [`update`](reference-pulsar-admin.md#update-1) command. The following is an example.
+
+```bash
+
+$ bin/pulsar-admin functions update \
+ --processing-guarantees ATMOST_ONCE \
+ # Other function configs
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/functions-runtime.md b/site2/website-next/versioned_docs/version-2.4.1/functions-runtime.md
new file mode 100644
index 0000000..37213a6
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/functions-runtime.md
@@ -0,0 +1,157 @@
+---
+id: functions-runtime
+title: Configure Functions runtime
+sidebar_label: "Setup: Configure Functions runtime"
+original_id: functions-runtime
+---
+
+Pulsar Functions support the following methods to run functions.
+
+- *Thread*: Invoke functions in threads in Functions Worker.
+- *Process*: Invoke functions in processes forked by Functions Worker.
+- *Kubernetes*: Submit functions as Kubernetes StatefulSets by Functions Worker.
+
+#### Note
+> Pulsar supports adding labels to the Kubernetes StatefulSets and services while launching functions, which facilitates selecting the target Kubernetes objects.
+
+The differences of the thread and process modes are:
+- Thread mode: when a function runs in thread mode, it runs on the same Java virtual machine (JVM) with Functions worker.
+- Process mode: when a function runs in process mode, it runs on the same machine that Functions worker runs.
+
+## Configure thread runtime
+It is easy to configure *Thread* runtime. In most cases, you do not need to configure anything. You can customize the thread group name with the following settings:
+
+```yaml
+
+threadContainerFactory:
+ threadGroupName: "Your Function Container Group"
+
+```
+
+*Thread* runtime is only supported in Java function.
+
+## Configure process runtime
+When you enable *Process* runtime, you do not need to configure anything.
+
+```yaml
+
+processContainerFactory:
+ # the directory for storing the function logs
+ logDirectory:
+ # change the jar location only when you put the java instance jar in a different location
+ javaInstanceJarLocation:
+ # change the python instance location only when you put the python instance jar in a different location
+ pythonInstanceLocation:
+ # change the extra dependencies location:
+ extraFunctionDependenciesDir:
+
+```
+
+*Process* runtime is supported in Java, Python, and Go functions.
+
+## Configure Kubernetes runtime
+
+It is easy to configure Kubernetes runtime. You can just uncomment the settings of `kubernetesContainerFactory` in the `functions_worker.yaml` file. The following is an example.
+
+```yaml
+
+kubernetesContainerFactory:
+ # uri to kubernetes cluster, leave it to empty and it will use the kubernetes settings in function worker
+ k8Uri:
+ # the kubernetes namespace to run the function instances. it is `default`, if this setting is left to be empty
+ jobNamespace:
+ # the docker image to run function instance. by default it is `apachepulsar/pulsar`
+ pulsarDockerImageName:
+ # the root directory of pulsar home directory in `pulsarDockerImageName`. by default it is `/pulsar`.
+ # if you are using your own built image in `pulsarDockerImageName`, you need to set this setting accordingly
+ pulsarRootDir:
+ # this setting only takes effects if `k8Uri` is set to null. if your function worker is running as a k8 pod,
+ # setting this to true is let function worker to submit functions to the same k8s cluster as function worker
+ # is running. setting this to false if your function worker is not running as a k8 pod.
+ submittingInsidePod: false
+ # setting the pulsar service url that pulsar function should use to connect to pulsar
+ # if it is not set, it will use the pulsar service url configured in worker service
+ pulsarServiceUrl:
+ # setting the pulsar admin url that pulsar function should use to connect to pulsar
+ # if it is not set, it will use the pulsar admin url configured in worker service
+ pulsarAdminUrl:
+ # the custom labels that function worker uses to select the nodes for pods
+ customLabels:
+ # the directory for dropping extra function dependencies
+ # if it is not an absolute path, it is relative to `pulsarRootDir`
+ extraFunctionDependenciesDir:
+ # Additional memory padding added on top of the memory requested by the function per on a per instance basis
+ percentMemoryPadding: 10
+
+```
+
+If you have already run a Pulsar cluster on Kubernetes, you can keep the settings unchanged at most of time.
+
+However, if you enable RBAC on deploying your Pulsar cluster, make sure the service account you use for
+running Functions Workers (or brokers, if Functions Workers run along with brokers) have permissions on the following
+kubernetes APIs.
+
+- services
+- configmaps
+- pods
+- apps.statefulsets
+
+Otherwise, you will not be able to create any functions. The following is an example of error message.
+
+```bash
+
+22:04:27.696 [Timer-0] ERROR org.apache.pulsar.functions.runtime.KubernetesRuntimeFactory - Error while trying to fetch configmap example-pulsar-4qvmb5gur3c6fc9dih0x1xn8b-function-worker-config at namespace pulsar
+io.kubernetes.client.ApiException: Forbidden
+ at io.kubernetes.client.ApiClient.handleResponse(ApiClient.java:882) ~[io.kubernetes-client-java-2.0.0.jar:?]
+ at io.kubernetes.client.ApiClient.execute(ApiClient.java:798) ~[io.kubernetes-client-java-2.0.0.jar:?]
+ at io.kubernetes.client.apis.CoreV1Api.readNamespacedConfigMapWithHttpInfo(CoreV1Api.java:23673) ~[io.kubernetes-client-java-api-2.0.0.jar:?]
+ at io.kubernetes.client.apis.CoreV1Api.readNamespacedConfigMap(CoreV1Api.java:23655) ~[io.kubernetes-client-java-api-2.0.0.jar:?]
+ at org.apache.pulsar.functions.runtime.KubernetesRuntimeFactory.fetchConfigMap(KubernetesRuntimeFactory.java:284) [org.apache.pulsar-pulsar-functions-runtime-2.4.0-42c3bf949.jar:2.4.0-42c3bf949]
+ at org.apache.pulsar.functions.runtime.KubernetesRuntimeFactory$1.run(KubernetesRuntimeFactory.java:275) [org.apache.pulsar-pulsar-functions-runtime-2.4.0-42c3bf949.jar:2.4.0-42c3bf949]
+ at java.util.TimerThread.mainLoop(Timer.java:555) [?:1.8.0_212]
+ at java.util.TimerThread.run(Timer.java:505) [?:1.8.0_212]
+
+```
+
+If this happens, you need to grant the required permissions to the service account used for running Functions Workers. An example to grant permissions is shown below: a service account `functions-worker` is granted with permissions to access Kubernetes resources `services`, `configmaps`, `pods` and `apps.statefulsets`.
+
+```yaml
+
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: ClusterRole
+metadata:
+ name: functions-worker
+rules:
+- apiGroups: [""]
+ resources:
+ - services
+ - configmaps
+ - pods
+ verbs:
+ - '*'
+- apiGroups:
+ - apps
+ resources:
+ - statefulsets
+ verbs:
+ - '*'
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+ name: functions-worker
+---
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: ClusterRoleBinding
+metadata:
+ name: functions-worker
+roleRef:
+ apiGroup: rbac.authorization.k8s.io
+ kind: ClusterRole
+ name: functions-worker
+subjects:
+- kind: ServiceAccount
+ name: functions-worker
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/functions-worker.md b/site2/website-next/versioned_docs/version-2.4.1/functions-worker.md
new file mode 100644
index 0000000..643d9ca
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/functions-worker.md
@@ -0,0 +1,280 @@
+---
+id: functions-worker
+title: Deploy and manage functions worker
+sidebar_label: "Setup: Pulsar Functions Worker"
+original_id: functions-worker
+---
+
+Before using Pulsar Functions, you need to learn how to set up Pulsar Functions worker and how to [configure Functions runtime](functions-runtime).
+
+Pulsar `functions-worker` is a logic component to run Pulsar Functions in cluster mode. Two options are available, and you can select either of the two options based on your requirements.
+- [run with brokers](#run-functions-worker-with-brokers)
+- [run it separately](#run-functions-worker-separately) in a different broker
+
+:::note
+
+The `--- Service Urls---` lines in the following diagrams represent Pulsar service URLs that Pulsar client and admin use to connect to a Pulsar cluster.
+
+:::
+
+## Run Functions-worker with brokers
+
+The following diagram illustrates the deployment of functions-workers running along with brokers.
+
+
+
+To enable functions-worker running as part of a broker, you need to set `functionsWorkerEnabled` to `true` in the `broker.conf` file.
+
+```conf
+
+functionsWorkerEnabled=true
+
+```
+
+When you set `functionsWorkerEnabled` to `true`, it means that you start functions-worker as part of a broker. You need to configure the `conf/functions_worker.yml` file to customize your functions_worker.
+
+Before you run Functions-work with broker, you have to configure Functions-worker, and then start it with brokers.
+
+### Configure Functions-Worker to run with brokers
+In this mode, since `functions-worker` is running as part of broker, most of the settings already inherit from your broker configuration (for example, configurationStore settings, authentication settings, and so on).
+
+Pay attention to the following required settings when configuring functions-worker in this mode.
+
+- `numFunctionPackageReplicas`: The number of replicas to store function packages. The default value is `1`, which is good for standalone deployment. For production deployment, to ensure high availability, set it to be more than `2` .
+- `pulsarFunctionsCluster`: Set the value to your Pulsar cluster name (same as the `clusterName` setting in the broker configuration).
+
+If authentication is enabled on the BookKeeper cluster, configure the following BookKeeper authentication settings.
+
+- `bookkeeperClientAuthenticationPlugin`: the BookKeeper client authentication plugin name.
+- `bookkeeperClientAuthenticationParametersName`: the BookKeeper client authentication plugin parameters name.
+- `bookkeeperClientAuthenticationParameters`: the BookKeeper client authentication plugin parameters.
+
+### Start Functions-worker with broker
+
+Once you have configured the `functions_worker.yml` file, you can start or restart your broker.
+
+And then you can use the following command to verify if `functions-worker` is running well.
+
+```bash
+
+curl <broker-ip>:8080/admin/v2/worker/cluster
+
+```
+
+After entering the command above, a list of active function workers in the cluster is returned. The output is something similar as follows.
+
+```json
+
+[{"workerId":"<worker-id>","workerHostname":"<worker-hostname>","port":8080}]
+
+```
+
+## Run Functions-worker separately
+
+This section illustrates how to run `functions-worker` as a separate process in separate machines.
+
+
+
+> Note
+In this mode, make sure `functionsWorkerEnabled` is set to `false`, so you won't start `functions-worker` with brokers by mistake.
+
+### Configure Functions-worker to run separately
+
+To run function-worker separately, you have to configure the following parameters.
+
+#### Worker parameters
+
+- `workerId`: The type is string. It is unique across clusters, used to identify a worker machine.
+- `workerHostname`: The hostname of the worker machine.
+- `workerPort`: The port that the worker server listens on. Keep it as default if you don't customize it.
+- `workerPortTls`: The TLS port that the worker server listens on. Keep it as default if you don't customize it.
+
+#### Function package parameter
+
+- `numFunctionPackageReplicas`: The number of replicas to store function packages. The default value is `1`.
+
+#### Function metadata parameter
+
+- `pulsarServiceUrl`: The Pulsar service URL for your broker cluster.
+- `pulsarWebServiceUrl`: The Pulsar web service URL for your broker cluster.
+- `pulsarFunctionsCluster`: Set the value to your Pulsar cluster name (same as the `clusterName` setting in the broker configuration).
+
+If authentication is enabled for your broker cluster, you *should* configure the authentication plugin and parameters for the functions worker to communicate with the brokers.
+
+- `clientAuthenticationPlugin`
+- `clientAuthenticationParameters`
+
+#### Security settings
+
+If you want to enable security on functions workers, you *should*:
+- [Enable TLS transport encryption](#enable-tls-transport-encryption)
+- [Enable Authentication Provider](#enable-authentication-provider)
+- [Enable Authorization Provider](#enable-authorization-provider)
+
+##### Enable TLS transport encryption
+
+To enable TLS transport encryption, configure the following settings.
+
+```
+
+tlsEnabled: true
+tlsCertificateFilePath: /path/to/functions-worker.cert.pem
+tlsKeyFilePath: /path/to/functions-worker.key-pk8.pem
+tlsTrustCertsFilePath: /path/to/ca.cert.pem
+
+```
+
+For details on TLS encryption, refer to [Transport Encryption using TLS](security-tls-transport).
+
+##### Enable Authentication Provider
+
+To enable authentication on Functions Worker, configure the following settings.
+> Note
+Substitute the *providers list* with the providers you want to enable.
+
+```
+
+authenticationEnabled: true
+authenticationProviders: [ provider1, provider2 ]
+
+```
+
+For *SASL Authentication* provider, add `saslJaasClientAllowedIds` and `saslJaasBrokerSectionName`
+under `properties` if needed.
+
+```
+
+properties:
+ saslJaasClientAllowedIds: .*pulsar.*
+ saslJaasBrokerSectionName: Broker
+
+```
+
+For *Token Authentication* prodivder, add necessary settings under `properties` if needed.
+See [Token Authentication](security-token-admin) for more details.
+
+```
+
+properties:
+ tokenSecretKey: file://my/secret.key
+ # If using public/private
+ # tokenPublicKey: file:///path/to/public.key
+
+```
+
+##### Enable Authorization Provider
+
+To enable authorization on Functions Worker, you need to configure `authorizationEnabled` and `configurationStoreServers`. The authentication provider connects to `configurationStoreServers` to receive namespace policies.
+
+```yaml
+
+authorizationEnabled: true
+configurationStoreServers: <configuration-store-servers>
+
+```
+
+You should also configure a list of superuser roles. The superuser roles are able to access any admin API. The following is a configuration example.
+
+```yaml
+
+superUserRoles:
+ - role1
+ - role2
+ - role3
+
+```
+
+#### BookKeeper Authentication
+
+If authentication is enabled on the BookKeeper cluster, you should configure the BookKeeper authentication settings as follows:
+
+- `bookkeeperClientAuthenticationPlugin`: the plugin name of BookKeeper client authentication.
+- `bookkeeperClientAuthenticationParametersName`: the plugin parameters name of BookKeeper client authentication.
+- `bookkeeperClientAuthenticationParameters`: the plugin parameters of BookKeeper client authentication.
+
+### Start Functions-worker
+
+Once you have finished configuring the `functions_worker.yml` configuration file, you can use the following command to start a `functions-worker`:
+
+```bash
+
+bin/pulsar functions-worker
+
+```
+
+### Configure Proxies for Functions-workers
+
+When you are running `functions-worker` in a separate cluster, the admin rest endpoints are split into two clusters. `functions`, `function-worker`, `source` and `sink` endpoints are now served
+by the `functions-worker` cluster, while all the other remaining endpoints are served by the broker cluster.
+Hence you need to configure your `pulsar-admin` to use the right service URL accordingly.
+
+In order to address this inconvenience, you can start a proxy cluster for routing the admin rest requests accordingly. Hence you will have one central entry point for your admin service.
+
+If you already have a proxy cluster, continue reading. If you haven't setup a proxy cluster before, you can follow the [instructions](http://pulsar.apache.org/docs/en/administration-proxy/) to
+start proxies.
+
+
+
+To enable routing functions related admin requests to `functions-worker` in a proxy, you can edit the `proxy.conf` file to modify the following settings:
+
+```conf
+
+functionWorkerWebServiceURL=<pulsar-functions-worker-web-service-url>
+functionWorkerWebServiceURLTLS=<pulsar-functions-worker-web-service-url>
+
+```
+
+## Compare the Run-with-Broker and Run-separately modes
+
+As described above, you can run Function-worker with brokers, or run it separately. And it is more convenient to run functions-workers along with brokers. However, running functions-workers in a separate cluster provides better resource isolation for running functions in `Process` or `Thread` mode.
+
+Use which mode for your cases, refer to the following guidelines to determine.
+
+Use the `Run-with-Broker` mode in the following cases:
+- a) if resource isolation is not required when running functions in `Process` or `Thread` mode;
+- b) if you configure the functions-worker to run functions on Kubernetes (where the resource isolation problem is addressed by Kubernetes).
+
+Use the `Run-separately` mode in the following cases:
+- a) you don't have a Kubernetes cluster;
+- b) if you want to run functions and brokers separately.
+
+## Troubleshooting
+
+**Error message: Namespace missing local cluster name in clusters list**
+
+```
+
+Failed to get partitioned topic metadata: org.apache.pulsar.client.api.PulsarClientException$BrokerMetadataException: Namespace missing local cluster name in clusters list: local_cluster=xyz ns=public/functions clusters=[standalone]
+
+```
+
+The error message prompts when either of the cases occurs:
+- a) a broker is started with `functionsWorkerEnabled=true`, but the `pulsarFunctionsCluster` is not set to the correct cluster in the `conf/functions_worker.yaml` file;
+- b) setting up a geo-replicated Pulsar cluster with `functionsWorkerEnabled=true`, while brokers in one cluster run well, brokers in the other cluster do not work well.
+
+**Workaround**
+
+If any of these cases happens, follow the instructions below to fix the problem:
+
+1. Disable Functions Worker by setting `functionsWorkerEnabled=false`, and restart brokers.
+
+2. Get the current clusters list of `public/functions` namespace.
+
+```bash
+
+bin/pulsar-admin namespaces get-clusters public/functions
+
+```
+
+3. Check if the cluster is in the clusters list. If the cluster is not in the list, add it to the list and update the clusters list.
+
+```bash
+
+bin/pulsar-admin namespaces set-clusters --clusters <existing-clusters>,<new-cluster> public/functions
+
+```
+
+4. After setting the cluster successfully, enable functions worker by setting `functionsWorkerEnabled=true`.
+
+5. Set the correct cluster name in `pulsarFunctionsCluster` in the `conf/functions_worker.yml` file, and restart brokers.
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-cdc.md b/site2/website-next/versioned_docs/version-2.4.1/io-cdc.md
new file mode 100644
index 0000000..20f16ae
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-cdc.md
@@ -0,0 +1,26 @@
+---
+id: io-cdc
+title: CDC connector
+sidebar_label: "CDC connector"
+original_id: io-cdc
+---
+
+CDC source connectors capture log changes of databases (such as MySQL, MongoDB, and PostgreSQL) into Pulsar.
+
+> CDC source connectors are built on top of [Canal](https://github.com/alibaba/canal) and [Debezium](https://debezium.io/) and store all data into Pulsar cluster in a persistent, replicated, and partitioned way.
+
+Currently, Pulsar has the following CDC connectors.
+
+Name|Java Class
+|---|---
+[Canal source connector](io-canal-source)|[org.apache.pulsar.io.canal.CanalStringSource.java](https://github.com/apache/pulsar/blob/master/pulsar-io/canal/src/main/java/org/apache/pulsar/io/canal/CanalStringSource.java)
+[Debezium source connector](io-cdc-debezium)|<li>[org.apache.pulsar.io.debezium.DebeziumSource.java](https://github.com/apache/pulsar/blob/master/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java)<br /></li><li>[org.apache.pulsar.io.debezium.mysql.DebeziumMysqlSource.java](https://github.com/apache/pulsar/blob/master/pulsar-io/debezium/mysql/src/main/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSource.java)<br /></li><li>[org.apache.pulsar.i [...]
+
+For more information about Canal and Debezium, see the information below.
+
+Subject | Reference
+|---|---
+How to use Canal source connector with MySQL|[Canal guide](https://github.com/alibaba/canal/wiki)
+How does Canal work | [Canal tutorial](https://github.com/alibaba/canal/wiki)
+How to use Debezium source connector with MySQL | [Debezium guide](https://debezium.io/docs/connectors/mysql/)
+How does Debezium work | [Debezium tutorial](https://debezium.io/docs/tutorial/)
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-connectors.md b/site2/website-next/versioned_docs/version-2.4.1/io-connectors.md
new file mode 100644
index 0000000..f434f87
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-connectors.md
@@ -0,0 +1,31 @@
+---
+id: io-connectors
+title: Builtin Connectors
+sidebar_label: "Builtin Connectors"
+original_id: io-connectors
+---
+
+Pulsar distribution includes a set of common connectors that have been packaged and tested with the rest of Apache Pulsar.
+These connectors import and export data from some of the most commonly used data systems. Using any these connectors is
+as easy as writing a simple connector configuration and running the connector locally or submitting the connector to a
+Pulsar Functions cluster.
+
+- [Aerospike Sink Connector](io-aerospike)
+- [Cassandra Sink Connector](io-cassandra)
+- [Kafka Sink Connector](io-kafka.md#sink)
+- [Kafka Source Connector](io-kafka.md#source)
+- [Kinesis Sink Connector](io-kinesis.md#sink)
+- [RabbitMQ Source Connector](io-rabbitmq.md#source)
+- [RabbitMQ Sink Connector](io-rabbitmq.md#sink)
+- [Twitter Firehose Source Connector](io-twitter)
+- [CDC Source Connector based on Debezium](io-cdc)
+- [Netty Source Connector](io-netty.md#source)
+- [Hbase Sink Connector](io-hbase.md#sink)
+- [ElasticSearch Sink Connector](io-elasticsearch.md#sink)
+- [File Source Connector](io-file.md#source)
+- [Hdfs Sink Connector](io-hdfs.md#sink)
+- [MongoDB Sink Connector](io-mongo.md#sink)
+- [Redis Sink Connector](io-redis.md#sink)
+- [Solr Sink Connector](io-solr.md#sink)
+- [InfluxDB Sink Connector](io-influxdb.md#sink)
+- [JDBC Sink Connector](io-jdbc)
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-debug.md b/site2/website-next/versioned_docs/version-2.4.1/io-debug.md
new file mode 100644
index 0000000..059f08b
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-debug.md
@@ -0,0 +1,407 @@
+---
+id: io-debug
+title: How to debug Pulsar connectors
+sidebar_label: "Debug"
+original_id: io-debug
+---
+This guide explains how to debug connectors in localrun or cluster mode and gives a debugging checklist.
+To better demonstrate how to debug Pulsar connectors, here takes a Mongo sink connector as an example.
+
+**Deploy a Mongo sink environment**
+1. Start a Mongo service.
+
+ ```bash
+
+ docker pull mongo:4
+ docker run -d -p 27017:27017 --name pulsar-mongo -v $PWD/data:/data/db mongo:4
+
+ ```
+
+2. Create a DB and a collection.
+
+ ```bash
+
+ docker exec -it pulsar-mongo /bin/bash
+ mongo
+ > use pulsar
+ > db.createCollection('messages')
+ > exit
+
+ ```
+
+3. Start Pulsar standalone.
+
+ ```bash
+
+ docker pull apachepulsar/pulsar:2.4.0
+ docker run -d -it -p 6650:6650 -p 8080:8080 -v $PWD/data:/pulsar/data --link pulsar-mongo --name pulsar-mongo-standalone apachepulsar/pulsar:2.4.0 bin/pulsar standalone
+
+ ```
+
+4. Configure the Mongo sink with the `mongo-sink-config.yaml` file.
+
+ ```bash
+
+ configs:
+ mongoUri: "mongodb://pulsar-mongo:27017"
+ database: "pulsar"
+ collection: "messages"
+ batchSize: 2
+ batchTimeMs: 500
+
+ ```
+
+ ```bash
+
+ docker cp mongo-sink-config.yaml pulsar-mongo-standalone:/pulsar/
+
+ ```
+
+5. Download the Mongo sink nar package.
+
+ ```bash
+
+ docker exec -it pulsar-mongo-standalone /bin/bash
+ curl -O http://apache.01link.hk/pulsar/pulsar-2.4.0/connectors/pulsar-io-mongo-2.4.0.nar
+
+ ```
+
+## Debug in localrun mode
+Start the Mongo sink in localrun mode using the `localrun` command.
+:::tip
+
+For more information about the `localrun` command, see [`localrun`](reference-connector-admin.md/#localrun-1).
+
+:::
+
+```bash
+
+./bin/pulsar-admin sinks localrun \
+--archive pulsar-io-mongo-2.4.0.nar \
+--tenant public --namespace default \
+--inputs test-mongo \
+--name pulsar-mongo-sink \
+--sink-config-file mongo-sink-config.yaml \
+--parallelism 1
+
+```
+
+### Use connector log
+Use one of the following methods to get a connector log in localrun mode:
+* After executing the `localrun` command, the **log is automatically printed on the console**.
+* The log is located at:
+
+ ```bash
+
+ logs/functions/tenant/namespace/function-name/function-name-instance-id.log
+
+ ```
+
+ **Example**
+
+ The path of the Mongo sink connector is:
+
+ ```bash
+
+ logs/functions/public/default/pulsar-mongo-sink/pulsar-mongo-sink-0.log
+
+ ```
+
+To clearly explain the log information, here breaks down the large block of information into small blocks and add descriptions for each block.
+* This piece of log information shows the storage path of the nar package after decompression.
+
+ ```
+
+ 08:21:54.132 [main] INFO org.apache.pulsar.common.nar.NarClassLoader - Created class loader with paths: [file:/tmp/pulsar-nar/pulsar-io-mongo-2.4.0.nar-unpacked/, file:/tmp/pulsar-nar/pulsar-io-mongo-2.4.0.nar-unpacked/META-INF/bundled-dependencies/,
+
+ ```
+
+ :::tip
+
+ If `class cannot be found` exception is thrown, check whether the nar file is decompressed in the folder `file:/tmp/pulsar-nar/pulsar-io-mongo-2.4.0.nar-unpacked/META-INF/bundled-dependencies/` or not.
+
+ :::
+
+* This piece of log information illustrates the basic information about the Mongo sink connector, such as tenant, namespace, name, parallelism, resources, and so on, which can be used to **check whether the Mongo sink connector is configured correctly or not**.
+
+ ```bash
+
+ 08:21:55.390 [main] INFO org.apache.pulsar.functions.runtime.ThreadRuntime - ThreadContainer starting function with instance config InstanceConfig(instanceId=0, functionId=853d60a1-0c48-44d5-9a5c-6917386476b2, functionVersion=c2ce1458-b69e-4175-88c0-a0a856a2be8c, functionDetails=tenant: "public"
+ namespace: "default"
+ name: "pulsar-mongo-sink"
+ className: "org.apache.pulsar.functions.api.utils.IdentityFunction"
+ autoAck: true
+ parallelism: 1
+ source {
+ typeClassName: "[B"
+ inputSpecs {
+ key: "test-mongo"
+ value {
+ }
+ }
+ cleanupSubscription: true
+ }
+ sink {
+ className: "org.apache.pulsar.io.mongodb.MongoSink"
+ configs: "{\"mongoUri\":\"mongodb://pulsar-mongo:27017\",\"database\":\"pulsar\",\"collection\":\"messages\",\"batchSize\":2,\"batchTimeMs\":500}"
+ typeClassName: "[B"
+ }
+ resources {
+ cpu: 1.0
+ ram: 1073741824
+ disk: 10737418240
+ }
+ componentType: SINK
+ , maxBufferedTuples=1024, functionAuthenticationSpec=null, port=38459, clusterName=local)
+
+ ```
+
+* This piece of log information demonstrates the status of the connections to Mongo and configuration information.
+
+ ```bash
+
+ 08:21:56.231 [cluster-ClusterId{value='5d6396a3c9e77c0569ff00eb', description='null'}-pulsar-mongo:27017] INFO org.mongodb.driver.connection - Opened connection [connectionId{localValue:1, serverValue:8}] to pulsar-mongo:27017
+ 08:21:56.326 [cluster-ClusterId{value='5d6396a3c9e77c0569ff00eb', description='null'}-pulsar-mongo:27017] INFO org.mongodb.driver.cluster - Monitor thread successfully connected to server with description ServerDescription{address=pulsar-mongo:27017, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[4, 2, 0]}, minWireVersion=0, maxWireVersion=8, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=89058800}
+
+ ```
+
+* This piece of log information explains the configuration of consumers and clients, including the topic name, subscription name, subscription type, and so on.
+
+ ```bash
+
+ 08:21:56.719 [pulsar-client-io-1-1] INFO org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - Starting Pulsar consumer perf with config: {
+ "topicNames" : [ "test-mongo" ],
+ "topicsPattern" : null,
+ "subscriptionName" : "public/default/pulsar-mongo-sink",
+ "subscriptionType" : "Shared",
+ "receiverQueueSize" : 1000,
+ "acknowledgementsGroupTimeMicros" : 100000,
+ "negativeAckRedeliveryDelayMicros" : 60000000,
+ "maxTotalReceiverQueueSizeAcrossPartitions" : 50000,
+ "consumerName" : null,
+ "ackTimeoutMillis" : 0,
+ "tickDurationMillis" : 1000,
+ "priorityLevel" : 0,
+ "cryptoFailureAction" : "CONSUME",
+ "properties" : {
+ "application" : "pulsar-sink",
+ "id" : "public/default/pulsar-mongo-sink",
+ "instance_id" : "0"
+ },
+ "readCompacted" : false,
+ "subscriptionInitialPosition" : "Latest",
+ "patternAutoDiscoveryPeriod" : 1,
+ "regexSubscriptionMode" : "PersistentOnly",
+ "deadLetterPolicy" : null,
+ "autoUpdatePartitions" : true,
+ "replicateSubscriptionState" : false,
+ "resetIncludeHead" : false
+ }
+ 08:21:56.726 [pulsar-client-io-1-1] INFO org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - Pulsar client config: {
+ "serviceUrl" : "pulsar://localhost:6650",
+ "authPluginClassName" : null,
+ "authParams" : null,
+ "operationTimeoutMs" : 30000,
+ "statsIntervalSeconds" : 60,
+ "numIoThreads" : 1,
+ "numListenerThreads" : 1,
+ "connectionsPerBroker" : 1,
+ "useTcpNoDelay" : true,
+ "useTls" : false,
+ "tlsTrustCertsFilePath" : null,
+ "tlsAllowInsecureConnection" : false,
+ "tlsHostnameVerificationEnable" : false,
+ "concurrentLookupRequest" : 5000,
+ "maxLookupRequest" : 50000,
+ "maxNumberOfRejectedRequestPerConnection" : 50,
+ "keepAliveIntervalSeconds" : 30,
+ "connectionTimeoutMs" : 10000,
+ "requestTimeoutMs" : 60000,
+ "defaultBackoffIntervalNanos" : 100000000,
+ "maxBackoffIntervalNanos" : 30000000000
+ }
+
+ ```
+
+## Debug in cluster mode
+You can use the following methods to debug a connector in cluster mode:
+* [Use connector log](#use-connector-log)
+* [Use admin CLI](#use-admin-cli)
+### Use connector log
+In cluster mode, multiple connectors can run on a worker. To find the log path of a specified connector, use the `workerId` to locate the connector log.
+### Use admin CLI
+Pulsar admin CLI helps you debug Pulsar connectors with the following subcommands:
+* [`get`](#get)
+
+* [`status`](#status)
+* [`topics stats`](#topics-stats)
+
+**Create a Mongo sink**
+
+```bash
+
+./bin/pulsar-admin sinks create \
+--archive pulsar-io-mongo-2.4.0.nar \
+--tenant public \
+--namespace default \
+--inputs test-mongo \
+--name pulsar-mongo-sink \
+--sink-config-file mongo-sink-config.yaml \
+--parallelism 1
+
+```
+
+### `get`
+Use the `get` command to get the basic information about the Mongo sink connector, such as tenant, namespace, name, parallelism, and so on.
+
+```bash
+
+./bin/pulsar-admin sinks get --tenant public --namespace default --name pulsar-mongo-sink
+{
+ "tenant": "public",
+ "namespace": "default",
+ "name": "pulsar-mongo-sink",
+ "className": "org.apache.pulsar.io.mongodb.MongoSink",
+ "inputSpecs": {
+ "test-mongo": {
+ "isRegexPattern": false
+ }
+ },
+ "configs": {
+ "mongoUri": "mongodb://pulsar-mongo:27017",
+ "database": "pulsar",
+ "collection": "messages",
+ "batchSize": 2.0,
+ "batchTimeMs": 500.0
+ },
+ "parallelism": 1,
+ "processingGuarantees": "ATLEAST_ONCE",
+ "retainOrdering": false,
+ "autoAck": true
+}
+
+```
+
+:::tip
+
+For more information about the `get` command, see [`get`](reference-connector-admin.md/#get-1).
+
+:::
+
+### `status`
+Use the `status` command to get the current status about the Mongo sink connector, such as the number of instance, the number of running instance, instanceId, workerId and so on.
+
+```bash
+
+./bin/pulsar-admin sinks status
+--tenant public \
+--namespace default \
+--name pulsar-mongo-sink
+{
+"numInstances" : 1,
+"numRunning" : 1,
+"instances" : [ {
+ "instanceId" : 0,
+ "status" : {
+ "running" : true,
+ "error" : "",
+ "numRestarts" : 0,
+ "numReadFromPulsar" : 0,
+ "numSystemExceptions" : 0,
+ "latestSystemExceptions" : [ ],
+ "numSinkExceptions" : 0,
+ "latestSinkExceptions" : [ ],
+ "numWrittenToSink" : 0,
+ "lastReceivedTime" : 0,
+ "workerId" : "c-standalone-fw-5d202832fd18-8080"
+ }
+} ]
+}
+
+```
+
+:::tip
+
+For more information about the `status` command, see [`status`](reference-connector-admin.md/#stauts-1).
+If there are multiple connectors running on a worker, `workerId` can locate the worker on which the specified connector is running.
+
+:::
+
+### `topics stats`
+Use the `topics stats` command to get the stats for a topic and its connected producer and consumer, such as whether the topic has received messages or not, whether there is a backlog of messages or not, the available permits and other key information. All rates are computed over a 1-minute window and are relative to the last completed 1-minute period.
+
+```bash
+
+./bin/pulsar-admin topics stats test-mongo
+{
+ "msgRateIn" : 0.0,
+ "msgThroughputIn" : 0.0,
+ "msgRateOut" : 0.0,
+ "msgThroughputOut" : 0.0,
+ "averageMsgSize" : 0.0,
+ "storageSize" : 1,
+ "publishers" : [ ],
+ "subscriptions" : {
+ "public/default/pulsar-mongo-sink" : {
+ "msgRateOut" : 0.0,
+ "msgThroughputOut" : 0.0,
+ "msgRateRedeliver" : 0.0,
+ "msgBacklog" : 0,
+ "blockedSubscriptionOnUnackedMsgs" : false,
+ "msgDelayed" : 0,
+ "unackedMessages" : 0,
+ "type" : "Shared",
+ "msgRateExpired" : 0.0,
+ "consumers" : [ {
+ "msgRateOut" : 0.0,
+ "msgThroughputOut" : 0.0,
+ "msgRateRedeliver" : 0.0,
+ "consumerName" : "dffdd",
+ "availablePermits" : 999,
+ "unackedMessages" : 0,
+ "blockedConsumerOnUnackedMsgs" : false,
+ "metadata" : {
+ "instance_id" : "0",
+ "application" : "pulsar-sink",
+ "id" : "public/default/pulsar-mongo-sink"
+ },
+ "connectedSince" : "2019-08-26T08:48:07.582Z",
+ "clientVersion" : "2.4.0",
+ "address" : "/172.17.0.3:57790"
+ } ],
+ "isReplicated" : false
+ }
+ },
+ "replication" : { },
+ "deduplicationStatus" : "Disabled"
+}
+
+```
+
+:::tip
+
+For more information about the `topic stats` command, see [`topic stats`](http://pulsar.apache.org/docs/en/pulsar-admin/#stats-1).
+
+:::
+
+## Checklist
+This checklist indicates the major areas to check when you debug connectors. It is a reminder of what to look for to ensure a thorough review and an evaluation tool to get the status of connectors.
+* Does Pulsar start successfully?
+
+* Does the external service run normally?
+
+* Is the nar package complete?
+
+* Is the connector configuration file correct?
+
+* In localrun mode, run a connector and check the printed information (connector log) on the console.
+
+* In cluster mode：
+
+ * Use the `get` command to get the basic information.
+
+ * Use the `status` command to get the current status.
+ * Use the `topics stats` command to get the stats for a specified topic and its connected producers and consumers.
+
+ * Check the connector log.
+* Enter into the external system and verify the result.
\ No newline at end of file
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-develop.md b/site2/website-next/versioned_docs/version-2.4.1/io-develop.md
new file mode 100644
index 0000000..6ff3742
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-develop.md
@@ -0,0 +1,205 @@
+---
+id: io-develop
+title: Develop Connectors
+sidebar_label: "Developing Connectors"
+original_id: io-develop
+---
+
+This guide describes how developers can write new connectors for Pulsar IO to move data
+between Pulsar and other systems. It describes how to create a Pulsar IO connector.
+
+Pulsar IO connectors are specialized [Pulsar Functions](functions-overview). So writing
+a Pulsar IO connector is as simple as writing a Pulsar function. Pulsar IO connectors come
+in two flavors: {@inject: github:Source:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Source.java},
+which import data from another system, and {@inject: github:Sink:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Sink.java},
+which export data to another system. For example, [KinesisSink](io-kinesis) would export
+the messages of a Pulsar topic to a Kinesis stream, and [RabbitmqSource](io-rabbitmq) would import
+the messages of a RabbitMQ queue to a Pulsar topic.
+
+### Developing
+
+#### Develop a source connector
+
+What you need to develop a source connector is to implement {@inject: github:Source:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Source.java}
+interface.
+
+First, you need to implement the {@inject: github:open:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Source.java#L33} method. This method will be called once when the source connector
+is initialized. In this method, you can retrieve all the connector specific settings through
+the passed `config` parameter, and initialize all the necessary resourcess. For example, a Kafka
+connector can create the Kafka client in this `open` method.
+
+Beside the passed-in `config` object, the Pulsar runtime also provides a `SourceContext` for the
+connector to access runtime resources for tasks like collecting metrics. The implementation can
+save the `SourceContext` for further usage.
+
+```java
+
+ /**
+ * Open connector with configuration
+ *
+ * @param config initialization config
+ * @param sourceContext
+ * @throws Exception IO type exceptions when opening a connector
+ */
+ void open(final Map<String, Object> config, SourceContext sourceContext) throws Exception;
+
+```
+
+The main task for a Source implementor is to implement {@inject: github:read:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Source.java#L41}
+method.
+
+```java
+
+ /**
+ * Reads the next message from source.
+ * If source does not have any new messages, this call should block.
+ * @return next message from source. The return result should never be null
+ * @throws Exception
+ */
+ Record<T> read() throws Exception;
+
+```
+
+The implementation should be blocking on this method if nothing to return. It should never return
+`null`. The returned {@inject: github:Record:/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java#L28} should encapsulates the information that is needed by
+Pulsar IO runtime.
+
+These information includes:
+
+- *Topic Name*: _Optional_. If the record is originated from a Pulsar topic, it should be the Pulsar topic name.
+- *Key*: _Optional_. If the record has a key associated with it.
+- *Value*: _Required_. The actual data of this record.
+- *Partition Id*: _Optional_. If the record is originated from a partitioned source,
+ return its partition id. The partition id will be used as part of the unique identifier
+ by Pulsar IO runtime to do message deduplication and achieve exactly-once processing guarantee.
+- *Record Sequence*: _Optional_. If the record is originated from a sequential source,
+ return its record sequence. The record sequence will be used as part of the unique identifier
+ by Pulsar IO runtime to do message deduplication and achieve exactly-once processing guarantee.
+- *Properties*: _Optional_. If the record carries user-defined properties, return those properties.
+
+Additionally, the implementation of the record should provide two methods: `ack` and `fail`. These
+two methods will be used by Pulsar IO connector to acknowledge the records that it has done
+processing and fail the records that it has failed to process.
+
+{@inject: github:KafkaSource:/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSource.java} is a good example to follow.
+
+#### Develop a sink connector
+
+Developing a sink connector is as easy as developing a source connector. You just need to
+implement {@inject: github:Sink:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Sink.java} interface.
+
+Similarly, you first need to implement the {@inject: github:open:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Sink.java#L36} method to initialize all the necessary resources
+before implementing the {@inject: github:write:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Sink.java#L44} method.
+
+```java
+
+ /**
+ * Open connector with configuration
+ *
+ * @param config initialization config
+ * @param sinkContext
+ * @throws Exception IO type exceptions when opening a connector
+ */
+ void open(final Map<String, Object> config, SinkContext sinkContext) throws Exception;
+
+```
+
+The main task for a Sink implementor is to implement {@inject: github:write:/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/Sink.java#L44} method.
+
+```java
+
+ /**
+ * Write a message to Sink
+ * @param record record to write to sink
+ * @throws Exception
+ */
+ void write(Record<T> record) throws Exception;
+
+```
+
+In the implementation of `write` method, the implementor can decide how to write the value and
+the optional key to the actual source, and leverage all the provided information such as
+`Partition Id`, `Record Sequence` for achieving different processing guarantees. The implementor
+is also responsible for acknowledging records if it has successfully written them or failing
+records if has failed to write them.
+
+### Testing
+
+Testing connectors can be challenging because Pulsar IO connectors interact with two systems
+that may be difficult to mock - Pulsar and the system the connector is connecting to. It is
+recommended to write very specifically test the functionalities of the connector classes
+while mocking the external services.
+
+Once you have written sufficient unit tests for your connector, we also recommend adding
+separate integration tests to verify end-to-end functionality. In Pulsar, we are using [testcontainers](https://www.testcontainers.org/) for all Pulsar integration tests. Pulsar IO
+{@inject: github:IntegrationTests:/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io} are good examples to follow on integration testing your connectors.
+
+### Packaging
+
+Once you've developed and tested your connector, you must package it so that it can be submitted
+to a [Pulsar Functions](functions-overview) cluster. There are two approaches described
+here work with Pulsar Functions' runtime.
+
+If you plan to package and distribute your connector for others to use, you are obligated to
+properly license and copyright your own code and to adhere to the licensing and copyrights of
+all libraries your code uses and that you include in your distribution. If you are using the
+approach described in ["Creating a NAR package"](#creating-a-nar-package), the NAR plugin will
+automatically create a `DEPENDENCIES` file in the generated NAR package, including the proper
+licensing and copyrights of all libraries of your connector.
+
+#### Creating a NAR package
+
+The easiest approach to packaging a Pulsar IO connector is to create a NAR package using [nifi-nar-maven-plugin](https://mvnrepository.com/artifact/org.apache.nifi/nifi-nar-maven-plugin).
+
+NAR stands for NiFi Archive. It is a custom packaging mechanism used by Apache NiFi, to provide
+a bit of Java ClassLoader isolation. For more details, you can read this [blog post](https://medium.com/hashmapinc/nifi-nar-files-explained-14113f7796fd) to understand
+how NAR works. Pulsar uses the same mechanism for packaging all the [builtin connectors](io-connectors).
+
+All what you need is to include this [nifi-nar-maven-plugin](https://mvnrepository.com/artifact/org.apache.nifi/nifi-nar-maven-plugin) in your maven project for your connector. For example:
+
+```xml
+
+<plugins>
+ <plugin>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-nar-maven-plugin</artifactId>
+ <version>1.2.0</version>
+ </plugin>
+</plugins>
+
+```
+
+The {@inject: github:TwitterFirehose:/pulsar-io/twitter} connector is a good example to follow.
+
+#### Creating an Uber JAR
+
+An alternative approach is to create an _uber JAR_ that contains all of the connector's JAR files
+and other resource files. No directory internal structure is necessary.
+
+You can use [maven-shade-plugin](https://maven.apache.org/plugins/maven-shade-plugin/examples/includes-excludes.html) to create a Uber JAR. For example:
+
+```xml
+
+<plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>3.1.1</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ </filter>
+ </filters>
+ </configuration>
+ </execution>
+ </executions>
+</plugin>
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-managing.md b/site2/website-next/versioned_docs/version-2.4.1/io-managing.md
new file mode 100644
index 0000000..cce795b
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-managing.md
@@ -0,0 +1,193 @@
+---
+id: io-managing
+title: Managing Connectors
+sidebar_label: "Managing Connectors"
+original_id: io-managing
+---
+
+This section describes how to manage Pulsar IO connectors in a Pulsar cluster. You will learn how to:
+
+- Deploy builtin connectors
+- Monitor and update running connectors with Pulsar Admin CLI
+- Deploy customized connectors
+- Upgrade a connector
+
+## Using Builtin Connectors
+
+Pulsar bundles several [builtin connectors](io-overview.md#working-with-connectors) that should be used for moving data in and out
+of commonly used systems such as databases, messaging systems. Getting set up to use these builtin connectors is simple. You can follow
+the [instructions](getting-started-standalone.md#installing-builtin-connectors) on installing builtin connectors. After setup, all
+the builtin connectors will be automatically discovered by Pulsar brokers (or function-workers), so no additional installation steps are
+required.
+
+## Configuring Connectors
+
+Configuring Pulsar IO connectors is straightforward. What you need to do is to provide a yaml configuration file when your [run connectors](#running-connectors).
+The yaml configuration file basically tells Pulsar where to locate the sources and sinks and how to connect those sources and sinks with Pulsar topics.
+
+Below is an example yaml configuration file for Cassandra Sink:
+
+```shell
+
+tenant: public
+namespace: default
+name: cassandra-test-sink
+...
+# cassandra specific config
+configs:
+ roots: "localhost:9042"
+ keyspace: "pulsar_test_keyspace"
+ columnFamily: "pulsar_test_table"
+ keyname: "key"
+ columnName: "col"
+
+```
+
+The example yaml basically tells Pulsar which Cassandra cluster to connect, what is the `keyspace` and `columnFamily` to be used in Cassandra for collecting data,
+and how to map a Pulsar message into Cassandra table key and columns.
+
+For details, consult the documentation for [individual connectors](io-overview.md#working-with-connectors).
+
+## Running Connectors
+
+Pulsar connectors can be managed using the [`source`](reference-pulsar-admin.md#source) and [`sink`](reference-pulsar-admin.md#sink) commands of the [`pulsar-admin`](reference-pulsar-admin) CLI tool.
+
+### Running sources
+
+You can submit a source to be run in an existing Pulsar cluster using a command of this form:
+
+```bash
+
+$ ./bin/pulsar-admin sources create --classname <classname> --archive <jar-location> --tenant <tenant> --namespace <namespace> --name <source-name> --destination-topic-name <output-topic>
+
+```
+
+Here’s an example command:
+
+```bash
+
+bin/pulsar-admin sources create --classname org.apache.pulsar.io.twitter.TwitterFireHose --archive ~/application.jar --tenant test --namespace ns1 --name twitter-source --destination-topic-name twitter_data
+
+```
+
+Instead of submitting a source to run on an existing Pulsar cluster, you alternatively can run a source as a process on your local machine:
+
+```bash
+
+bin/pulsar-admin sources localrun --classname org.apache.pulsar.io.twitter.TwitterFireHose --archive ~/application.jar --tenant test --namespace ns1 --name twitter-source --destination-topic-name twitter_data
+
+```
+
+If you are submitting a built-in source, you don't need to specify `--classname` and `--archive`.
+You can simply specify the source type `--source-type`. The command to submit a built-in source is
+in following form:
+
+```bash
+
+./bin/pulsar-admin sources create \
+ --tenant <tenant> \
+ --namespace <namespace> \
+ --name <source-name> \
+ --destination-topic-name <input-topics> \
+ --source-type <source-type>
+
+```
+
+Here's an example to submit a Kafka source:
+
+```bash
+
+./bin/pulsar-admin sources create \
+ --tenant test-tenant \
+ --namespace test-namespace \
+ --name test-kafka-source \
+ --destination-topic-name pulsar_sink_topic \
+ --source-type kafka
+
+```
+
+### Running Sinks
+
+You can submit a sink to be run in an existing Pulsar cluster using a command of this form:
+
+```bash
+
+./bin/pulsar-admin sinks create --classname <classname> --archive <jar-location> --tenant test --namespace <namespace> --name <sink-name> --inputs <input-topics>
+
+```
+
+Here’s an example command:
+
+```bash
+
+./bin/pulsar-admin sinks create --classname org.apache.pulsar.io.cassandra --archive ~/application.jar --tenant test --namespace ns1 --name cassandra-sink --inputs test_topic
+
+```
+
+Instead of submitting a sink to run on an existing Pulsar cluster, you alternatively can run a sink as a process on your local machine:
+
+```bash
+
+./bin/pulsar-admin sinks localrun --classname org.apache.pulsar.io.cassandra --archive ~/application.jar --tenant test --namespace ns1 --name cassandra-sink --inputs test_topic
+
+```
+
+If you are submitting a built-in sink, you don't need to specify `--classname` and `--archive`.
+You can simply specify the sink type `--sink-type`. The command to submit a built-in sink is
+in following form:
+
+#### Note
+
+> The `sink-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+
+```bash
+
+./bin/pulsar-admin sinks create \
+ --tenant <tenant> \
+ --namespace <namespace> \
+ --name <sink-name> \
+ --inputs <input-topics> \
+ --sink-type <sink-type>
+
+```
+
+Here's an example to submit a Cassandra sink:
+
+```bash
+
+./bin/pulsar-admin sinks create \
+ --tenant test-tenant \
+ --namespace test-namespace \
+ --name test-cassandra-sink \
+ --inputs pulsar_input_topic \
+ --sink-type cassandra
+
+```
+
+## Monitoring Connectors
+
+Since Pulsar IO connectors are running as [Pulsar Functions](functions-overview), so you can use [`functions`](reference-pulsar-admin.md#source) commands
+available in the [`pulsar-admin`](reference-pulsar-admin) CLI tool.
+
+### Retrieve Connector Metadata
+
+```
+
+bin/pulsar-admin functions get \
+ --tenant <tenant> \
+ --namespace <namespace> \
+ --name <connector-name>
+
+```
+
+### Retrieve Connector Running Status
+
+```
+
+bin/pulsar-admin functions getstatus \
+ --tenant <tenant> \
+ --namespace <namespace> \
+ --name <connector-name>
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-overview.md b/site2/website-next/versioned_docs/version-2.4.1/io-overview.md
new file mode 100644
index 0000000..68960a8
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-overview.md
@@ -0,0 +1,158 @@
+---
+id: io-overview
+title: Pulsar connector overview
+sidebar_label: "Overview"
+original_id: io-overview
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+Messaging systems are most powerful when you can easily use them with external systems like databases and other messaging systems.
+
+**Pulsar IO connectors** enable you to easily create, deploy, and manage connectors that interact with external systems, such as [Apache Cassandra](https://cassandra.apache.org), [Aerospike](https://www.aerospike.com), and many others.
+
+
+## Concept
+
+Pulsar IO connectors come in two types: **source** and **sink**.
+
+This diagram illustrates the relationship between source, Pulsar, and sink:
+
+
+
+
+### Source
+
+> Sources **feed data from external systems into Pulsar**.
+
+Common sources include other messaging systems and firehose-style data pipeline APIs.
+
+For the complete list of Pulsar built-in source connectors, see [source connector](io-connectors.md#source-connector).
+
+### Sink
+
+> Sinks **feed data from Pulsar into external systems**.
+
+Common sinks include other messaging systems and SQL and NoSQL databases.
+
+For the complete list of Pulsar built-in sink connectors, see [sink connector](io-connectors.md#sink-connector).
+
+## Processing guarantee
+
+Processing guarantees are used to handle errors when writing messages to Pulsar topics.
+
+> Pulsar connectors and Functions use the **same** processing guarantees as below.
+
+Delivery semantic | Description
+:------------------|:-------
+`at-most-once` | Each message sent to a connector is to be **processed once** or **not to be processed**.
+`at-least-once` | Each message sent to a connector is to be **processed once** or **more than once**.
+`effectively-once` | Each message sent to a connector has **one output associated** with it.
+
+> Processing guarantees for connectors not just rely on Pulsar guarantee but also **relate to external systems**, that is, **the implementation of source and sink**.
+
+* Source: Pulsar ensures that writing messages to Pulsar topics respects to the processing guarantees. It is within Pulsar's control.
+
+* Sink: the processing guarantees rely on the sink implementation. If the sink implementation does not handle retries in an idempotent way, the sink does not respect to the processing guarantees.
+
+### Set
+
+When creating a connector, you can set the processing guarantee with the following semantics:
+
+* ATLEAST_ONCE
+
+* ATMOST_ONCE
+
+* EFFECTIVELY_ONCE
+
+> If `--processing-guarantees` is not specified when creating a connector, the default semantic is `ATLEAST_ONCE`.
+
+Here takes **Admin CLI** as an example. For more information about **REST API** or **JAVA Admin API**, see [here](io-use.md#create).
+
+<Tabs
+ defaultValue="Source"
+ values={[{"label":"Source","value":"Source"},{"label":"Sink","value":"Sink"}]}>
+
+<TabItem value="Source">
+
+```bash
+
+$ bin/pulsar-admin sources create \
+ --processing-guarantees ATMOST_ONCE \
+ # Other source configs
+
+```
+
+For more information about the options of `pulsar-admin sources create`, see [here](reference-connector-admin.md#create).
+
+</TabItem>
+<TabItem value="Sink">
+
+```bash
+
+$ bin/pulsar-admin sinks create \
+ --processing-guarantees EFFECTIVELY_ONCE \
+ # Other sink configs
+
+```
+
+For more information about the options of `pulsar-admin sinks create`, see [here](reference-connector-admin.md#create-1).
+
+</TabItem>
+
+</Tabs>
+
+### Update
+
+After creating a connector, you can update the processing guarantee with the following semantics:
+
+* ATLEAST_ONCE
+
+* ATMOST_ONCE
+
+* EFFECTIVELY_ONCE
+
+Here takes **Admin CLI** as an example. For more information about **REST API** or **JAVA Admin API**, see [here](io-use.md#create).
+
+<Tabs
+ defaultValue="Source"
+ values={[{"label":"Source","value":"Source"},{"label":"Sink","value":"Sink"}]}>
+
+<TabItem value="Source">
+
+```bash
+
+$ bin/pulsar-admin sources update \
+ --processing-guarantees EFFECTIVELY_ONCE \
+ # Other source configs
+
+```
+
+For more information about the options of `pulsar-admin sources update`, see [here](reference-connector-admin.md#update).
+
+</TabItem>
+<TabItem value="Sink">
+
+```bash
+
+$ bin/pulsar-admin sinks update \
+ --processing-guarantees ATMOST_ONCE \
+ # Other sink configs
+
+```
+
+For more information about the options of `pulsar-admin sinks update`, see [here](reference-connector-admin.md#update-1).
+
+</TabItem>
+
+</Tabs>
+
+
+## Work with connector
+
+You can manage Pulsar connectors (for example, create, update, start, stop, restart, reload, delete and perform other operations on connectors) via the [Connector Admin CLI](reference-connector-admin) with [sources](io-cli.md#sources) and [sinks](io-cli.md#sinks) subcommands.
+
+Connectors (sources and sinks) and Functions are components of instances, and they all run on Functions workers. When managing a source, sink or function via [Connector Admin CLI](reference-connector-admin.md) or [Functions Admin CLI](functions-cli), an instance is started on a worker. For more information, see [Functions worker](functions-worker.md#run-functions-worker-separately).
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-quickstart.md b/site2/website-next/versioned_docs/version-2.4.1/io-quickstart.md
new file mode 100644
index 0000000..10a0380
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-quickstart.md
@@ -0,0 +1,892 @@
+---
+id: io-quickstart
+title: "Tutorial: Connect Pulsar with Database"
+sidebar_label: "Get started"
+original_id: io-quickstart
+---
+
+This tutorial provides a hands-on look at how you can move data out of Pulsar without writing a single line of code.
+
+It is helpful to review the [concepts](io-overview) for Pulsar I/O with running the steps in this guide to gain a deeper understanding.
+
+At the end of this tutorial, you will be able to:
+
+- [Connect Pulsar to Apache Cassandra](#Connect-Pulsar-to-Apache-Cassandra)
+
+- [Connect Pulsar to MySQL](#Connect-Pulsar-to-MySQL)
+
+:::tip
+
+* These instructions assume you are running Pulsar in [standalone mode](getting-started-standalone). However all
+the commands used in this tutorial should be able to be used in a multi-nodes Pulsar cluster without any changes.
+* All the instructions are assumed to run at the root directory of a Pulsar binary distribution.
+
+:::
+
+## Install Pulsar and builtin connector
+
+Before connecting Pulsar to a database, we need to install Pulsar and the desired builtin connector.
+
+For more information about how to install a standalone Pulsar and builtin connectors, see [here](getting-started-standalone.md/#installing-pulsar).
+
+## Start a standalone Pulsar
+
+1. Start Pulsar locally.
+
+ ```bash
+
+ bin/pulsar standalone
+
+ ```
+
+ All the components of a Pulsar service will start in order. You can curl those pulsar service endpoints to make sure Pulsar service is up running correctly.
+
+2. Check Pulsar binary protocol port.
+
+ ```bash
+
+ telnet localhost 6650
+
+ ```
+
+3. Check Pulsar Function cluster.
+
+ ```bash
+
+ curl -s http://localhost:8080/admin/v2/worker/cluster
+
+ ```
+
+ Example output:
+
+ ```shell
+
+ [{"workerId":"c-standalone-fw-localhost-6750","workerHostname":"localhost","port":6750}]
+
+ ```
+
+4. Make sure public tenant and default namespace exist.
+
+ ```bash
+
+ curl -s http://localhost:8080/admin/v2/namespaces/public
+
+ ```
+
+ Example outoupt:
+
+ ```shell
+
+ ["public/default","public/functions"]
+
+ ```
+
+5. All builtin connectors should be listed as available.
+
+ ```bash
+
+ curl -s http://localhost:8080/admin/v2/functions/connectors
+
+ ```
+
+ Example output:
+
+ ```json
+
+ [{"name":"aerospike","description":"Aerospike database sink","sinkClass":"org.apache.pulsar.io.aerospike.AerospikeStringSink"},{"name":"cassandra","description":"Writes data into Cassandra","sinkClass":"org.apache.pulsar.io.cassandra.CassandraStringSink"},{"name":"kafka","description":"Kafka source and sink connector","sourceClass":"org.apache.pulsar.io.kafka.KafkaStringSource","sinkClass":"org.apache.pulsar.io.kafka.KafkaBytesSink"},{"name":"kinesis","description":"Kinesis sink conne [...]
+
+ ```
+
+ If an error occurred while starting Pulsar service, you may be able to seen exception at the terminal you are running `pulsar/standalone`,
+ or you can navigate the `logs` directory under the Pulsar directory to view the logs.
+
+## Connect Pulsar to Apache Cassandra
+
+:::tip
+
+Make sure you have docker available at your computer. If you don't have docker installed, follow the instructions [here](https://docs.docker.com/docker-for-mac/install/).
+
+:::
+
+We are using `cassandra` docker image to start a single-node cassandra cluster in Docker.
+
+### Setup the Cassandra cluster
+
+#### Start a Cassandra cluster
+
+```bash
+
+docker run -d --rm --name=cassandra -p 9042:9042 cassandra
+
+```
+
+Before moving to next steps, make sure the cassandra cluster is up running.
+
+1. Make sure the docker process is running.
+
+```bash
+
+docker ps
+
+```
+
+2. Check the cassandra logs to make sure cassandra process is running as expected.
+
+```bash
+
+docker logs cassandra
+
+```
+
+3. Check the cluster status
+
+```bash
+
+docker exec cassandra nodetool status
+
+```
+
+Example output:
+
+```
+
+Datacenter: datacenter1
+=======================
+Status=Up/Down
+|/ State=Normal/Leaving/Joining/Moving
+-- Address Load Tokens Owns (effective) Host ID Rack
+UN 172.17.0.2 103.67 KiB 256 100.0% af0e4b2f-84e0-4f0b-bb14-bd5f9070ff26 rack1
+
+```
+
+#### Create keyspace and table
+
+We are using `cqlsh` to connect to the cassandra cluster to create keyspace and table.
+
+```bash
+
+$ docker exec -ti cassandra cqlsh localhost
+Connected to Test Cluster at localhost:9042.
+[cqlsh 5.0.1 | Cassandra 3.11.2 | CQL spec 3.4.4 | Native protocol v4]
+Use HELP for help.
+cqlsh>
+
+```
+
+All the following commands are executed in `cqlsh`.
+
+#### Create keyspace `pulsar_test_keyspace`
+
+```bash
+
+cqlsh> CREATE KEYSPACE pulsar_test_keyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':1};
+
+```
+
+#### Create table `pulsar_test_table`
+
+```bash
+
+cqlsh> USE pulsar_test_keyspace;
+cqlsh:pulsar_test_keyspace> CREATE TABLE pulsar_test_table (key text PRIMARY KEY, col text);
+
+```
+
+### Configure a Cassandra sink
+
+Now that we have a Cassandra cluster running locally. In this section, we will configure a Cassandra sink connector.
+The Cassandra sink connector will read messages from a Pulsar topic and write the messages into a Cassandra table.
+
+In order to run a Cassandra sink connector, you need to prepare a yaml config file including information that Pulsar IO
+runtime needs to know. For example, how Pulsar IO can find the cassandra cluster, what is the keyspace and table that
+Pulsar IO will be using for writing Pulsar messages to.
+
+Create a file `examples/cassandra-sink.yml` and edit it to fill in following content:
+
+```
+
+configs:
+ roots: "localhost:9042"
+ keyspace: "pulsar_test_keyspace"
+ columnFamily: "pulsar_test_table"
+ keyname: "key"
+ columnName: "col"
+
+```
+
+To learn more about Cassandra Connector, see [Cassandra Connector](io-cassandra).
+
+### Submit a Cassandra sink
+
+Pulsar provides the [CLI](reference-cli-tools) for running and managing Pulsar I/O connectors.
+
+We can run following command to sink a sink connector with type `cassandra` and config file `examples/cassandra-sink.yml`.
+
+#### Note
+
+> The `sink-type` parameter of the currently built-in connectors is determined by the setting of the `name` parameter specified in the pulsar-io.yaml file.
+
+```shell
+
+bin/pulsar-admin sinks create \
+ --tenant public \
+ --namespace default \
+ --name cassandra-test-sink \
+ --sink-type cassandra \
+ --sink-config-file examples/cassandra-sink.yml \
+ --inputs test_cassandra
+
+```
+
+Once the command is executed, Pulsar will create a sink connector named `cassandra-test-sink` and the sink connector will be running
+as a Pulsar Function and write the messages produced in topic `test_cassandra` to Cassandra table `pulsar_test_table`.
+
+### Inspect the Cassandra sink
+
+You can use [sink CLI](reference-pulsar-admin.md#sink) and [source CLI](reference-pulsar-admin.md#source)
+for inspecting and managing the IO connectors.
+
+#### Retrieve Sink Info
+
+```bash
+
+bin/pulsar-admin sinks get \
+ --tenant public \
+ --namespace default \
+ --name cassandra-test-sink
+
+```
+
+Example output:
+
+```shell
+
+{
+ "tenant": "public",
+ "namespace": "default",
+ "name": "cassandra-test-sink",
+ "className": "org.apache.pulsar.io.cassandra.CassandraStringSink",
+ "inputSpecs": {
+ "test_cassandra": {
+ "isRegexPattern": false
+ }
+ },
+ "configs": {
+ "roots": "localhost:9042",
+ "keyspace": "pulsar_test_keyspace",
+ "columnFamily": "pulsar_test_table",
+ "keyname": "key",
+ "columnName": "col"
+ },
+ "parallelism": 1,
+ "processingGuarantees": "ATLEAST_ONCE",
+ "retainOrdering": false,
+ "autoAck": true,
+ "archive": "builtin://cassandra"
+}
+
+```
+
+#### Check Sink running status
+
+```bash
+
+bin/pulsar-admin sinks status \
+ --tenant public \
+ --namespace default \
+ --name cassandra-test-sink
+
+```
+
+Example output:
+
+```shell
+
+{
+ "numInstances" : 1,
+ "numRunning" : 1,
+ "instances" : [ {
+ "instanceId" : 0,
+ "status" : {
+ "running" : true,
+ "error" : "",
+ "numRestarts" : 0,
+ "numReadFromPulsar" : 0,
+ "numSystemExceptions" : 0,
+ "latestSystemExceptions" : [ ],
+ "numSinkExceptions" : 0,
+ "latestSinkExceptions" : [ ],
+ "numWrittenToSink" : 0,
+ "lastReceivedTime" : 0,
+ "workerId" : "c-standalone-fw-localhost-8080"
+ }
+ } ]
+}
+
+```
+
+### Verify the Cassandra sink
+
+Now lets produce some messages to the input topic of the Cassandra sink `test_cassandra`.
+
+```bash
+
+for i in {0..9}; do bin/pulsar-client produce -m "key-$i" -n 1 test_cassandra; done
+
+```
+
+Inspect the sink running status again. You should be able to see 10 messages are processed by the Cassandra sink.
+
+```bash
+
+bin/pulsar-admin sinks status \
+ --tenant public \
+ --namespace default \
+ --name cassandra-test-sink
+
+```
+
+Example output:
+
+```shell
+
+{
+ "numInstances" : 1,
+ "numRunning" : 1,
+ "instances" : [ {
+ "instanceId" : 0,
+ "status" : {
+ "running" : true,
+ "error" : "",
+ "numRestarts" : 0,
+ "numReadFromPulsar" : 10,
+ "numSystemExceptions" : 0,
+ "latestSystemExceptions" : [ ],
+ "numSinkExceptions" : 0,
+ "latestSinkExceptions" : [ ],
+ "numWrittenToSink" : 10,
+ "lastReceivedTime" : 1551685489136,
+ "workerId" : "c-standalone-fw-localhost-8080"
+ }
+ } ]
+}
+
+```
+
+Finally, lets inspect the results in Cassandra using `cqlsh`
+
+```bash
+
+docker exec -ti cassandra cqlsh localhost
+
+```
+
+Select the rows from the Cassandra table `pulsar_test_table`:
+
+```bash
+
+cqlsh> use pulsar_test_keyspace;
+cqlsh:pulsar_test_keyspace> select * from pulsar_test_table;
+
+ key | col
+--------+--------
+ key-5 | key-5
+ key-0 | key-0
+ key-9 | key-9
+ key-2 | key-2
+ key-1 | key-1
+ key-3 | key-3
+ key-6 | key-6
+ key-7 | key-7
+ key-4 | key-4
+ key-8 | key-8
+
+```
+
+### Delete the Cassandra Sink
+
+```shell
+
+bin/pulsar-admin sinks delete \
+ --tenant public \
+ --namespace default \
+ --name cassandra-test-sink
+
+```
+
+## Connect Pulsar to MySQL
+
+:::tip
+
+Make sure you have Docker available at your computer. If you don't have Docker installed, follow the instructions [here](https://docs.docker.com/docker-for-mac/install/).
+
+:::
+
+### Setup a MySQL cluster
+
+Use the MySQL 5.7 docker image to start a single-node MySQL cluster in Docker.
+
+1. Pull the MySQL 5.7 image from Docker Hub.
+
+ ```text
+
+ $ docker pull mysql:5.7
+
+ ```
+
+2. Start MySQL.
+
+ ```text
+
+ $ docker run -d -it --rm \
+ --name pulsar-mysql \
+ -p 3306:3306 \
+ -e MYSQL_ROOT_PASSWORD=jdbc \
+ -e MYSQL_USER=mysqluser \
+ -e MYSQL_PASSWORD=mysqlpw \
+ mysql:5.7
+
+ ```
+
+ :::tip
+
+ Flag | Description | This example
+ - | - | -
+ `-d` | To start a container in detached mode. | /
+ `-it` | Keep STDIN open even if not attached and allocate a terminal. | /
+ `--rm` | Remove the container automatically when it exits. | /
+ `-name` | Assign a name to the container. | This example specifies _pulsar-mysql_ for the container.
+ `-p` | Publish the port of the container to the host. | This example publishes the port _3306_ of the container to the host.
+ `-e` | Set environment variables. | This example sets the following variables:<br />- The password for the root user is _jdbc_. <br />- The name for the normal user is _mysqluser_. <br />- The password for the normal user is _mysqlpw_.
+ For more information about Docker command, see [here](https://docs.docker.com/engine/reference/commandline/run/).
+
+ :::
+
+3. Check if MySQL has been started successfully.
+
+ ```text
+
+ $ docker logs -f pulsar-mysql
+
+ ```
+
+ MySQL has been started successfully if the following message appears.
+
+ ```text
+
+ 2019-05-11T10:40:58.709964Z 0 [Note] Found ca.pem, server-cert.pem and server-key.pem in data directory. Trying to enable SSL support using them.
+ 2019-05-11T10:40:58.710155Z 0 [Warning] CA certificate ca.pem is self signed.
+ 2019-05-11T10:40:58.711921Z 0 [Note] Server hostname (bind-address): '*'; port: 3306
+ 2019-05-11T10:40:58.711985Z 0 [Note] IPv6 is available.
+ 2019-05-11T10:40:58.712695Z 0 [Note] - '::' resolves to '::';
+ 2019-05-11T10:40:58.712742Z 0 [Note] Server socket created on IP: '::'.
+ 2019-05-11T10:40:58.714334Z 0 [Warning] Insecure configuration for --pid-file: Location '/var/run/mysqld' in the path is accessible to all OS users. Consider choosing a different directory.
+ 2019-05-11T10:40:58.723802Z 0 [Note] Event Scheduler: Loaded 0 events
+ 2019-05-11T10:40:58.724200Z 0 [Note] mysqld: ready for connections.
+ Version: '5.7.26' socket: '/var/run/mysqld/mysqld.sock' port: 3306 MySQL Community Server (GPL)
+
+ ```
+
+4. Access to MySQL.
+
+ ```text
+
+ $ docker exec -it pulsar-mysql /bin/bash
+ mysql -h localhost -uroot -pjdbc
+
+ ```
+
+5. Create a _pulsar_mysql_jdbc_sink_ table.
+
+ ```text
+
+ $ create database pulsar_mysql_jdbc_sink;
+
+ $ use pulsar_mysql_jdbc_sink;
+
+ $ create table if not exists pulsar_mysql_jdbc_sink
+ (
+ id INT AUTO_INCREMENT,
+ name VARCHAR(255) NOT NULL,
+ primary key (id)
+ )
+ engine=innodb;
+
+ ```
+
+### Configure a JDBC sink
+
+Now that we have a MySQL running locally. In this section, we will configure a JDBC sink connector. The JDBC sink connector will read messages from a Pulsar topic and write messages into a MySQL table.
+
+1. Add a configuration file.
+
+ To run a JDBC sink connector, you need to prepare a yaml config file including the information that Pulsar IO runtime needs to know. For example, how Pulsar IO can find the MySQL cluster, what is the JDBCURL and the table that Pulsar IO will use for writing messages to.
+
+ Create a _pulsar-mysql-jdbc-sink.yaml_ file , copy the following contents to this file, and place the file in the `pulsar/connectors` folder.
+
+ ```text
+
+ configs:
+ userName: "root"
+ password: "jdbc"
+ jdbcUrl: "jdbc:mysql://127.0.0.1:3306/pulsar_mysql_jdbc_sink"
+ tableName: "pulsar_mysql_jdbc_sink"
+
+ ```
+
+2. Create a schema.
+
+ Create a _avro-schema_ file, copy the following contents to this file, and place the file in the `pulsar/connectors` folder.
+
+ ```text
+
+ {
+ "type": "AVRO",
+ "schema": "{\"type\":\"record\",\"name\":\"Test\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"]},{\"name\":\"name\",\"type\":[\"null\",\"string\"]}]}",
+ "properties": {}
+ }
+
+ ```
+
+ :::tip
+
+ For more information about AVRO, see [Apache Avro Documentation](https://avro.apache.org/docs/1.8.2/).
+
+ :::
+
+3. Upload a schema to a topic.
+
+ This example uploads the _avro-schema_ schema to the _pulsar-mysql-jdbc-sink-topic_ topic.
+
+ ```text
+
+ $ bin/pulsar-admin schemas upload pulsar-mysql-jdbc-sink-topic -f ./connectors/avro-schema
+
+ ```
+
+4. Check if the schema has been uploaded successfully.
+
+ ```text
+
+ $ bin/pulsar-admin schemas get pulsar-mysql-jdbc-sink-topic
+
+ ```
+
+ The schema has been uploaded successfully if the following message appears.
+
+ ```text
+
+ {"name":"pulsar-mysql-jdbc-sink-topic","schema":"{\"type\":\"record\",\"name\":\"Test\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"int\"]},{\"name\":\"name\",\"type\":[\"null\",\"string\"]}]}","type":"AVRO","properties":{}}
+
+ ```
+
+### Submit a JDBC sink
+
+Pulsar provides the [CLI](admin-api-overview) for running and managing Pulsar I/O connectors.
+
+This example creates a sink connector and specifies the desired information.
+
+```text
+
+$ bin/pulsar-admin sinks create \
+--archive ./connectors/pulsar-io-jdbc-@pulsar:version@.nar \
+--inputs pulsar-mysql-jdbc-sink-topic \
+--name pulsar-mysql-jdbc-sink \
+--sink-config-file ./connectors/pulsar-mysql-jdbc-sink.yaml \
+--parallelism 1
+
+```
+
+Once the command is executed, Pulsar will create a sink connector named _pulsar-mysql-jdbc-sink_, and the sink connector will be running as a Pulsar Function and write the messages produced in the _pulsar-mysql-jdbc-sink-topic_ topic to the MySQL _pulsar_mysql_jdbc_sink_ table.
+
+:::tip
+
+Flag | Description | This example
+- | - | - |
+`--archive` | Path to the archive file for the sink. | _pulsar-io-jdbc-@pulsar:version@.nar_
+`--inputs` | The input topic or topics of the sink. <br /> (Multiple topics can be specified as a comma-separated list.)
+`--name` | The name of the sink. | _pulsar-mysql-jdbc-sink_
+`--sink-config-file` | The path to a YAML config file specifying the configuration of the sink. | _pulsar-mysql-jdbc-sink.yaml_
+`--parallelism` | The parallelism factor of the sink. <br /> For example, the number of sink instances to run. | _1_
+For more information about `pulsar-admin sinks create options`, see [here](reference-pulsar-admin.md#create-3).
+
+:::
+
+The sink has been created successfully if the following message appears.
+
+```text
+
+"Created successfully"
+
+```
+
+### Inspect a JDBC sink
+
+#### List all running JDBC sink(s)
+
+This example lists all running sink connectors.
+
+```text
+
+$ bin/pulsar-admin sinks list \
+--tenant public \
+--namespace default
+
+```
+
+The result shows that only the _mysql-jdbc-sink_ sink is running.
+
+```text
+
+[
+ "pulsar-mysql-jdbc-sink"
+]
+
+```
+
+#### Get information of a JDBC sink
+
+This example gets the information about the _pulsar-mysql-jdbc-sink_ sink connector.
+
+```text
+
+$ bin/pulsar-admin sinks get \
+--tenant public \
+--namespace default \
+--name pulsar-mysql-jdbc-sink
+
+```
+
+The result show the information of the sink connector, including tenant, namespace, topic and so on.
+
+```text
+
+{
+ "tenant": "public",
+ "namespace": "default",
+ "name": "pulsar-mysql-jdbc-sink",
+ "className": "org.apache.pulsar.io.jdbc.JdbcAutoSchemaSink",
+ "inputSpecs": {
+ "pulsar-mysql-jdbc-sink-topic": {
+ "isRegexPattern": false
+ }
+ },
+ "configs": {
+ "password": "jdbc",
+ "jdbcUrl": "jdbc:mysql://127.0.0.1:3306/pulsar_mysql_jdbc_sink",
+ "userName": "root",
+ "tableName": "pulsar_mysql_jdbc_sink"
+ },
+ "parallelism": 1,
+ "processingGuarantees": "ATLEAST_ONCE",
+ "retainOrdering": false,
+ "autoAck": true
+}
+
+```
+
+#### Get status of a JDBC sink
+
+This example checks the current status of the _pulsar-mysql-jdbc-sink_ sink connector.
+
+```text
+
+$ bin/pulsar-admin sinks status \
+--tenant public \
+--namespace default \
+--name pulsar-mysql-jdbc-sink
+
+```
+
+The result shows the current status of sink connector, including the number of instance, running status, worker ID and so on.
+
+```text
+
+{
+ "numInstances" : 1,
+ "numRunning" : 1,
+ "instances" : [ {
+ "instanceId" : 0,
+ "status" : {
+ "running" : true,
+ "error" : "",
+ "numRestarts" : 0,
+ "numReadFromPulsar" : 0,
+ "numSystemExceptions" : 0,
+ "latestSystemExceptions" : [ ],
+ "numSinkExceptions" : 0,
+ "latestSinkExceptions" : [ ],
+ "numWrittenToSink" : 0,
+ "lastReceivedTime" : 0,
+ "workerId" : "c-standalone-fw-192.168.2.52-8080"
+ }
+ } ]
+}
+
+```
+
+### Stop a JDBC sink
+
+This example stops the _pulsar-mysql-jdbc-sink_ sink instance.
+
+```text
+
+$ bin/pulsar-admin sinks stop \
+--tenant public \
+--namespace default \
+--name pulsar-mysql-jdbc-sink \
+--instance-id 0
+
+```
+
+The sink instance has been stopped successfully if the following message disappears.
+
+```text
+
+"Stopped successfully"
+
+```
+
+### Restart a JDBC sink
+
+This example starts the _pulsar-mysql-jdbc-sink_ sink instance.
+
+```text
+
+$ bin/pulsar-admin sinks start \
+--tenant public \
+--namespace default \
+--name pulsar-mysql-jdbc-sink \
+--instance-id 0
+
+```
+
+The sink instance has been started successfully if the following message disappears.
+
+```text
+
+"Started successfully"
+
+```
+
+:::tip
+
+Optionally, you can run a standalone sink connector using `pulsar-admin sinks localrun options`.
+Note that `pulsar-admin sinks localrun options` runs a sink connector locally, while `pulsar-admin sinks start options` starts a sink connector in a cluster.
+For more information about `pulsar-admin sinks localrun options`, see [here](reference-pulsar-admin.md#localrun-1).
+
+:::
+
+### Update a JDBC sink
+
+This example updates the parallelism of the _pulsar-mysql-jdbc-sink_ sink connector to 2.
+
+```text
+
+$ bin/pulsar-admin sinks update \
+--name pulsar-mysql-jdbc-sink \
+--parallelism 2
+
+```
+
+:::tip
+
+For more information about `pulsar-admin sinks update options`, see [here](reference-pulsar-admin.md#update-2).
+
+:::
+
+The sink connector has been updated successfully if the following message disappears.
+
+```text
+
+"Updated successfully"
+
+```
+
+This example double-checks the information.
+
+```text
+
+$ bin/pulsar-admin sinks get \
+--tenant public \
+--namespace default \
+--name pulsar-mysql-jdbc-sink
+
+```
+
+The result shows that the parallelism is 2.
+
+```text
+
+{
+ "tenant": "public",
+ "namespace": "default",
+ "name": "pulsar-mysql-jdbc-sink",
+ "className": "org.apache.pulsar.io.jdbc.JdbcAutoSchemaSink",
+ "inputSpecs": {
+ "pulsar-mysql-jdbc-sink-topic": {
+ "isRegexPattern": false
+ }
+ },
+ "configs": {
+ "password": "jdbc",
+ "jdbcUrl": "jdbc:mysql://127.0.0.1:3306/pulsar_mysql_jdbc_sink",
+ "userName": "root",
+ "tableName": "pulsar_mysql_jdbc_sink"
+ },
+ "parallelism": 2,
+ "processingGuarantees": "ATLEAST_ONCE",
+ "retainOrdering": false,
+ "autoAck": true
+}
+
+```
+
+### Delete a JDBC sink
+
+This example deletes the _pulsar-mysql-jdbc-sink_ sink connector.
+
+```text
+
+$ bin/pulsar-admin sinks delete \
+--tenant public \
+--namespace default \
+--name pulsar-mysql-jdbc-sink
+
+```
+
+:::tip
+
+For more information about `pulsar-admin sinks delete options`, see [here](reference-pulsar-admin.md#delete-4).
+
+:::
+
+The sink connector has been deleted successfully if the following message appears.
+
+```text
+
+"Deleted successfully"
+
+```
+
+This example double-checks the existence of the sink connector.
+
+```text
+
+$ bin/pulsar-admin sinks get \
+--tenant public \
+--namespace default \
+--name pulsar-mysql-jdbc-sink
+
+```
+
+The results shows that the sink connector does not exist.
+
+```text
+
+HTTP 404 Not Found
+
+Reason: Sink pulsar-mysql-jdbc-sink doesn't exist
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/io-use.md b/site2/website-next/versioned_docs/version-2.4.1/io-use.md
new file mode 100644
index 0000000..c960f38
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/io-use.md
@@ -0,0 +1,1745 @@
+---
+id: io-use
+title: How to use Pulsar connectors
+sidebar_label: "Use"
+original_id: io-use
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+
+This guide describes how to use Pulsar connectors.
+
+## Install a connector
+
+Pulsar bundles several [builtin connectors](io-connectors) used to move data in and out of commonly used systems (such as database and messaging system). Optionally, you can create and use your desired non-builtin connectors.
+
+:::note
+
+When using a non-builtin connector, you need to specify the path of a archive file for the connector.
+
+:::
+
+To set up a builtin connector, follow
+the instructions [here](getting-started-standalone.md#installing-builtin-connectors).
+
+After the setup, the builtin connector is automatically discovered by Pulsar brokers (or function-workers), so no additional installation steps are required.
+
+## Configure a connector
+
+You can configure the following information:
+
+* [Configure a default storage location for a connector](#configure-a-default-storage-location-for-a-connector)
+
+* [Configure a connector with a YAML file](#configure-a-connector-with-yaml-file)
+
+### Configure a default storage location for a connector
+
+To configure a default folder for builtin connectors, set the `connectorsDirectory` parameter in the `./conf/functions_worker.yml` configuration file.
+
+**Example**
+
+Set the `./connectors` folder as the default storage location for builtin connectors.
+
+```
+
+########################
+# Connectors
+########################
+
+connectorsDirectory: ./connectors
+
+```
+
+### Configure a connector with a YAML file
+
+To configure a connector, you need to provide a YAML configuration file when creating a connector.
+
+The YAML configuration file tells Pulsar where to locate connectors and how to connect connectors with Pulsar topics.
+
+**Example 1**
+
+Below is a YAML configuration file of a Cassandra sink, which tells Pulsar:
+
+* Which Cassandra cluster to connect
+
+* What is the `keyspace` and `columnFamily` to be used in Cassandra for collecting data
+
+* How to map Pulsar messages into Cassandra table key and columns
+
+```shell
+
+tenant: public
+namespace: default
+name: cassandra-test-sink
+...
+# cassandra specific config
+configs:
+ roots: "localhost:9042"
+ keyspace: "pulsar_test_keyspace"
+ columnFamily: "pulsar_test_table"
+ keyname: "key"
+ columnName: "col"
+
+```
+
+**Example 2**
+
+Below is a YAML configuration file of a Kafka source.
+
+```shell
+
+configs:
+ bootstrapServers: "pulsar-kafka:9092"
+ groupId: "test-pulsar-io"
+ topic: "my-topic"
+ sessionTimeoutMs: "10000"
+ autoCommitEnabled: "false"
+
+```
+
+**Example 3**
+
+Below is a YAML configuration file of a MySQL JDBC sink.
+
+```shell
+
+configs:
+ userName: "root"
+ password: "jdbc"
+ jdbcUrl: "jdbc:mysql://127.0.0.1:3306/test_jdbc"
+ tableName: "test_jdbc"
+
+```
+
+## Get available connectors
+
+Before starting using connectors, you can perform the following operations:
+
+* [Reload connectors](#reload)
+
+* [Get a list of available connectors](#get-available-connectors)
+
+### `reload`
+
+If you add or delete a nar file in a connector folder, reload the available builtin connector before using it.
+
+#### Source
+
+Use the `reload` subcommand.
+
+```shell
+
+$ pulsar-admin sources reload
+
+```
+
+For more information, see [`here`](reference-connector-admin/#reload).
+
+#### Sink
+
+Use the `reload` subcommand.
+
+```shell
+
+$ pulsar-admin sinks reload
+
+```
+
+For more information, see [`here`](reference-connector-admin/#reload-1).
+
+### `available`
+
+After reloading connectors (optional), you can get a list of available connectors.
+
+#### Source
+
+Use the `available-sources` subcommand.
+
+```shell
+
+$ pulsar-admin sources available-sources
+
+```
+
+#### Sink
+
+Use the `available-sinks` subcommand.
+
+```shell
+
+$ pulsar-admin sinks available-sinks
+
+```
+
+## Run a connector
+
+To run a connector, you can perform the following operations:
+
+* [Create a connector](#create)
+
+* [Start a connector](#start)
+
+* [Run a connector locally](#localrun)
+
+### `create`
+
+You can create a connector using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Create a source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `create` subcommand.
+
+```
+
+$ pulsar-admin sources create options
+
+```
+
+For more information, see [here](reference-connector-admin.md#create).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sourceName|operation/registerSource?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Create a source connector with a **local file**.
+
+ ```java
+
+ void createSource(SourceConfig sourceConfig,
+ String fileName)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ |Name|Description
+ |---|---
+ `sourceConfig` | The source configuration object
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`createSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#createSource-SourceConfig-java.lang.String-).
+
+* Create a source connector using a **remote file** with a URL from which fun-pkg can be downloaded.
+
+ ```java
+
+ void createSourceWithUrl(SourceConfig sourceConfig,
+ String pkgUrl)
+ throws PulsarAdminException
+
+ ```
+
+ Supported URLs are `http` and `file`.
+
+ **Example**
+
+ * HTTP: http://www.repo.com/fileName.jar
+
+ * File: file:///dir/fileName.jar
+
+ **Parameter**
+
+ Parameter| Description
+ |---|---
+ `sourceConfig` | The source configuration object
+ `pkgUrl` | URL from which pkg can be downloaded
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`createSourceWithUrl`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#createSourceWithUrl-SourceConfig-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Create a sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `create` subcommand.
+
+```
+
+$ pulsar-admin sinks create options
+
+```
+
+For more information, see [here](reference-connector-admin.md#create-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sinks/:tenant/:namespace/:sinkName|operation/registerSink?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Create a sink connector with a **local file**.
+
+ ```java
+
+ void createSink(SinkConfig sinkConfig,
+ String fileName)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ |Name|Description
+ |---|---
+ `sinkConfig` | The sink configuration object
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`createSink`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#createSink-SinkConfig-java.lang.String-).
+
+* Create a sink connector using a **remote file** with a URL from which fun-pkg can be downloaded.
+
+ ```java
+
+ void createSinkWithUrl(SinkConfig sinkConfig,
+ String pkgUrl)
+ throws PulsarAdminException
+
+ ```
+
+ Supported URLs are `http` and `file`.
+
+ **Example**
+
+ * HTTP: http://www.repo.com/fileName.jar
+
+ * File: file:///dir/fileName.jar
+
+ **Parameter**
+
+ Parameter| Description
+ |---|---
+ `sinkConfig` | The sink configuration object
+ `pkgUrl` | URL from which pkg can be downloaded
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`createSinkWithUrl`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#createSinkWithUrl-SinkConfig-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+### `start`
+
+You can start a connector using **Admin CLI** or **REST API**.
+
+#### Source
+
+Start a source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `start` subcommand.
+
+```
+
+$ pulsar-admin sources start options
+
+```
+
+For more information, see [here](reference-connector-admin.md#start).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Start **all** source connectors.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sourceName/start|operation/startSource?version=@pulsar:version_number@}
+
+* Start a **specified** source connector.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sourceName/:instanceId/start|operation/startSource?version=@pulsar:version_number@}
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Start a sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `start` subcommand.
+
+```
+
+$ pulsar-admin sinks start options
+
+```
+
+For more information, see [here](reference-connector-admin.md#start-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Start **all** sink connectors.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sinkName/start|operation/startSink?version=@pulsar:version_number@}
+
+* Start a **specified** sink connector.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sinks/:tenant/:namespace/:sourceName/:instanceId/start|operation/startSink?version=@pulsar:version_number@}
+
+</TabItem>
+
+</Tabs>
+
+### `localrun`
+
+You can run a connector locally rather than deploying it on a Pulsar cluster using **Admin CLI**.
+
+#### Source
+
+Run a source connector locally.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `localrun` subcommand.
+
+```
+
+$ pulsar-admin sources localrun options
+
+```
+
+For more information, see [here](reference-connector-admin.md#localrun).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Run a sink connector locally.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `localrun` subcommand.
+
+```
+
+$ pulsar-admin sinks localrun options
+
+```
+
+For more information, see [here](reference-connector-admin.md#localrun-1).
+
+</TabItem>
+
+</Tabs>
+
+## Monitor a connector
+
+To monitor a connector, you can perform the following operations:
+
+* [Get the information of a connector](#get)
+
+* [Get the list of all running connectors](#list)
+
+* [Get the current status of a connector](#status)
+
+### `get`
+
+You can get the information of a connector using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Get the information of a source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `get` subcommand.
+
+```
+
+$ pulsar-admin sources get options
+
+```
+
+For more information, see [here](reference-connector-admin.md#get).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sources/:tenant/:namespace/:sourceName|operation/getSourceInfo?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+```java
+
+SourceConfig getSource(String tenant,
+ String namespace,
+ String source)
+ throws PulsarAdminException
+
+```
+
+**Example**
+
+This is a sourceConfig.
+
+```java
+
+{
+ "tenant": "tenantName",
+ "namespace": "namespaceName",
+ "name": "sourceName",
+ "className": "className",
+ "topicName": "topicName",
+ "configs": {},
+ "parallelism": 1,
+ "processingGuarantees": "ATLEAST_ONCE",
+ "resources": {
+ "cpu": 1.0,
+ "ram": 1073741824,
+ "disk": 10737418240
+ }
+}
+
+```
+
+This is a sourceConfig example.
+
+```
+
+{
+ "tenant": "public",
+ "namespace": "default",
+ "name": "debezium-mysql-source",
+ "className": "org.apache.pulsar.io.debezium.mysql.DebeziumMysqlSource",
+ "topicName": "debezium-mysql-topic",
+ "configs": {
+ "database.user": "debezium",
+ "database.server.id": "184054",
+ "database.server.name": "dbserver1",
+ "database.port": "3306",
+ "database.hostname": "localhost",
+ "database.password": "dbz",
+ "database.history.pulsar.service.url": "pulsar://127.0.0.1:6650",
+ "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+ "database.whitelist": "inventory",
+ "key.converter": "org.apache.kafka.connect.json.JsonConverter",
+ "database.history": "org.apache.pulsar.io.debezium.PulsarDatabaseHistory",
+ "pulsar.service.url": "pulsar://127.0.0.1:6650",
+ "database.history.pulsar.topic": "history-topic2"
+ },
+ "parallelism": 1,
+ "processingGuarantees": "ATLEAST_ONCE",
+ "resources": {
+ "cpu": 1.0,
+ "ram": 1073741824,
+ "disk": 10737418240
+ }
+}
+
+```
+
+**Exception**
+
+Exception name | Description
+|---|---
+`PulsarAdminException.NotAuthorizedException` | You don't have the admin permission
+`PulsarAdminException.NotFoundException` | Cluster doesn't exist
+`PulsarAdminException` | Unexpected error
+
+For more information, see [`getSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#getSource-java.lang.String-java.lang.String-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Get the information of a sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `get` subcommand.
+
+```
+
+$ pulsar-admin sinks get options
+
+```
+
+For more information, see [here](reference-connector-admin.md#get-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sinks/:tenant/:namespace/:sinkName|operation/getSinkInfo?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+```java
+
+SinkConfig getSink(String tenant,
+ String namespace,
+ String sink)
+ throws PulsarAdminException
+
+```
+
+**Example**
+
+This is a sinkConfig.
+
+```
+
+{
+"tenant": "tenantName",
+"namespace": "namespaceName",
+"name": "sinkName",
+"className": "className",
+"inputSpecs": {
+"topicName": {
+ "isRegexPattern": false
+}
+},
+"configs": {},
+"parallelism": 1,
+"processingGuarantees": "ATLEAST_ONCE",
+"retainOrdering": false,
+"autoAck": true
+}
+
+```
+
+This is a sinkConfig example.
+
+```
+
+{
+"tenant": "public",
+"namespace": "default",
+"name": "pulsar-mysql-jdbc-sink",
+"className": "org.apache.pulsar.io.jdbc.JdbcAutoSchemaSink",
+"inputSpecs": {
+"pulsar-mysql-jdbc-sink-topic": {
+ "isRegexPattern": false
+}
+},
+"configs": {
+"password": "jdbc",
+"jdbcUrl": "jdbc:mysql://127.0.0.1:3306/pulsar_mysql_jdbc_sink",
+"userName": "root",
+"tableName": "pulsar_mysql_jdbc_sink"
+},
+"parallelism": 1,
+"processingGuarantees": "ATLEAST_ONCE",
+"retainOrdering": false,
+"autoAck": true
+}
+
+```
+
+**Parameter description**
+
+Name| Description
+|---|---
+`tenant` | Tenant name
+`namespace` | Namespace name
+`sink` | Sink name
+
+For more information, see [`getSink`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#getSink-java.lang.String-java.lang.String-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+### `list`
+
+You can get the list of all running connectors using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Get the list of all running source connectors.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `list` subcommand.
+
+```
+
+$ pulsar-admin sources list options
+
+```
+
+For more information, see [here](reference-connector-admin.md#list).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sources/:tenant/:namespace/|operation/listSources?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+```java
+
+List<String> listSources(String tenant,
+ String namespace)
+ throws PulsarAdminException
+
+```
+
+**Response example**
+
+```java
+
+["f1", "f2", "f3"]
+
+```
+
+**Exception**
+
+Exception name | Description
+|---|---
+`PulsarAdminException.NotAuthorizedException` | You don't have the admin permission
+`PulsarAdminException` | Unexpected error
+
+For more information, see [`listSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#listSources-java.lang.String-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Get the list of all running sink connectors.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `list` subcommand.
+
+```
+
+$ pulsar-admin sinks list options
+
+```
+
+For more information, see [here](reference-connector-admin.md#list-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sinks/:tenant/:namespace/|operation/listSinks?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+```java
+
+List<String> listSinks(String tenant,
+ String namespace)
+ throws PulsarAdminException
+
+```
+
+**Response example**
+
+```java
+
+["f1", "f2", "f3"]
+
+```
+
+**Exception**
+
+Exception name | Description
+|---|---
+`PulsarAdminException.NotAuthorizedException` | You don't have the admin permission
+`PulsarAdminException` | Unexpected error
+
+For more information, see [`listSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#listSinks-java.lang.String-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+### `status`
+
+You can get the current status of a connector using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Get the current status of a source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `status` subcommand.
+
+```
+
+$ pulsar-admin sources status options
+
+```
+
+For more information, see [here](reference-connector-admin.md#status).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Get the current status of **all** source connectors.
+
+ Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sources/:tenant/:namespace/:sourceName/status|operation/getSourceStatus?version=@pulsar:version_number@}
+
+* Gets the current status of a **specified** source connector.
+
+ Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sources/:tenant/:namespace/:sourceName/:instanceId/status|operation/getSourceStatus?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Get the current status of **all** source connectors.
+
+ ```java
+
+ SourceStatus getSourceStatus(String tenant,
+ String namespace,
+ String source)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ Parameter| Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `sink` | Source name
+
+ **Exception**
+
+ Name | Description
+ |---|---
+ `PulsarAdminException` | Unexpected error
+
+ For more information, see [`getSourceStatus`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#getSource-java.lang.String-java.lang.String-java.lang.String-).
+
+* Gets the current status of a **specified** source connector.
+
+ ```java
+
+ SourceStatus.SourceInstanceStatus.SourceInstanceStatusData getSourceStatus(String tenant,
+ String namespace,
+ String source,
+ int id)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ Parameter| Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `sink` | Source name
+ `id` | Source instanceID
+
+ **Exception**
+
+ Exception name | Description
+ |---|---
+ `PulsarAdminException` | Unexpected error
+
+ For more information, see [`getSourceStatus`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#getSourceStatus-java.lang.String-java.lang.String-java.lang.String-int-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Get the current status of a Pulsar sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `status` subcommand.
+
+```
+
+$ pulsar-admin sinks status options
+
+```
+
+For more information, see [here](reference-connector-admin.md#status-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Get the current status of **all** sink connectors.
+
+ Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sinks/:tenant/:namespace/:sinkName/status|operation/getSinkStatus?version=@pulsar:version_number@}
+
+* Gets the current status of a **specified** sink connector.
+
+ Send a `GET` request to this endpoint: {@inject: endpoint|GET|/admin/v3/sinks/:tenant/:namespace/:sourceName/:instanceId/status|operation/getSinkInstanceStatus?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Get the current status of **all** sink connectors.
+
+ ```java
+
+ SinkStatus getSinkStatus(String tenant,
+ String namespace,
+ String sink)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ Parameter| Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `sink` | Source name
+
+ **Exception**
+
+ Exception name | Description
+ |---|---
+ `PulsarAdminException` | Unexpected error
+
+ For more information, see [`getSinkStatus`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#getSinkStatus-java.lang.String-java.lang.String-java.lang.String-).
+
+* Gets the current status of a **specified** source connector.
+
+ ```java
+
+ SinkStatus.SinkInstanceStatus.SinkInstanceStatusData getSinkStatus(String tenant,
+ String namespace,
+ String sink,
+ int id)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ Parameter| Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `sink` | Source name
+ `id` | Sink instanceID
+
+ **Exception**
+
+ Exception name | Description
+ |---|---
+ `PulsarAdminException` | Unexpected error
+
+ For more information, see [`getSinkStatusWithInstanceID`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#getSinkStatus-java.lang.String-java.lang.String-java.lang.String-int-).
+
+</TabItem>
+
+</Tabs>
+
+## Update a connector
+
+### `update`
+
+You can update a running connector using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Update a running Pulsar source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `update` subcommand.
+
+```
+
+$ pulsar-admin sources update options
+
+```
+
+For more information, see [here](reference-connector-admin.md#update).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `PUT` request to this endpoint: {@inject: endpoint|PUT|/admin/v3/sources/:tenant/:namespace/:sourceName|operation/updateSource?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Update a running source connector with a **local file**.
+
+ ```java
+
+ void updateSource(SourceConfig sourceConfig,
+ String fileName)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ |`sourceConfig` | The source configuration object
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ |`PulsarAdminException.NotAuthorizedException`| You don't have the admin permission
+ | `PulsarAdminException.NotFoundException` | Cluster doesn't exist
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`updateSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#updateSource-SourceConfig-java.lang.String-).
+
+* Update a source connector using a **remote file** with a URL from which fun-pkg can be downloaded.
+
+ ```java
+
+ void updateSourceWithUrl(SourceConfig sourceConfig,
+ String pkgUrl)
+ throws PulsarAdminException
+
+ ```
+
+ Supported URLs are `http` and `file`.
+
+ **Example**
+
+ * HTTP: http://www.repo.com/fileName.jar
+
+ * File: file:///dir/fileName.jar
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ | `sourceConfig` | The source configuration object
+ | `pkgUrl` | URL from which pkg can be downloaded
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ |`PulsarAdminException.NotAuthorizedException`| You don't have the admin permission
+ | `PulsarAdminException.NotFoundException` | Cluster doesn't exist
+ | `PulsarAdminException` | Unexpected error
+
+For more information, see [`createSourceWithUrl`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#updateSourceWithUrl-SourceConfig-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Update a running Pulsar sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `update` subcommand.
+
+```
+
+$ pulsar-admin sinks update options
+
+```
+
+For more information, see [here](reference-connector-admin.md#update-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+Send a `PUT` request to this endpoint: {@inject: endpoint|PUT|/admin/v3/sinks/:tenant/:namespace/:sinkName|operation/updateSink?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Update a running sink connector with a **local file**.
+
+ ```java
+
+ void updateSink(SinkConfig sinkConfig,
+ String fileName)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ |`sinkConfig` | The sink configuration object
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ |`PulsarAdminException.NotAuthorizedException`| You don't have the admin permission
+ | `PulsarAdminException.NotFoundException` | Cluster doesn't exist
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`updateSink`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#updateSink-SinkConfig-java.lang.String-).
+
+* Update a sink connector using a **remote file** with a URL from which fun-pkg can be downloaded.
+
+ ```java
+
+ void updateSinkWithUrl(SinkConfig sinkConfig,
+ String pkgUrl)
+ throws PulsarAdminException
+
+ ```
+
+ Supported URLs are `http` and `file`.
+
+ **Example**
+
+ * HTTP: http://www.repo.com/fileName.jar
+
+ * File: file:///dir/fileName.jar
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ | `sinkConfig` | The sink configuration object
+ | `pkgUrl` | URL from which pkg can be downloaded
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ |`PulsarAdminException.NotAuthorizedException`| You don't have the admin permission
+ |`PulsarAdminException.NotFoundException` | Cluster doesn't exist
+ |`PulsarAdminException` | Unexpected error
+
+For more information, see [`updateSinkWithUrl`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#updateSinkWithUrl-SinkConfig-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+## Stop a connector
+
+### `stop`
+
+You can stop a connector using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Stop a source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `stop` subcommand.
+
+```
+
+$ pulsar-admin sources stop options
+
+```
+
+For more information, see [here](reference-connector-admin.md#stop).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Stop **all** source connectors.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sourceName|operation/stopSource?version=@pulsar:version_number@}
+
+* Stop a **specified** source connector.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sourceName/:instanceId|operation/stopSource?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Stop **all** source connectors.
+
+ ```java
+
+ void stopSource(String tenant,
+ String namespace,
+ String source)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `source` | Source name
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`stopSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#stopSource-java.lang.String-java.lang.String-java.lang.String-).
+
+* Stop a **specified** source connector.
+
+ ```java
+
+ void stopSource(String tenant,
+ String namespace,
+ String source,
+ int instanceId)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `source` | Source name
+ `instanceId` | Source instanceID
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`stopSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#stopSource-java.lang.String-java.lang.String-java.lang.String-int-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Stop a sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `stop` subcommand.
+
+```
+
+$ pulsar-admin sinks stop options
+
+```
+
+For more information, see [here](reference-connector-admin.md#stop-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Stop **all** sink connectors.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sinks/:tenant/:namespace/:sinkName/stop|operation/stopSink?version=@pulsar:version_number@}
+
+* Stop a **specified** sink connector.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sinkeName/:instanceId/stop|operation/stopSink?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Stop **all** sink connectors.
+
+ ```java
+
+ void stopSink(String tenant,
+ String namespace,
+ String sink)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `source` | Source name
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`stopSink`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#stopSink-java.lang.String-java.lang.String-java.lang.String-).
+
+* Stop a **specified** sink connector.
+
+ ```java
+
+ void stopSink(String tenant,
+ String namespace,
+ String sink,
+ int instanceId)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `source` | Source name
+ `instanceId` | Source instanceID
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`stopSink`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#stopSink-java.lang.String-java.lang.String-java.lang.String-int-).
+
+</TabItem>
+
+</Tabs>
+
+## Restart a connector
+
+### `restart`
+
+You can restart a connector using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Restart a source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `restart` subcommand.
+
+```
+
+$ pulsar-admin sources restart options
+
+```
+
+For more information, see [here](reference-connector-admin.md#restart).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Restart **all** source connectors.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sourceName/restart|operation/restartSource?version=@pulsar:version_number@}
+
+* Restart a **specified** source connector.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sourceName/:instanceId/restart|operation/restartSource?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Restart **all** source connectors.
+
+ ```java
+
+ void restartSource(String tenant,
+ String namespace,
+ String source)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `source` | Source name
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`restartSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#restartSource-java.lang.String-java.lang.String-java.lang.String-).
+
+* Restart a **specified** source connector.
+
+ ```java
+
+ void restartSource(String tenant,
+ String namespace,
+ String source,
+ int instanceId)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `source` | Source name
+ `instanceId` | Source instanceID
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`restartSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#restartSource-java.lang.String-java.lang.String-java.lang.String-int-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Restart a sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `restart` subcommand.
+
+```
+
+$ pulsar-admin sinks restart options
+
+```
+
+For more information, see [here](reference-connector-admin.md#restart-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+* Restart **all** sink connectors.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sinkName/restart|operation/restartSource?version=@pulsar:version_number@}
+
+* Restart a **specified** sink connector.
+
+ Send a `POST` request to this endpoint: {@inject: endpoint|POST|/admin/v3/sources/:tenant/:namespace/:sinkName/:instanceId/restart|operation/restartSource?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+* Restart all Pulsar sink connectors.
+
+ ```java
+
+ void restartSink(String tenant,
+ String namespace,
+ String sink)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `sink` | Sink name
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`restartSink`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#restartSink-java.lang.String-java.lang.String-java.lang.String-).
+
+* Restart a **specified** sink connector.
+
+ ```java
+
+ void restartSink(String tenant,
+ String namespace,
+ String sink,
+ int instanceId)
+ throws PulsarAdminException
+
+ ```
+
+ **Parameter**
+
+ | Name | Description
+ |---|---
+ `tenant` | Tenant name
+ `namespace` | Namespace name
+ `source` | Source name
+ `instanceId` | Sink instanceID
+
+ **Exception**
+
+ |Name|Description|
+ |---|---
+ | `PulsarAdminException` | Unexpected error
+
+ For more information, see [`restartSink`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#restartSink-java.lang.String-java.lang.String-java.lang.String-int-).
+
+</TabItem>
+
+</Tabs>
+
+## Delete a connector
+
+### `delete`
+
+You can delete a connector using **Admin CLI**, **REST API** or **JAVA admin API**.
+
+#### Source
+
+Delete a source connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `delete` subcommand.
+
+```
+
+$ pulsar-admin sources delete options
+
+```
+
+For more information, see [here](reference-connector-admin.md#delete).
+
+</TabItem>
+<TabItem value="REST API">
+
+Delete al Pulsar source connector.
+
+Send a `DELETE` request to this endpoint: {@inject: endpoint|DELETE|/admin/v3/sources/:tenant/:namespace/:sourceName|operation/deregisterSource?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+Delete a source connector.
+
+```java
+
+void deleteSource(String tenant,
+ String namespace,
+ String source)
+ throws PulsarAdminException
+
+```
+
+**Parameter**
+
+| Name | Description
+|---|---
+`tenant` | Tenant name
+`namespace` | Namespace name
+`source` | Source name
+
+**Exception**
+
+|Name|Description|
+|---|---
+|`PulsarAdminException.NotAuthorizedException`| You don't have the admin permission
+| `PulsarAdminException.NotFoundException` | Cluster doesn't exist
+| `PulsarAdminException.PreconditionFailedException` | Cluster is not empty
+| `PulsarAdminException` | Unexpected error
+
+For more information, see [`deleteSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Source.html#deleteSource-java.lang.String-java.lang.String-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
+
+#### Sink
+
+Delete a sink connector.
+
+<Tabs
+ defaultValue="Admin CLI"
+ values={[{"label":"Admin CLI","value":"Admin CLI"},{"label":"REST API","value":"REST API"},{"label":"Java Admin API","value":"Java Admin API"}]}>
+
+<TabItem value="Admin CLI">
+
+Use the `delete` subcommand.
+
+```
+
+$ pulsar-admin sinks delete options
+
+```
+
+For more information, see [here](reference-connector-admin.md#delete-1).
+
+</TabItem>
+<TabItem value="REST API">
+
+Delete a sink connector.
+
+Send a `DELETE` request to this endpoint: {@inject: endpoint|DELETE|/admin/v3/sinks/:tenant/:namespace/:sinkName|operation/deregisterSink?version=@pulsar:version_number@}
+
+</TabItem>
+<TabItem value="Java Admin API">
+
+Delete a Pulsar sink connector.
+
+```java
+
+void deleteSink(String tenant,
+ String namespace,
+ String source)
+ throws PulsarAdminException
+
+```
+
+**Parameter**
+
+| Name | Description
+|---|---
+`tenant` | Tenant name
+`namespace` | Namespace name
+`sink` | Sink name
+
+**Exception**
+
+|Name|Description|
+|---|---
+|`PulsarAdminException.NotAuthorizedException`| You don't have the admin permission
+| `PulsarAdminException.NotFoundException` | Cluster doesn't exist
+| `PulsarAdminException.PreconditionFailedException` | Cluster is not empty
+| `PulsarAdminException` | Unexpected error
+
+For more information, see [`deleteSource`](https://pulsar.apache.org/api/admin/org/apache/pulsar/client/admin/Sink.html#deleteSink-java.lang.String-java.lang.String-java.lang.String-).
+
+</TabItem>
+
+</Tabs>
diff --git a/site2/website-next/versioned_docs/version-2.4.1/sql-deployment-configurations.md b/site2/website-next/versioned_docs/version-2.4.1/sql-deployment-configurations.md
new file mode 100644
index 0000000..6c6fd87
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/sql-deployment-configurations.md
@@ -0,0 +1,189 @@
+---
+id: sql-deployment-configurations
+title: Pulsar SQL configuration and deployment
+sidebar_label: "Configuration and deployment"
+original_id: sql-deployment-configurations
+---
+
+You can configure Presto Pulsar connector and deploy a cluster with the following instruction.
+
+## Configure Presto Pulsar Connector
+You can configure Presto Pulsar Connector in the `${project.root}/conf/presto/catalog/pulsar.properties` properties file. The configuration for the connector and the default values are as follows.
+
+```properties
+
+# name of the connector to be displayed in the catalog
+connector.name=pulsar
+
+# the url of Pulsar broker service
+pulsar.web-service-url=http://localhost:8080
+
+# URI of Zookeeper cluster
+pulsar.zookeeper-uri=localhost:2181
+
+# minimum number of entries to read at a single time
+pulsar.entry-read-batch-size=100
+
+# default number of splits to use per query
+pulsar.target-num-splits=4
+
+```
+
+You can connect Presto to a Pulsar cluster with multiple hosts. To configure multiple hosts for brokers, add multiple URLs to `pulsar.web-service-url`. To configure multiple hosts for ZooKeeper, add multiple URIs to `pulsar.zookeeper-uri`. The following is an example.
+
+```
+
+pulsar.web-service-url=http://localhost:8080,localhost:8081,localhost:8082
+pulsar.zookeeper-uri=localhost1,localhost2:2181
+
+```
+
+## Query data from existing Presto clusters
+
+If you already have a Presto cluster, you can copy the Presto Pulsar connector plugin to your existing cluster. Download the archived plugin package with the following command.
+
+```bash
+
+$ wget pulsar:binary_release_url
+
+```
+
+## Deploy a new cluster
+
+Since Pulsar SQL is powered by [Trino (formerly Presto SQL)](https://trino.io), the configuration for deployment is the same for the Pulsar SQL worker.
+
+:::note
+
+For how to set up a standalone single node environment, refer to [Query data](sql-getting-started).
+
+:::
+
+You can use the same CLI args as the Presto launcher.
+
+```bash
+
+$ ./bin/pulsar sql-worker --help
+Usage: launcher [options] command
+
+Commands: run, start, stop, restart, kill, status
+
+Options:
+ -h, --help show this help message and exit
+ -v, --verbose Run verbosely
+ --etc-dir=DIR Defaults to INSTALL_PATH/etc
+ --launcher-config=FILE
+ Defaults to INSTALL_PATH/bin/launcher.properties
+ --node-config=FILE Defaults to ETC_DIR/node.properties
+ --jvm-config=FILE Defaults to ETC_DIR/jvm.config
+ --config=FILE Defaults to ETC_DIR/config.properties
+ --log-levels-file=FILE
+ Defaults to ETC_DIR/log.properties
+ --data-dir=DIR Defaults to INSTALL_PATH
+ --pid-file=FILE Defaults to DATA_DIR/var/run/launcher.pid
+ --launcher-log-file=FILE
+ Defaults to DATA_DIR/var/log/launcher.log (only in
+ daemon mode)
+ --server-log-file=FILE
+ Defaults to DATA_DIR/var/log/server.log (only in
+ daemon mode)
+ -D NAME=VALUE Set a Java system property
+
+```
+
+The default configuration for the cluster is located in `${project.root}/conf/presto`. You can customize your deployment by modifying the default configuration.
+
+You can set the worker to read from a different configuration directory, or set a different directory to write data.
+
+```bash
+
+$ ./bin/pulsar sql-worker run --etc-dir /tmp/incubator-pulsar/conf/presto --data-dir /tmp/presto-1
+
+```
+
+You can start the worker as daemon process.
+
+```bash
+
+$ ./bin/pulsar sql-worker start
+
+```
+
+### Deploy a cluster on multiple nodes
+
+You can deploy a Pulsar SQL cluster or Presto cluster on multiple nodes. The following example shows how to deploy a cluster on three-node cluster.
+
+1. Copy the Pulsar binary distribution to three nodes.
+
+The first node runs as Presto coordinator. The minimal configuration requirement in the `${project.root}/conf/presto/config.properties` file is as follows.
+
+```properties
+
+coordinator=true
+node-scheduler.include-coordinator=true
+http-server.http.port=8080
+query.max-memory=50GB
+query.max-memory-per-node=1GB
+discovery-server.enabled=true
+discovery.uri=<coordinator-url>
+
+```
+
+The other two nodes serve as worker nodes, you can use the following configuration for worker nodes.
+
+```properties
+
+coordinator=false
+http-server.http.port=8080
+query.max-memory=50GB
+query.max-memory-per-node=1GB
+discovery.uri=<coordinator-url>
+
+```
+
+2. Modify `pulsar.web-service-url` and `pulsar.zookeeper-uri` configuration in the `${project.root}/conf/presto/catalog/pulsar.properties` file accordingly for the three nodes.
+
+3. Start the coordinator node.
+
+```
+
+$ ./bin/pulsar sql-worker run
+
+```
+
+4. Start worker nodes.
+
+```
+
+$ ./bin/pulsar sql-worker run
+
+```
+
+5. Start the SQL CLI and check the status of your cluster.
+
+```bash
+
+$ ./bin/pulsar sql --server <coordinate_url>
+
+```
+
+6. Check the status of your nodes.
+
+```bash
+
+presto> SELECT * FROM system.runtime.nodes;
+ node_id | http_uri | node_version | coordinator | state
+---------+-------------------------+--------------+-------------+--------
+ 1 | http://192.168.2.1:8081 | testversion | true | active
+ 3 | http://192.168.2.2:8081 | testversion | false | active
+ 2 | http://192.168.2.3:8081 | testversion | false | active
+
+```
+
+For more information about deployment in Presto, refer to [Presto deployment](https://trino.io/docs/current/installation/deployment.html).
+
+:::note
+
+The broker does not advance LAC, so when Pulsar SQL bypass broker to query data, it can only read entries up to the LAC that all the bookies learned. You can enable periodically write LAC on the broker by setting "bookkeeperExplicitLacIntervalInMills" in the broker.conf.
+
+:::
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/sql-getting-started.md b/site2/website-next/versioned_docs/version-2.4.1/sql-getting-started.md
new file mode 100644
index 0000000..8a5cd71
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/sql-getting-started.md
@@ -0,0 +1,187 @@
+---
+id: sql-getting-started
+title: Query data with Pulsar SQL
+sidebar_label: "Query data"
+original_id: sql-getting-started
+---
+
+Before querying data in Pulsar, you need to install Pulsar and built-in connectors.
+
+## Requirements
+1. Install [Pulsar](getting-started-standalone.md#install-pulsar-standalone).
+2. Install Pulsar [built-in connectors](getting-started-standalone.md#install-builtin-connectors-optional).
+
+## Query data in Pulsar
+To query data in Pulsar with Pulsar SQL, complete the following steps.
+
+1. Start a Pulsar standalone cluster.
+
+```bash
+
+./bin/pulsar standalone
+
+```
+
+2. Start a Pulsar SQL worker.
+
+```bash
+
+./bin/pulsar sql-worker run
+
+```
+
+3. After initializing Pulsar standalone cluster and the SQL worker, run SQL CLI.
+
+```bash
+
+./bin/pulsar sql
+
+```
+
+4. Test with SQL commands.
+
+```bash
+
+presto> show catalogs;
+ Catalog
+---------
+ pulsar
+ system
+(2 rows)
+
+Query 20180829_211752_00004_7qpwh, FINISHED, 1 node
+Splits: 19 total, 19 done (100.00%)
+0:00 [0 rows, 0B] [0 rows/s, 0B/s]
+
+
+presto> show schemas in pulsar;
+ Schema
+-----------------------
+ information_schema
+ public/default
+ public/functions
+ sample/standalone/ns1
+(4 rows)
+
+Query 20180829_211818_00005_7qpwh, FINISHED, 1 node
+Splits: 19 total, 19 done (100.00%)
+0:00 [4 rows, 89B] [21 rows/s, 471B/s]
+
+
+presto> show tables in pulsar."public/default";
+ Table
+-------
+(0 rows)
+
+Query 20180829_211839_00006_7qpwh, FINISHED, 1 node
+Splits: 19 total, 19 done (100.00%)
+0:00 [0 rows, 0B] [0 rows/s, 0B/s]
+
+```
+
+Since there is no data in Pulsar, no records is returned.
+
+5. Start the built-in connector _DataGeneratorSource_ and ingest some mock data.
+
+```bash
+
+./bin/pulsar-admin sources create --name generator --destinationTopicName generator_test --source-type data-generator
+
+```
+
+And then you can query a topic in the namespace "public/default".
+
+```bash
+
+presto> show tables in pulsar."public/default";
+ Table
+----------------
+ generator_test
+(1 row)
+
+Query 20180829_213202_00000_csyeu, FINISHED, 1 node
+Splits: 19 total, 19 done (100.00%)
+0:02 [1 rows, 38B] [0 rows/s, 17B/s]
+
+```
+
+You can now query the data within the topic "generator_test".
+
+```bash
+
+presto> select * from pulsar."public/default".generator_test;
+
+ firstname | middlename | lastname | email | username | password | telephonenumber | age | companyemail | nationalidentitycardnumber |
+-------------+-------------+-------------+----------------------------------+--------------+----------+-----------------+-----+-----------------------------------------------+----------------------------+
+ Genesis | Katherine | Wiley | genesis.wiley@gmail.com | genesisw | y9D2dtU3 | 959-197-1860 | 71 | genesis.wiley@interdemconsulting.eu | 880-58-9247 |
+ Brayden | | Stanton | brayden.stanton@yahoo.com | braydens | ZnjmhXik | 220-027-867 | 81 | brayden.stanton@supermemo.eu | 604-60-7069 |
+ Benjamin | Julian | Velasquez | benjamin.velasquez@yahoo.com | benjaminv | 8Bc7m3eb | 298-377-0062 | 21 | benjamin.velasquez@hostesltd.biz | 213-32-5882 |
+ Michael | Thomas | Donovan | donovan@mail.com | michaeld | OqBm9MLs | 078-134-4685 | 55 | michael.donovan@memortech.eu | 443-30-3442 |
+ Brooklyn | Avery | Roach | brooklynroach@yahoo.com | broach | IxtBLafO | 387-786-2998 | 68 | brooklyn.roach@warst.biz | 085-88-3973 |
+ Skylar | | Bradshaw | skylarbradshaw@yahoo.com | skylarb | p6eC6cKy | 210-872-608 | 96 | skylar.bradshaw@flyhigh.eu | 453-46-0334 |
+.
+.
+.
+
+```
+
+You can query the mock data.
+
+## Query your own data
+If you want to query your own data, you need to ingest your own data first. You can write a simple producer and write custom defined data to Pulsar. The following is an example.
+
+```java
+
+public class TestProducer {
+
+ public static class Foo {
+ private int field1 = 1;
+ private String field2;
+ private long field3;
+
+ public Foo() {
+ }
+
+ public int getField1() {
+ return field1;
+ }
+
+ public void setField1(int field1) {
+ this.field1 = field1;
+ }
+
+ public String getField2() {
+ return field2;
+ }
+
+ public void setField2(String field2) {
+ this.field2 = field2;
+ }
+
+ public long getField3() {
+ return field3;
+ }
+
+ public void setField3(long field3) {
+ this.field3 = field3;
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
+ Producer<Foo> producer = pulsarClient.newProducer(AvroSchema.of(Foo.class)).topic("test_topic").create();
+
+ for (int i = 0; i < 1000; i++) {
+ Foo foo = new Foo();
+ foo.setField1(i);
+ foo.setField2("foo" + i);
+ foo.setField3(System.currentTimeMillis());
+ producer.newMessage().value(foo).send();
+ }
+ producer.close();
+ pulsarClient.close();
+ }
+}
+
+```
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/sql-overview.md b/site2/website-next/versioned_docs/version-2.4.1/sql-overview.md
new file mode 100644
index 0000000..4a4d5f0
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/sql-overview.md
@@ -0,0 +1,18 @@
+---
+id: sql-overview
+title: Pulsar SQL Overview
+sidebar_label: "Overview"
+original_id: sql-overview
+---
+
+Apache Pulsar is used to store streams of event data, and the event data is structured with predefined fields. With the implementation of the [Schema Registry](schema-get-started), you can store structured data in Pulsar and query the data by using [Trino (formerly Presto SQL)](https://trino.io/).
+
+As the core of Pulsar SQL, Presto Pulsar connector enables Presto workers within a Presto cluster to query data from Pulsar.
+
+
+
+The query performance is efficient and highly scalable, because Pulsar adopts [two level segment based architecture](concepts-architecture-overview.md#apache-bookkeeper).
+
+Topics in Pulsar are stored as segments in [Apache BookKeeper](https://bookkeeper.apache.org/). Each topic segment is replicated to some BookKeeper nodes, which enables concurrent reads and high read throughput. You can configure the number of BookKeeper nodes, and the default number is `3`. In Presto Pulsar connector, data is read directly from BookKeeper, so Presto workers can read concurrently from horizontally scalable number BookKeeper nodes.
+
+
diff --git a/site2/website-next/versioned_docs/version-2.4.1/window-functions-context.md b/site2/website-next/versioned_docs/version-2.4.1/window-functions-context.md
new file mode 100644
index 0000000..f80fea5
--- /dev/null
+++ b/site2/website-next/versioned_docs/version-2.4.1/window-functions-context.md
@@ -0,0 +1,581 @@
+---
+id: window-functions-context
+title: Window Functions Context
+sidebar_label: "Window Functions: Context"
+original_id: window-functions-context
+---
+
+Java SDK provides access to a **window context object** that can be used by a window function. This context object provides a wide variety of information and functionality for Pulsar window functions as below.
+
+- [Spec](#spec)
+
+ * Names of all input topics and the output topic associated with the function.
+ * Tenant and namespace associated with the function.
+ * Pulsar window function name, ID, and version.
+ * ID of the Pulsar function instance running the window function.
+ * Number of instances that invoke the window function.
+ * Built-in type or custom class name of the output schema.
+
+- [Logger](#logger)
+
+ * Logger object used by the window function, which can be used to create window function log messages.
+
+- [User config](#user-config)
+
+ * Access to arbitrary user configuration values.
+
+- [Routing](#routing)
+
+ * Routing is supported in Pulsar window functions. Pulsar window functions send messages to arbitrary topics as per the `publish` interface.
+
+- [Metrics](#metrics)
+
+ * Interface for recording metrics.
+
+- [State storage](#state-storage)
+
+ * Interface for storing and retrieving state in [state storage](#state-storage).
+
+## Spec
+
+Spec contains the basic information of a function.
+
+### Get input topics
+
+The `getInputTopics` method gets the **name list** of all input topics.
+
+This example demonstrates how to get the name list of all input topics in a Java window function.
+
+```java
+
+public class GetInputTopicsWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ Collection<String> inputTopics = context.getInputTopics();
+ System.out.println(inputTopics);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get output topic
+
+The `getOutputTopic` method gets the **name of a topic** to which the message is sent.
+
+This example demonstrates how to get the name of an output topic in a Java window function.
+
+```java
+
+public class GetOutputTopicWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String outputTopic = context.getOutputTopic();
+ System.out.println(outputTopic);
+
+ return null;
+ }
+}
+
+```
+
+### Get tenant
+
+The `getTenant` method gets the tenant name associated with the window function.
+
+This example demonstrates how to get the tenant name in a Java window function.
+
+```java
+
+public class GetTenantWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String tenant = context.getTenant();
+ System.out.println(tenant);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get namespace
+
+The `getNamespace` method gets the namespace associated with the window function.
+
+This example demonstrates how to get the namespace in a Java window function.
+
+```java
+
+public class GetNamespaceWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String ns = context.getNamespace();
+ System.out.println(ns);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get function name
+
+The `getFunctionName` method gets the window function name.
+
+This example demonstrates how to get the function name in a Java window function.
+
+```java
+
+public class GetNameOfWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String functionName = context.getFunctionName();
+ System.out.println(functionName);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get function ID
+
+The `getFunctionId` method gets the window function ID.
+
+This example demonstrates how to get the function ID in a Java window function.
+
+```java
+
+public class GetFunctionIDWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String functionID = context.getFunctionId();
+ System.out.println(functionID);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get function version
+
+The `getFunctionVersion` method gets the window function version.
+
+This example demonstrates how to get the function version of a Java window function.
+
+```java
+
+public class GetVersionOfWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String functionVersion = context.getFunctionVersion();
+ System.out.println(functionVersion);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get instance ID
+
+The `getInstanceId` method gets the instance ID of a window function.
+
+This example demonstrates how to get the instance ID in a Java window function.
+
+```java
+
+public class GetInstanceIDWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ int instanceId = context.getInstanceId();
+ System.out.println(instanceId);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get num instances
+
+The `getNumInstances` method gets the number of instances that invoke the window function.
+
+This example demonstrates how to get the number of instances in a Java window function.
+
+```java
+
+public class GetNumInstancesWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ int numInstances = context.getNumInstances();
+ System.out.println(numInstances);
+
+ return null;
+ }
+
+}
+
+```
+
+### Get output schema type
+
+The `getOutputSchemaType` method gets the built-in type or custom class name of the output schema.
+
+This example demonstrates how to get the output schema type of a Java window function.
+
+```java
+
+public class GetOutputSchemaTypeWindowFunction implements WindowFunction<String, Void> {
+
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ String schemaType = context.getOutputSchemaType();
+ System.out.println(schemaType);
+
+ return null;
+ }
+}
+
+```
+
+## Logger
+
+Pulsar window functions using Java SDK has access to an [SLF4j](https://www.slf4j.org/) [`Logger`](https://www.slf4j.org/api/org/apache/log4j/Logger.html) object that can be used to produce logs at the chosen log level.
+
+This example logs either a `WARNING`-level or `INFO`-level log based on whether the incoming string contains the word `danger` or not in a Java function.
+
+```java
+
+import java.util.Collection;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.functions.api.WindowContext;
+import org.apache.pulsar.functions.api.WindowFunction;
+import org.slf4j.Logger;
+
+public class LoggingWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ Logger log = context.getLogger();
+ for (Record<String> record : inputs) {
+ log.info(record + "-window-log");
+ }
+ return null;
+ }
+
+}
+
+```
+
+If you need your function to produce logs, specify a log topic when creating or running the function.
+
+```bash
+
+bin/pulsar-admin functions create \
+ --jar my-functions.jar \
+ --classname my.package.LoggingFunction \
+ --log-topic persistent://public/default/logging-function-logs \
+ # Other function configs
+
+```
+
+You can access all logs produced by `LoggingFunction` via the `persistent://public/default/logging-function-logs` topic.
+
+## Metrics
+
+Pulsar window functions can publish arbitrary metrics to the metrics interface which can be queried.
+
+:::note
+
+If a Pulsar window function uses the language-native interface for Java, that function is not able to publish metrics and stats to Pulsar.
+
+:::
+
+You can record metrics using the context object on a per-key basis.
+
+This example sets a metric for the `process-count` key and a different metric for the `elevens-count` key every time the function processes a message in a Java function.
+
+```java
+
+import java.util.Collection;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.functions.api.WindowContext;
+import org.apache.pulsar.functions.api.WindowFunction;
+
+
+/**
+ * Example function that wants to keep track of
+ * the event time of each message sent.
+ */
+public class UserMetricWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+
+ for (Record<String> record : inputs) {
+ if (record.getEventTime().isPresent()) {
+ context.recordMetric("MessageEventTime", record.getEventTime().get().doubleValue());
+ }
+ }
+
+ return null;
+ }
+}
+
+```
+
+## User config
+
+When you run or update Pulsar Functions that are created using SDK, you can pass arbitrary key/value pairs to them with the `--user-config` flag. Key/value pairs **must** be specified as JSON.
+
+This example passes a user configured key/value to a function.
+
+```bash
+
+bin/pulsar-admin functions create \
+ --name word-filter \
+ --user-config '{"forbidden-word":"rosebud"}' \
+ # Other function configs
+
+```
+
+### API
+You can use the following APIs to get user-defined information for window functions.
+#### getUserConfigMap
+
+`getUserConfigMap` API gets a map of all user-defined key/value configurations for the window function.
+
+```java
+
+/**
+ * Get a map of all user-defined key/value configs for the function.
+ *
+ * @return The full map of user-defined config values
+ */
+ Map<String, Object> getUserConfigMap();
+
+```
+
+#### getUserConfigValue
+
+The `getUserConfigValue` API gets a user-defined key/value.
+
+```java
+
+/**
+ * Get any user-defined key/value.
+ *
+ * @param key The key
+ * @return The Optional value specified by the user for that key.
+ */
+ Optional<Object> getUserConfigValue(String key);
+
+```
+
+#### getUserConfigValueOrDefault
+
+The `getUserConfigValueOrDefault` API gets a user-defined key/value or a default value if none is present.
+
+```java
+
+/**
+ * Get any user-defined key/value or a default value if none is present.
+ *
+ * @param key
+ * @param defaultValue
+ * @return Either the user config value associated with a given key or a supplied default value
+ */
+ Object getUserConfigValueOrDefault(String key, Object defaultValue);
+
+```
+
+This example demonstrates how to access key/value pairs provided to Pulsar window functions.
+
+Java SDK context object enables you to access key/value pairs provided to Pulsar window functions via the command line (as JSON).
+
+:::tip
+
+For all key/value pairs passed to Java window functions, both the `key` and the `value` are `String`. To set the value to be a different type, you need to deserialize it from the `String` type.
+
+:::
+
+This example passes a key/value pair in a Java window function.
+
+```bash
+
+bin/pulsar-admin functions create \
+ --user-config '{"word-of-the-day":"verdure"}' \
+ # Other function configs
+
+```
+
+This example accesses values in a Java window function.
+
+The `UserConfigFunction` function logs the string `"The word of the day is verdure"` every time the function is invoked (which means every time a message arrives). The user config of `word-of-the-day` is changed **only** when the function is updated with a new config value via
+multiple ways, such as the command line tool or REST API.
+
+```java
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+import org.slf4j.Logger;
+
+import java.util.Optional;
+
+public class UserConfigWindowFunction implements WindowFunction<String, String> {
+ @Override
+ public String process(Collection<Record<String>> input, WindowContext context) throws Exception {
+ Optional<Object> whatToWrite = context.getUserConfigValue("WhatToWrite");
+ if (whatToWrite.get() != null) {
+ return (String)whatToWrite.get();
+ } else {
+ return "Not a nice way";
+ }
+ }
+
+}
+
+```
+
+If no value is provided, you can access the entire user config map or set a default value.
+
+```java
+
+// Get the whole config map
+Map<String, String> allConfigs = context.getUserConfigMap();
+
+// Get value or resort to default
+String wotd = context.getUserConfigValueOrDefault("word-of-the-day", "perspicacious");
+
+```
+
+## Routing
+
+You can use the `context.publish()` interface to publish as many results as you want.
+
+This example shows that the `PublishFunction` class uses the built-in function in the context to publish messages to the `publishTopic` in a Java function.
+
+```java
+
+public class PublishWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> input, WindowContext context) throws Exception {
+ String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic");
+ String output = String.format("%s!", input);
+ context.publish(publishTopic, output);
+
+ return null;
+ }
+
+}
+
+```
+
+## State storage
+
+Pulsar window functions use [Apache BookKeeper](https://bookkeeper.apache.org) as a state storage interface. Apache Pulsar installation (including the standalone installation) includes the deployment of BookKeeper bookies.
+
+Apache Pulsar integrates with Apache BookKeeper `table service` to store the `state` for functions. For example, the `WordCount` function can store its `counters` state into BookKeeper table service via Pulsar Functions state APIs.
+
+States are key-value pairs, where the key is a string and the value is arbitrary binary data—counters are stored as 64-bit big-endian binary values. Keys are scoped to an individual Pulsar Function and shared between instances of that function.
+
+Currently, Pulsar window functions expose Java API to access, update, and manage states. These APIs are available in the context object when you use Java SDK functions.
+
+| Java API| Description
+|---|---
+|`incrCounter`|Increases a built-in distributed counter referred by key.
+|`getCounter`|Gets the counter value for the key.
+|`putState`|Updates the state value for the key.
+
+You can use the following APIs to access, update, and manage states in Java window functions.
+
+#### incrCounter
+
+The `incrCounter` API increases a built-in distributed counter referred by key.
+
+Applications use the `incrCounter` API to change the counter of a given `key` by the given `amount`. If the `key` does not exist, a new key is created.
+
+```java
+
+ /**
+ * Increment the builtin distributed counter referred by key
+ * @param key The name of the key
+ * @param amount The amount to be incremented
+ */
+ void incrCounter(String key, long amount);
+
+```
+
+#### getCounter
+
+The `getCounter` API gets the counter value for the key.
+
+Applications uses the `getCounter` API to retrieve the counter of a given `key` changed by the `incrCounter` API.
+
+```java
+
+ /**
+ * Retrieve the counter value for the key.
+ *
+ * @param key name of the key
+ * @return the amount of the counter value for this key
+ */
+ long getCounter(String key);
+
+```
+
+Except the `getCounter` API, Pulsar also exposes a general key/value API (`putState`) for functions to store general key/value state.
+
+#### putState
+
+The `putState` API updates the state value for the key.
+
+```java
+
+ /**
+ * Update the state value for the key.
+ *
+ * @param key name of the key
+ * @param value state value of the key
+ */
+ void putState(String key, ByteBuffer value);
+
+```
+
+This example demonstrates how applications store states in Pulsar window functions.
+
+The logic of the `WordCountWindowFunction` is simple and straightforward.
+
+1. The function first splits the received string into multiple words using regex `\\.`.
+
+2. For each `word`, the function increments the corresponding `counter` by 1 via `incrCounter(key, amount)`.
+
+```java
+
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+
+import java.util.Arrays;
+
+public class WordCountWindowFunction implements WindowFunction<String, Void> {
+ @Override
+ public Void process(Collection<Record<String>> inputs, WindowContext context) throws Exception {
+ for (Record<String> input : inputs) {
+ Arrays.asList(input.getValue().split("\\.")).forEach(word -> context.incrCounter(word, 1));
+ }
+ return null;
+
+ }
+}
+
+```
+
diff --git a/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json b/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json
index 7b68978..3a21053 100644
--- a/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json
+++ b/site2/website-next/versioned_sidebars/version-2.4.1-sidebars.json
@@ -89,6 +89,100 @@
"id": "version-2.4.1/schema-manage"
}
]
+ },
+ {
+ "type": "category",
+ "label": "Pulsar Functions",
+ "items": [
+ {
+ "type": "doc",
+ "id": "version-2.4.1/functions-overview"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/functions-worker"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/functions-runtime"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/functions-develop"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/functions-debug"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/functions-deploy"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/functions-cli"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/window-functions-context"
+ }
+ ]
+ },
+ {
+ "type": "category",
+ "label": "Pulsar IO",
+ "items": [
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-overview"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-quickstart"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-use"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-managing"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-debug"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-connectors"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-develop"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/io-cdc"
+ }
+ ]
+ },
+ {
+ "type": "category",
+ "label": "Pulsar SQL",
+ "items": [
+ {
+ "type": "doc",
+ "id": "version-2.4.1/sql-overview"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/sql-getting-started"
+ },
+ {
+ "type": "doc",
+ "id": "version-2.4.1/sql-deployment-configurations"
+ }
+ ]
}
]
}
\ No newline at end of file