[Doc] Add descriptions for function runtime and context

site2/docs/functions-develop.md

@@ -275,6 +275,7 @@ Java, Python and Go SDKs provide access to a **context object** that can be used
 * 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).
+* (Java) get Pulsar admin client.
 
 <!--DOCUSAURUS_CODE_TABS-->
 <!--Java-->
@@ -310,6 +311,8 @@ public interface Context {
     <O> CompletableFuture<Void> publish(String topicName, O object);
     <O> TypedMessageBuilder<O> newOutputMessage(String topicName, Schema<O> schema) throws PulsarClientException;
     <O> ConsumerBuilder<O> newConsumerBuilder(Schema<O> schema) throws PulsarClientException;
+    PulsarAdmin getPulsarAdmin();
+    PulsarAdmin getPulsarAdmin(String clusterName);
 }
 ```
 
@@ -710,6 +713,57 @@ When you use `logTopic` related functionalities in Go Function, import `github.c
 
 <!--END_DOCUSAURUS_CODE_TABS-->
 
+### Pulsar admin
+
+Pulsar Functions using the Java SDK has access to the Pulsar admin client, which allows the Pulsar admin client to manage API calls to current Pulsar clusters or external clusters (if `external-pulsars` is provided).
+
+<!--DOCUSAURUS_CODE_TABS-->
+<!--Java-->
+
+Below is an example of how to use the Pulsar admin client exposed from the Function `context`.
+
+```
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+
+/**
+ * In this particular example, for every input message,
+ * the function resets the cursor of the current function's subscription to a
+ * specified timestamp.
+ */
+public class CursorManagementFunction implements Function<String, String> {
+
+    @Override
+    public String process(String input, Context context) throws Exception {
+        PulsarAdmin adminClient = context.getPulsarAdmin();
+        if (adminClient != null) {
+            String topic = context.getCurrentRecord().getTopicName().isPresent() ?
+                    context.getCurrentRecord().getTopicName().get() : null;
+            String subName = context.getTenant() + "/" + context.getNamespace() + "/" + context.getFunctionName();
+            if (topic != null) {
+                // 1578188166 below is a random-pick timestamp
+                adminClient.topics().resetCursor(topic, subName, 1578188166);
+                return "reset cursor successfully";
+            }
+        }
+        return null;
+    }
+}
+```
+
+If you want your function to get access to the Pulsar admin client, you need to enable this feature by setting `exposeAdminClientEnabled=true` in the `functions_worker.yml` file. You can test whether this feature is enabled or not using the command `pulsar-admin functions localrun` with the flag `--web-service-url`.
+
+```
+$ bin/pulsar-admin functions localrun \
+ --jar my-functions.jar \
+ --classname my.package.CursorManagementFunction \
+ --web-service-url http://pulsar-web-service:8080 \
+ # Other function configs
+```
+
+<!--END_DOCUSAURUS_CODE_TABS-->
+
 ## Metrics
 Pulsar Functions can publish arbitrary metrics to the metrics interface which can be queried. 
 

site2/docs/functions-runtime.md

@@ -278,7 +278,11 @@ The Kubernetes integration enables you to implement a class and customize how to
 
 The functions (and sinks/sources) API provides a flag, `customRuntimeOptions`, which is passed to this interface.
 
-Pulsar includes a built-in implementation. To use the basic implementation, set `runtimeCustomizerClassName` to `org.apache.pulsar.functions.runtime.kubernetes.BasicKubernetesManifestCustomizer`. The built-in implementation enables you to pass a JSON document with certain properties to augment how the manifests are generated. The following is an example.
+To initialize the `KubernetesManifestCustomizer`, you can provide `runtimeCustomizerConfig` in the `functions-worker.yml` file. `runtimeCustomizerConfig` is passed to the `public void initialize(Map<String, Object> config)` function of the interface. `runtimeCustomizerConfig`is different from the `customRuntimeOptions` as `runtimeCustomizerConfig` is the same across all functions. If you provide both `runtimeCustomizerConfig`  and `customRuntimeOptions`, you need to decide how to manage these two configurations in your implementation of `KubernetesManifestCustomizer`. 
+
+Pulsar includes a built-in implementation. To use the basic implementation, set `runtimeCustomizerClassName` to `org.apache.pulsar.functions.runtime.kubernetes.BasicKubernetesManifestCustomizer`. The built-in implementation initialized with `runtimeCustomizerConfig` enables you to pass a JSON document as `customRuntimeOptions` with certain properties to augment, which decides how the manifests are generated. If both `runtimeCustomizerConfig` and `customRuntimeOptions` are provided, `BasicKubernetesManifestCustomizer` uses `customRuntimeOptions` to override the configuration if there are conflicts in these two configurations. 
+
+Below is an example of `customRuntimeOptions`.
 
 ```Json
 {

site2/website/versioned_docs/version-2.7.1/functions-develop.md

@@ -276,6 +276,7 @@ Java, Python and Go SDKs provide access to a **context object** that can be used
 * 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).
+* (Java) get Pulsar admin client.
 
 <!--DOCUSAURUS_CODE_TABS-->
 <!--Java-->
@@ -311,6 +312,8 @@ public interface Context {
     <O> CompletableFuture<Void> publish(String topicName, O object);
     <O> TypedMessageBuilder<O> newOutputMessage(String topicName, Schema<O> schema) throws PulsarClientException;
     <O> ConsumerBuilder<O> newConsumerBuilder(Schema<O> schema) throws PulsarClientException;
+    PulsarAdmin getPulsarAdmin();
+    PulsarAdmin getPulsarAdmin(String clusterName);
 }
 ```
 
