From e2768440a0ab85c20e3729d60e1a2f02bf714082 Mon Sep 17 00:00:00 2001 From: Anonymitaet Date: Sun, 7 Feb 2021 17:01:05 +0800 Subject: [PATCH 1/3] add --- site2/docs/functions-develop.md | 54 +++++++++++++++++++++++++++++++++ site2/docs/functions-runtime.md | 6 +++- 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/site2/docs/functions-develop.md b/site2/docs/functions-develop.md index de059b8320bef..a2f99af84ac9e 100644 --- a/site2/docs/functions-develop.md +++ b/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. @@ -310,6 +311,8 @@ public interface Context { CompletableFuture publish(String topicName, O object); TypedMessageBuilder newOutputMessage(String topicName, Schema schema) throws PulsarClientException; ConsumerBuilder newConsumerBuilder(Schema 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 +### 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). + + + + +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 { + + @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 +``` + + + ## Metrics Pulsar Functions can publish arbitrary metrics to the metrics interface which can be queried. diff --git a/site2/docs/functions-runtime.md b/site2/docs/functions-runtime.md index 4df34dbdb4f24..63ffcb0640432 100644 --- a/site2/docs/functions-runtime.md +++ b/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 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` defined in the `functions-worker.yml` file enables you pass a JSON document as `customRuntimeOptions` with certain properties to augment to decide 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 { From 5be9aeb7323f2a37fd821a542e7064d6f876cf02 Mon Sep 17 00:00:00 2001 From: Anonymitaet Date: Mon, 8 Feb 2021 12:36:57 +0800 Subject: [PATCH 2/3] update --- site2/docs/functions-runtime.md | 2 +- .../version-2.7.1/functions-develop.md | 52 +++++++++++++++++++ .../version-2.7.1/functions-runtime.md | 6 ++- 3 files changed, 58 insertions(+), 2 deletions(-) diff --git a/site2/docs/functions-runtime.md b/site2/docs/functions-runtime.md index 63ffcb0640432..9b7deb745461f 100644 --- a/site2/docs/functions-runtime.md +++ b/site2/docs/functions-runtime.md @@ -280,7 +280,7 @@ The functions (and sinks/sources) API provides a flag, `customRuntimeOptions`, w To initialize the `KubernetesManifestCustomizer`, you can provide `runtimeCustomizerConfig` in the `functions-worker.yml` file. `runtimeCustomizerConfig` is passed to the `public void initialize(Map 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` defined in the `functions-worker.yml` file enables you pass a JSON document as `customRuntimeOptions` with certain properties to augment to decide 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. +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`. diff --git a/site2/website/versioned_docs/version-2.7.1/functions-develop.md b/site2/website/versioned_docs/version-2.7.1/functions-develop.md index 565f1aae87676..36e0758bf8cf4 100644 --- a/site2/website/versioned_docs/version-2.7.1/functions-develop.md +++ b/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. @@ -311,6 +312,8 @@ public interface Context { CompletableFuture publish(String topicName, O object); TypedMessageBuilder newOutputMessage(String topicName, Schema schema) throws PulsarClientException; ConsumerBuilder newConsumerBuilder(Schema 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 +### 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). + + + + +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 { + @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 +``` + + + ## Metrics Pulsar Functions can publish arbitrary metrics to the metrics interface which can be queried. diff --git a/site2/website/versioned_docs/version-2.7.1/functions-runtime.md b/site2/website/versioned_docs/version-2.7.1/functions-runtime.md index 6588c14f158d2..9a933122efbf8 100644 --- a/site2/website/versioned_docs/version-2.7.1/functions-runtime.md +++ b/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 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 { From 50a1d097943a50471bb85dfcbad4c90096d21ad0 Mon Sep 17 00:00:00 2001 From: Anonymitaet Date: Mon, 8 Feb 2021 12:38:15 +0800 Subject: [PATCH 3/3] update --- site2/docs/functions-runtime.md | 2 +- site2/website/versioned_docs/version-2.7.1/functions-runtime.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/site2/docs/functions-runtime.md b/site2/docs/functions-runtime.md index 9b7deb745461f..8bb4dbbef7dc4 100644 --- a/site2/docs/functions-runtime.md +++ b/site2/docs/functions-runtime.md @@ -280,7 +280,7 @@ The functions (and sinks/sources) API provides a flag, `customRuntimeOptions`, w To initialize the `KubernetesManifestCustomizer`, you can provide `runtimeCustomizerConfig` in the `functions-worker.yml` file. `runtimeCustomizerConfig` is passed to the `public void initialize(Map 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. +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`. diff --git a/site2/website/versioned_docs/version-2.7.1/functions-runtime.md b/site2/website/versioned_docs/version-2.7.1/functions-runtime.md index 9a933122efbf8..6bca7ecdb0483 100644 --- a/site2/website/versioned_docs/version-2.7.1/functions-runtime.md +++ b/site2/website/versioned_docs/version-2.7.1/functions-runtime.md @@ -244,7 +244,7 @@ The functions (and sinks/sources) API provides a flag, `customRuntimeOptions`, w To initialize the `KubernetesManifestCustomizer`, you can provide `runtimeCustomizerConfig` in the `functions-worker.yml` file. `runtimeCustomizerConfig` is passed to the `public void initialize(Map 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. +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`.