@@ -711,6 +714,55 @@ When you use `logTopic` related functionalities in Go Function, import `github.c
 
 <!--END_DOCUSAURUS_CODE_TABS-->
 
+### Pulsar admin
+
+Pulsar Functions using the Java SDK has access to the Pulsar admin client, which allows the Pulsar admin client to manage API calls to current Pulsar clusters or external clusters (if `external-pulsars` is provided).
+
+<!--DOCUSAURUS_CODE_TABS-->
+<!--Java-->
+
+Below is an example of how to use the Pulsar admin client exposed from the Function `context`.
+
+```
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+/**
+ * In this particular example, for every input message,
+ * the function resets the cursor of the current function's subscription to a
+ * specified timestamp.
+ */
+public class CursorManagementFunction implements Function<String, String> {
+    @Override
+    public String process(String input, Context context) throws Exception {
+        PulsarAdmin adminClient = context.getPulsarAdmin();
+        if (adminClient != null) {
+            String topic = context.getCurrentRecord().getTopicName().isPresent() ?
+                    context.getCurrentRecord().getTopicName().get() : null;
+            String subName = context.getTenant() + "/" + context.getNamespace() + "/" + context.getFunctionName();
+            if (topic != null) {
+                // 1578188166 below is a random-pick timestamp
+                adminClient.topics().resetCursor(topic, subName, 1578188166);
+                return "reset cursor successfully";
+            }
+        }
+        return null;
+    }
+}
+```
+
+If you want your function to get access to the Pulsar admin client, you need to enable this feature by setting `exposeAdminClientEnabled=true` in the `functions_worker.yml` file. You can test whether this feature is enabled or not using the command `pulsar-admin functions localrun` with the flag `--web-service-url`.
+
+```
+$ bin/pulsar-admin functions localrun \
+ --jar my-functions.jar \
+ --classname my.package.CursorManagementFunction \
+ --web-service-url http://pulsar-web-service:8080 \
+ # Other function configs
+```
+
+<!--END_DOCUSAURUS_CODE_TABS-->
+
 ## Metrics
 Pulsar Functions can publish arbitrary metrics to the metrics interface which can be queried. 
 

site2/website/versioned_docs/version-2.7.1/functions-runtime.md

@@ -242,7 +242,11 @@ The Kubernetes integration enables you to implement a class and customize how to
 
 The functions (and sinks/sources) API provides a flag, `customRuntimeOptions`, which is passed to this interface.
 
-Pulsar includes a built-in implementation. To use the basic implementation, set `runtimeCustomizerClassName` to `org.apache.pulsar.functions.runtime.kubernetes.BasicKubernetesManifestCustomizer`. The built-in implementation enables you to pass a JSON document with certain properties to augment how the manifests are generated. The following is an example.
+To initialize the `KubernetesManifestCustomizer`, you can provide `runtimeCustomizerConfig` in the `functions-worker.yml` file. `runtimeCustomizerConfig` is passed to the `public void initialize(Map<String, Object> config)` function of the interface. `runtimeCustomizerConfig`is different from the `customRuntimeOptions` as `runtimeCustomizerConfig` is the same across all functions. If you provide both `runtimeCustomizerConfig`  and `customRuntimeOptions`, you need to decide how to manage these two configurations in your implementation of `KubernetesManifestCustomizer`. 
+
+Pulsar includes a built-in implementation. To use the basic implementation, set `runtimeCustomizerClassName` to `org.apache.pulsar.functions.runtime.kubernetes.BasicKubernetesManifestCustomizer`. The built-in implementation initialized with `runtimeCustomizerConfig` enables you to pass a JSON document as `customRuntimeOptions` with certain properties to augment, which decides how the manifests are generated. If both `runtimeCustomizerConfig` and `customRuntimeOptions` are provided, `BasicKubernetesManifestCustomizer` uses `customRuntimeOptions` to override the configuration if there are conflicts in these two configurations. 
+
+Below is an example of `customRuntimeOptions`.
 
 ```Json
 {