You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rx...@apache.org on 2020/08/05 10:44:11 UTC
[pulsar] branch branch-2.6 updated (67d3d98 -> 8b4a921)
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a change to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 67d3d98 Change some WebApplicationException log level to debug (#7725)
new 069d720 [Pulsar SQL] Make Pulsar SQL get correct offload configurations (#7701)
new 84d5ab0 [Issue] Fix get schemaName by partitioned topic name (#7708)
new f509613 [Issue 7711][pulsar-broker] Use original role instead of proxy role to check permissions (#7712)
new d088aeb Replay delayed messages in order. (#7731)
new 36484df Fix broker lookup return wrong url when specify advertised listener (#7737)
new 5dfaac5 Fix functions-worker typos (#7746)
new 8b4a921 Add window context function example
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/ci-integration-sql.yaml | 9 +
conf/presto/config.properties | 2 +
.../authorization/PulsarAuthorizationProvider.java | 6 +-
.../pulsar/broker/namespace/NamespaceService.java | 26 +-
.../PersistentDispatcherMultipleConsumers.java | 6 +-
.../broker/namespace/NamespaceServiceTest.java | 32 +-
.../service/persistent/DelayedDeliveryTest.java | 44 ++
.../pulsar/schema/PartitionedTopicSchemaTest.java | 112 +++++
.../org/apache/pulsar/common/naming/TopicName.java | 2 +-
.../common/policies/data/OffloadPolicies.java | 5 +-
.../pulsar/sql/presto/PulsarConnectorCache.java | 6 +-
.../pulsar/sql/presto/PulsarConnectorConfig.java | 12 +
.../sql/presto/TestPulsarConnectorConfig.java | 32 ++
site2/docs/functions-worker.md | 24 +-
site2/docs/window-functions-context.md | 528 ++++++++++++++++++++
site2/website/sidebars.json | 3 +-
.../version-2.3.0/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.3.1/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.3.2/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.4.0/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.4.1/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.4.2/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.5.0/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.5.1/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.5.2/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.6.0/window-functions-context.md | 529 +++++++++++++++++++++
.../versioned_sidebars/version-2.3.0-sidebars.json | 3 +-
.../versioned_sidebars/version-2.3.1-sidebars.json | 3 +-
.../versioned_sidebars/version-2.3.2-sidebars.json | 3 +-
.../versioned_sidebars/version-2.4.0-sidebars.json | 3 +-
.../versioned_sidebars/version-2.4.1-sidebars.json | 3 +-
.../versioned_sidebars/version-2.4.2-sidebars.json | 3 +-
.../versioned_sidebars/version-2.5.0-sidebars.json | 3 +-
.../versioned_sidebars/version-2.5.1-sidebars.json | 3 +-
.../versioned_sidebars/version-2.5.2-sidebars.json | 3 +-
.../versioned_sidebars/version-2.6.0-sidebars.json | 153 ++++++
.../tests/integration/presto/TestBasicPresto.java | 1 +
...esto.java => TestPrestoQueryTieredStorage.java} | 173 ++++++-
.../integration/topologies/PulsarCluster.java | 17 +-
39 files changed, 6444 insertions(+), 66 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/schema/PartitionedTopicSchemaTest.java
create mode 100644 site2/docs/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.3.0/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.3.1/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.3.2/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.4.0/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.4.1/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.4.2/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.5.0/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.5.1/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.5.2/window-functions-context.md
create mode 100644 site2/website/versioned_docs/version-2.6.0/window-functions-context.md
create mode 100644 site2/website/versioned_sidebars/version-2.6.0-sidebars.json
copy tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/{TestBasicPresto.java => TestPrestoQueryTieredStorage.java} (53%)
[pulsar] 04/07: Replay delayed messages in order. (#7731)
Posted by rx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d088aeba55bf6f8cba5e5855185063f110aafc42
Author: lipenghui <pe...@apache.org>
AuthorDate: Wed Aug 5 11:09:23 2020 +0800
Replay delayed messages in order. (#7731)
### Motivation
Replay delayed messages in order.
### Verifying this change
A new unit test added.
(cherry picked from commit 79df097d18ed29ba0c9a35e07437ba98ae218555)
---
.../PersistentDispatcherMultipleConsumers.java | 6 ++-
.../service/persistent/DelayedDeliveryTest.java | 44 ++++++++++++++++++++++
2 files changed, 49 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index c24e762..b8607f5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -336,7 +336,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
}
havePendingReplayRead = true;
- Set<? extends Position> deletedMessages = asyncReplayEntries(messagesToReplayNow);
+ Set<? extends Position> deletedMessages = topic.delayedDeliveryEnabled ?
+ asyncReplayEntriesInOrder(messagesToReplayNow) : asyncReplayEntries(messagesToReplayNow);
// clear already acked positions from replay bucket
deletedMessages.forEach(position -> messagesToRedeliver.remove(((PositionImpl) position).getLedgerId(),
@@ -372,6 +373,9 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul
return cursor.asyncReplayEntries(positions, this, ReadType.Replay);
}
+ protected Set<? extends Position> asyncReplayEntriesInOrder(Set<? extends Position> positions) {
+ return cursor.asyncReplayEntries(positions, this, ReadType.Replay, true);
+ }
@Override
public boolean isConsumerConnected() {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
index 3cd2e25..82f6241 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/DelayedDeliveryTest.java
@@ -22,6 +22,8 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
@@ -32,6 +34,7 @@ import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.testng.annotations.AfterClass;
@@ -271,4 +274,45 @@ public class DelayedDeliveryTest extends ProducerConsumerBase {
}
t.interrupt();
}
+
+ @Test
+ public void testOrderingDispatch() throws PulsarClientException {
+ String topic = "persistent://public/default/testOrderingDispatch-" + System.nanoTime();
+
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("shared-sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .subscribe();
+
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .create();
+
+ final int N = 1000;
+
+ for (int i = 0; i < N; i++) {
+ producer.newMessage()
+ .value("msg-" + i)
+ .deliverAfter(5, TimeUnit.SECONDS)
+ .send();
+ }
+
+ List<Message<String>> receives = new ArrayList<>(N);
+ for (int i = 0; i < N; i++) {
+ Message<String> received = consumer.receive();
+ receives.add(received);
+ consumer.acknowledge(received);
+ }
+
+ assertEquals(receives.size(), N);
+
+ for (int i = 0; i < N; i++) {
+ if (i < N - 1) {
+ assertTrue(receives.get(i).getMessageId().compareTo(receives.get(i + 1).getMessageId()) < 0);
+ }
+ }
+ }
}
[pulsar] 07/07: Add window context function example
Posted by rx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8b4a92186b45f3109012e32b5aea3b46b264f252
Author: 冉小龙 <rx...@apache.org>
AuthorDate: Wed Aug 5 11:12:24 2020 +0800
Add window context function example
Signed-off-by: xiaolong.ran <rx...@apache.org>
---
site2/docs/window-functions-context.md | 528 ++++++++++++++++++++
site2/website/sidebars.json | 3 +-
.../version-2.3.0/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.3.1/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.3.2/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.4.0/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.4.1/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.4.2/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.5.0/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.5.1/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.5.2/window-functions-context.md | 529 +++++++++++++++++++++
.../version-2.6.0/window-functions-context.md | 529 +++++++++++++++++++++
.../versioned_sidebars/version-2.3.0-sidebars.json | 3 +-
.../versioned_sidebars/version-2.3.1-sidebars.json | 3 +-
.../versioned_sidebars/version-2.3.2-sidebars.json | 3 +-
.../versioned_sidebars/version-2.4.0-sidebars.json | 3 +-
.../versioned_sidebars/version-2.4.1-sidebars.json | 3 +-
.../versioned_sidebars/version-2.4.2-sidebars.json | 3 +-
.../versioned_sidebars/version-2.5.0-sidebars.json | 3 +-
.../versioned_sidebars/version-2.5.1-sidebars.json | 3 +-
.../versioned_sidebars/version-2.5.2-sidebars.json | 3 +-
.../versioned_sidebars/version-2.6.0-sidebars.json | 153 ++++++
22 files changed, 5991 insertions(+), 10 deletions(-)
diff --git a/site2/docs/window-functions-context.md b/site2/docs/window-functions-context.md
new file mode 100644
index 0000000..e81f6e8
--- /dev/null
+++ b/site2/docs/window-functions-context.md
@@ -0,0 +1,528 @@
+---
+id: window-functions-context
+title: Window Functions Context
+sidebar_label: "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/sidebars.json b/site2/website/sidebars.json
index 480e276..a402a13 100644
--- a/site2/website/sidebars.json
+++ b/site2/website/sidebars.json
@@ -32,7 +32,8 @@
"functions-develop",
"functions-debug",
"functions-deploy",
- "functions-cli"
+ "functions-cli",
+ "window-functions-context"
],
"Pulsar IO": [
"io-overview",
diff --git a/site2/website/versioned_docs/version-2.3.0/window-functions-context.md b/site2/website/versioned_docs/version-2.3.0/window-functions-context.md
new file mode 100644
index 0000000..752f95a
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.3.0/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.3.0-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/versioned_docs/version-2.3.1/window-functions-context.md b/site2/website/versioned_docs/version-2.3.1/window-functions-context.md
new file mode 100644
index 0000000..94280f6
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.3.1/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.3.1-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/versioned_docs/version-2.3.2/window-functions-context.md b/site2/website/versioned_docs/version-2.3.2/window-functions-context.md
new file mode 100644
index 0000000..0a4ae0e
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.3.2/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.3.2-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/versioned_docs/version-2.4.0/window-functions-context.md b/site2/website/versioned_docs/version-2.4.0/window-functions-context.md
new file mode 100644
index 0000000..f1a72e0
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.4.0/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.4.0-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/versioned_docs/version-2.4.1/window-functions-context.md b/site2/website/versioned_docs/version-2.4.1/window-functions-context.md
new file mode 100644
index 0000000..e0b0e36
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.4.1/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.4.1-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/versioned_docs/version-2.4.2/window-functions-context.md b/site2/website/versioned_docs/version-2.4.2/window-functions-context.md
new file mode 100644
index 0000000..1faa230
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.4.2/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.4.2-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/versioned_docs/version-2.5.0/window-functions-context.md b/site2/website/versioned_docs/version-2.5.0/window-functions-context.md
new file mode 100644
index 0000000..d9dfd21
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.5.0/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.5.0-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/versioned_docs/version-2.5.1/window-functions-context.md b/site2/website/versioned_docs/version-2.5.1/window-functions-context.md
new file mode 100644
index 0000000..2541cb1
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.5.1/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.5.1-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/versioned_docs/version-2.5.2/window-functions-context.md b/site2/website/versioned_docs/version-2.5.2/window-functions-context.md
new file mode 100644
index 0000000..b51add3
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.5.2/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.5.2-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/versioned_docs/version-2.6.0/window-functions-context.md b/site2/website/versioned_docs/version-2.6.0/window-functions-context.md
new file mode 100644
index 0000000..f4216b9
--- /dev/null
+++ b/site2/website/versioned_docs/version-2.6.0/window-functions-context.md
@@ -0,0 +1,529 @@
+---
+id: version-2.6.0-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/versioned_sidebars/version-2.3.0-sidebars.json b/site2/website/versioned_sidebars/version-2.3.0-sidebars.json
index 5a66aad..4da977c 100644
--- a/site2/website/versioned_sidebars/version-2.3.0-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.3.0-sidebars.json
@@ -25,7 +25,8 @@
"version-2.3.0-functions-deploying",
"version-2.3.0-functions-guarantees",
"version-2.3.0-functions-state",
- "version-2.3.0-functions-metrics"
+ "version-2.3.0-functions-metrics",
+ "version-2.3.0-window-functions-context"
],
"Pulsar IO": [
"version-2.3.0-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.3.1-sidebars.json b/site2/website/versioned_sidebars/version-2.3.1-sidebars.json
index b732733..cd37319 100644
--- a/site2/website/versioned_sidebars/version-2.3.1-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.3.1-sidebars.json
@@ -25,7 +25,8 @@
"version-2.3.1-functions-deploying",
"version-2.3.1-functions-guarantees",
"version-2.3.1-functions-state",
- "version-2.3.1-functions-metrics"
+ "version-2.3.1-functions-metrics",
+ "version-2.3.1-window-functions-context"
],
"Pulsar IO": [
"version-2.3.1-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.3.2-sidebars.json b/site2/website/versioned_sidebars/version-2.3.2-sidebars.json
index f221ae3..16731e3 100644
--- a/site2/website/versioned_sidebars/version-2.3.2-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.3.2-sidebars.json
@@ -26,7 +26,8 @@
"version-2.3.2-functions-guarantees",
"version-2.3.2-functions-state",
"version-2.3.2-functions-metrics",
- "version-2.3.2-functions-worker"
+ "version-2.3.2-functions-worker",
+ "version-2.3.2-window-functions-context"
],
"Pulsar IO": [
"version-2.3.2-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.4.0-sidebars.json b/site2/website/versioned_sidebars/version-2.4.0-sidebars.json
index a6af729..93cef5a 100644
--- a/site2/website/versioned_sidebars/version-2.4.0-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.4.0-sidebars.json
@@ -28,7 +28,8 @@
"version-2.4.0-functions-metrics",
"version-2.4.0-functions-worker",
"version-2.4.0-functions-runtime",
- "version-2.4.0-functions-debugging"
+ "version-2.4.0-functions-debugging",
+ "version-2.4.0-window-functions-context"
],
"Pulsar IO": [
"version-2.4.0-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.4.1-sidebars.json b/site2/website/versioned_sidebars/version-2.4.1-sidebars.json
index c53d0fb..9fc7ff7 100644
--- a/site2/website/versioned_sidebars/version-2.4.1-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.4.1-sidebars.json
@@ -35,7 +35,8 @@
"version-2.4.1-functions-monitor",
"version-2.4.1-functions-secure",
"version-2.4.1-functions-troubleshoot",
- "version-2.4.1-functions-cli"
+ "version-2.4.1-functions-cli",
+ "version-2.4.1-window-functions-context"
],
"Pulsar IO": [
"version-2.4.1-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.4.2-sidebars.json b/site2/website/versioned_sidebars/version-2.4.2-sidebars.json
index e82a731..c119706 100644
--- a/site2/website/versioned_sidebars/version-2.4.2-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.4.2-sidebars.json
@@ -35,7 +35,8 @@
"version-2.4.2-functions-monitor",
"version-2.4.2-functions-secure",
"version-2.4.2-functions-troubleshoot",
- "version-2.4.2-functions-cli"
+ "version-2.4.2-functions-cli",
+ "version-2.4.2-window-functions-context"
],
"Pulsar IO": [
"version-2.4.2-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.5.0-sidebars.json b/site2/website/versioned_sidebars/version-2.5.0-sidebars.json
index 5270e22..dd98dc1 100644
--- a/site2/website/versioned_sidebars/version-2.5.0-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.5.0-sidebars.json
@@ -32,7 +32,8 @@
"version-2.5.0-functions-develop",
"version-2.5.0-functions-debug",
"version-2.5.0-functions-deploy",
- "version-2.5.0-functions-cli"
+ "version-2.5.0-functions-cli",
+ "version-2.5.0-window-functions-context"
],
"Pulsar IO": [
"version-2.5.0-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.5.1-sidebars.json b/site2/website/versioned_sidebars/version-2.5.1-sidebars.json
index e0348e1..64fbb99 100644
--- a/site2/website/versioned_sidebars/version-2.5.1-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.5.1-sidebars.json
@@ -31,7 +31,8 @@
"version-2.5.1-functions-develop",
"version-2.5.1-functions-debug",
"version-2.5.1-functions-deploy",
- "version-2.5.1-functions-cli"
+ "version-2.5.1-functions-cli",
+ "version-2.5.1-window-functions-context"
],
"Pulsar IO": [
"version-2.5.1-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.5.2-sidebars.json b/site2/website/versioned_sidebars/version-2.5.2-sidebars.json
index e97135e..c2456cc 100644
--- a/site2/website/versioned_sidebars/version-2.5.2-sidebars.json
+++ b/site2/website/versioned_sidebars/version-2.5.2-sidebars.json
@@ -31,7 +31,8 @@
"version-2.5.2-functions-develop",
"version-2.5.2-functions-debug",
"version-2.5.2-functions-deploy",
- "version-2.5.2-functions-cli"
+ "version-2.5.2-functions-cli",
+ "version-2.5.2-window-functions-context"
],
"Pulsar IO": [
"version-2.5.2-io-overview",
diff --git a/site2/website/versioned_sidebars/version-2.6.0-sidebars.json b/site2/website/versioned_sidebars/version-2.6.0-sidebars.json
new file mode 100644
index 0000000..6cd23e7
--- /dev/null
+++ b/site2/website/versioned_sidebars/version-2.6.0-sidebars.json
@@ -0,0 +1,153 @@
+{
+ "version-2.6.0-docs": {
+ "Get started": [
+ "version-2.6.0-pulsar-2.0",
+ "version-2.6.0-standalone",
+ "version-2.6.0-standalone-docker",
+ "version-2.6.0-kubernetes-helm",
+ "version-2.6.0-client-libraries"
+ ],
+ "Concepts and Architecture": [
+ "version-2.6.0-concepts-overview",
+ "version-2.6.0-concepts-messaging",
+ "version-2.6.0-concepts-architecture-overview",
+ "version-2.6.0-concepts-clients",
+ "version-2.6.0-concepts-replication",
+ "version-2.6.0-concepts-multi-tenancy",
+ "version-2.6.0-concepts-authentication",
+ "version-2.6.0-concepts-topic-compaction",
+ "version-2.6.0-concepts-tiered-storage"
+ ],
+ "Pulsar Schema": [
+ "version-2.6.0-schema-get-started",
+ "version-2.6.0-schema-understand",
+ "version-2.6.0-schema-evolution-compatibility",
+ "version-2.6.0-schema-manage"
+ ],
+ "Pulsar Functions": [
+ "version-2.6.0-functions-overview",
+ "version-2.6.0-functions-worker",
+ "version-2.6.0-functions-runtime",
+ "version-2.6.0-functions-develop",
+ "version-2.6.0-functions-debug",
+ "version-2.6.0-functions-deploy",
+ "version-2.6.0-functions-cli",
+ "version-2.6.0-window-functions-context"
+ ],
+ "Pulsar IO": [
+ "version-2.6.0-io-overview",
+ "version-2.6.0-io-quickstart",
+ "version-2.6.0-io-use",
+ "version-2.6.0-io-debug",
+ "version-2.6.0-io-connectors",
+ "version-2.6.0-io-cdc",
+ "version-2.6.0-io-develop",
+ "version-2.6.0-io-cli"
+ ],
+ "Pulsar SQL": [
+ "version-2.6.0-sql-overview",
+ "version-2.6.0-sql-getting-started",
+ "version-2.6.0-sql-deployment-configurations",
+ "version-2.6.0-sql-rest-api"
+ ],
+ "Kubernetes (Helm)": [
+ "version-2.6.0-helm-overview",
+ "version-2.6.0-helm-prepare",
+ "version-2.6.0-helm-install",
+ "version-2.6.0-helm-deploy",
+ "version-2.6.0-helm-upgrade",
+ "version-2.6.0-helm-tools"
+ ],
+ "Deployment": [
+ "version-2.6.0-deploy-aws",
+ "version-2.6.0-deploy-kubernetes",
+ "version-2.6.0-deploy-bare-metal",
+ "version-2.6.0-deploy-bare-metal-multi-cluster",
+ "version-2.6.0-deploy-dcos",
+ "version-2.6.0-deploy-monitoring"
+ ],
+ "Administration": [
+ "version-2.6.0-administration-zk-bk",
+ "version-2.6.0-administration-geo",
+ "version-2.6.0-administration-pulsar-manager",
+ "version-2.6.0-administration-stats",
+ "version-2.6.0-administration-load-balance",
+ "version-2.6.0-administration-proxy",
+ "version-2.6.0-administration-upgrade"
+ ],
+ "Security": [
+ "version-2.6.0-security-overview",
+ "version-2.6.0-security-tls-transport",
+ "version-2.6.0-security-tls-authentication",
+ "version-2.6.0-security-tls-keystore",
+ "version-2.6.0-security-jwt",
+ "version-2.6.0-security-athenz",
+ "version-2.6.0-security-kerberos",
+ "version-2.6.0-security-authorization",
+ "version-2.6.0-security-encryption",
+ "version-2.6.0-security-extending",
+ "version-2.6.0-security-bouncy-castle"
+ ],
+ "Performance": [
+ "version-2.6.0-performance-pulsar-perf"
+ ],
+ "Client libraries": [
+ "version-2.6.0-client-libraries-java",
+ "version-2.6.0-client-libraries-go",
+ "version-2.6.0-client-libraries-python",
+ "version-2.6.0-client-libraries-cpp",
+ "version-2.6.0-client-libraries-node",
+ "version-2.6.0-client-libraries-websocket",
+ "version-2.6.0-client-libraries-dotnet"
+ ],
+ "Admin API": [
+ "version-2.6.0-admin-api-overview",
+ "version-2.6.0-admin-api-clusters",
+ "version-2.6.0-admin-api-tenants",
+ "version-2.6.0-admin-api-brokers",
+ "version-2.6.0-admin-api-namespaces",
+ "version-2.6.0-admin-api-permissions",
+ "version-2.6.0-admin-api-persistent-topics",
+ "version-2.6.0-admin-api-non-persistent-topics",
+ "version-2.6.0-admin-api-partitioned-topics",
+ "version-2.6.0-admin-api-non-partitioned-topics",
+ "version-2.6.0-admin-api-schemas",
+ "version-2.6.0-admin-api-functions"
+ ],
+ "Adaptors": [
+ "version-2.6.0-adaptors-kafka",
+ "version-2.6.0-adaptors-spark",
+ "version-2.6.0-adaptors-storm"
+ ],
+ "Cookbooks": [
+ "version-2.6.0-cookbooks-tiered-storage",
+ "version-2.6.0-cookbooks-compaction",
+ "version-2.6.0-cookbooks-deduplication",
+ "version-2.6.0-cookbooks-non-persistent",
+ "version-2.6.0-cookbooks-partitioned",
+ "version-2.6.0-cookbooks-retention-expiry",
+ "version-2.6.0-cookbooks-encryption",
+ "version-2.6.0-cookbooks-message-queue",
+ "version-2.6.0-cookbooks-bookkeepermetadata"
+ ],
+ "Development": [
+ "version-2.6.0-develop-tools",
+ "version-2.6.0-develop-binary-protocol",
+ "version-2.6.0-develop-schema",
+ "version-2.6.0-develop-load-manager",
+ "version-2.6.0-develop-cpp"
+ ],
+ "Reference": [
+ "version-2.6.0-reference-terminology",
+ "version-2.6.0-reference-cli-tools",
+ "version-2.6.0-reference-configuration",
+ "version-2.6.0-reference-metrics"
+ ]
+ },
+ "version-2.6.0-docs-other": {
+ "First Category": [
+ "version-2.6.0-doc4",
+ "version-2.6.0-doc5"
+ ]
+ }
+}
[pulsar] 05/07: Fix broker lookup return wrong url when specify
advertised listener (#7737)
Posted by rx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 36484df8c247e03ad8b28eb66b98accd2aa625bb
Author: wangjialing218 <65...@users.noreply.github.com>
AuthorDate: Wed Aug 5 18:29:49 2020 +0800
Fix broker lookup return wrong url when specify advertised listener (#7737)
get advertised listener name from lookup options
add test case
Co-authored-by: wangjialing <wa...@cmss.chinamobile.com>
(cherry picked from commit 109c8bfcdc4abbbc9fa31e75843959c23b877c51)
---
.../pulsar/broker/namespace/NamespaceService.java | 26 ++++++++++++++----
.../broker/namespace/NamespaceServiceTest.java | 32 ++++++++++++++++++++--
2 files changed, 50 insertions(+), 8 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index b50e769..cb8cc27 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -65,6 +65,7 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener;
+import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException;
@@ -522,7 +523,7 @@ public class NamespaceService {
}
// Now setting the redirect url
- createLookupResult(candidateBroker, authoritativeRedirect)
+ createLookupResult(candidateBroker, authoritativeRedirect, options.getAdvertisedListenerName())
.thenAccept(lookupResult -> lookupFuture.complete(Optional.of(lookupResult)))
.exceptionally(ex -> {
lookupFuture.completeExceptionally(ex);
@@ -536,7 +537,7 @@ public class NamespaceService {
}
}
- protected CompletableFuture<LookupResult> createLookupResult(String candidateBroker, boolean authoritativeRedirect)
+ protected CompletableFuture<LookupResult> createLookupResult(String candidateBroker, boolean authoritativeRedirect, final String advertisedListenerName)
throws Exception {
CompletableFuture<LookupResult> lookupFuture = new CompletableFuture<>();
@@ -547,10 +548,23 @@ public class NamespaceService {
uri.getPort());
pulsar.getLocalZkCache().getDataAsync(path, pulsar.getLoadManager().get().getLoadReportDeserializer()).thenAccept(reportData -> {
if (reportData.isPresent()) {
- ServiceLookupData lookupData = reportData.get();
- lookupFuture.complete(new LookupResult(lookupData.getWebServiceUrl(),
- lookupData.getWebServiceUrlTls(), lookupData.getPulsarServiceUrl(),
- lookupData.getPulsarServiceUrlTls(), authoritativeRedirect));
+ LocalBrokerData lookupData = (LocalBrokerData) reportData.get();
+ if (StringUtils.isNotBlank(advertisedListenerName)) {
+ AdvertisedListener listener = lookupData.getAdvertisedListeners().get(advertisedListenerName);
+ if (listener == null) {
+ lookupFuture.completeExceptionally(
+ new PulsarServerException("the broker do not have " + advertisedListenerName + " listener"));
+ } else {
+ URI urlTls = listener.getBrokerServiceUrlTls();
+ lookupFuture.complete(new LookupResult(lookupData.getWebServiceUrl(),
+ lookupData.getWebServiceUrlTls(), listener.getBrokerServiceUrl().toString(),
+ urlTls == null ? null : urlTls.toString(), authoritativeRedirect));
+ }
+ } else {
+ lookupFuture.complete(new LookupResult(lookupData.getWebServiceUrl(),
+ lookupData.getWebServiceUrlTls(), lookupData.getPulsarServiceUrl(),
+ lookupData.getPulsarServiceUrlTls(), authoritativeRedirect));
+ }
} else {
lookupFuture.completeExceptionally(new KeeperException.NoNodeException(path));
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
index 743bba0..bcf49bf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
@@ -35,6 +35,7 @@ import static org.testng.Assert.fail;
import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.hash.Hashing;
import java.lang.reflect.Field;
@@ -42,6 +43,7 @@ import java.lang.reflect.Method;
import java.net.URI;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -68,6 +70,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
+import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener;
import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
import org.apache.zookeeper.CreateMode;
@@ -358,19 +361,44 @@ public class NamespaceServiceTest extends BrokerTestBase {
ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), path2,
ObjectMapperFactory.getThreadLocal().writeValueAsBytes(ld), ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
- LookupResult result1 = pulsar.getNamespaceService().createLookupResult(candidateBroker1, false).get();
+ LookupResult result1 = pulsar.getNamespaceService().createLookupResult(candidateBroker1, false, null).get();
// update to new load manager
LoadManager oldLoadManager = pulsar.getLoadManager()
.getAndSet(new ModularLoadManagerWrapper(new ModularLoadManagerImpl()));
oldLoadManager.stop();
- LookupResult result2 = pulsar.getNamespaceService().createLookupResult(candidateBroker2, false).get();
+ LookupResult result2 = pulsar.getNamespaceService().createLookupResult(candidateBroker2, false, null).get();
Assert.assertEquals(result1.getLookupData().getBrokerUrl(), candidateBroker1);
Assert.assertEquals(result2.getLookupData().getBrokerUrl(), candidateBroker2);
System.out.println(result2);
}
@Test
+ public void testCreateLookupResult() throws Exception {
+
+ final String candidateBroker = "pulsar://localhost:6650";
+ final String listenerUrl = "pulsar://localhost:7000";
+ final String listenerUrlTls = "pulsar://localhost:8000";
+ final String listener = "listenerName";
+ Map<String, AdvertisedListener> advertisedListeners = Maps.newHashMap();
+ advertisedListeners.put(listener, AdvertisedListener.builder().brokerServiceUrl(new URI(listenerUrl)).brokerServiceUrlTls(new URI(listenerUrlTls)).build());
+ LocalBrokerData ld = new LocalBrokerData(null, null, candidateBroker, null, advertisedListeners);
+ URI uri = new URI(candidateBroker);
+ String path = String.format("%s/%s:%s", LoadManager.LOADBALANCE_BROKERS_ROOT, uri.getHost(), uri.getPort());
+ ZkUtils.createFullPathOptimistic(pulsar.getZkClient(), path,
+ ObjectMapperFactory.getThreadLocal().writeValueAsBytes(ld), ZooDefs.Ids.OPEN_ACL_UNSAFE,
+ CreateMode.EPHEMERAL);
+
+ LookupResult noListener = pulsar.getNamespaceService().createLookupResult(candidateBroker, false, null).get();
+ LookupResult withListener = pulsar.getNamespaceService().createLookupResult(candidateBroker, false, listener).get();
+
+ Assert.assertEquals(noListener.getLookupData().getBrokerUrl(), candidateBroker);
+ Assert.assertEquals(withListener.getLookupData().getBrokerUrl(), listenerUrl);
+ Assert.assertEquals(withListener.getLookupData().getBrokerUrlTls(), listenerUrlTls);
+ System.out.println(withListener);
+ }
+
+ @Test
public void testCreateNamespaceWithDefaultNumberOfBundles() throws Exception {
OwnershipCache MockOwnershipCache = spy(pulsar.getNamespaceService().getOwnershipCache());
doNothing().when(MockOwnershipCache).disableOwnership(any(NamespaceBundle.class));
[pulsar] 03/07: [Issue 7711][pulsar-broker] Use original role
instead of proxy role to check permissions (#7712)
Posted by rx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f50961387bf399376ddccc172c0563bb8f0ad797
Author: Sergii Zhevzhyk <vz...@users.noreply.github.com>
AuthorDate: Wed Aug 5 10:50:11 2020 +0200
[Issue 7711][pulsar-broker] Use original role instead of proxy role to check permissions (#7712)
Fixes #7711
### Motivation
A client can have more permissions than its original role allows and it is a security violation.
### Modifications
Use the original role (if available) instead of the proxy role to check if a client is allowed to consume or produce messages.
(cherry picked from commit 5546c5fed6487faf7e7bb5023eb936573aa929bf)
---
.../pulsar/broker/authorization/PulsarAuthorizationProvider.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index 1aa79bf..60038f9 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -539,11 +539,11 @@ public class PulsarAuthorizationProvider implements AuthorizationProvider {
CompletableFuture<Boolean> isAuthorizedFuture;
switch (operation) {
- case LOOKUP: isAuthorizedFuture = canLookupAsync(topicName, role, authData);
+ case LOOKUP: isAuthorizedFuture = canLookupAsync(topicName, StringUtils.isBlank(originalRole) ? role : originalRole, authData);
break;
- case PRODUCE: isAuthorizedFuture= canProduceAsync(topicName, role, authData);
+ case PRODUCE: isAuthorizedFuture = canProduceAsync(topicName, StringUtils.isBlank(originalRole) ? role : originalRole, authData);
break;
- case CONSUME: isAuthorizedFuture = canConsumeAsync(topicName, role, authData, authData.getSubscription());
+ case CONSUME: isAuthorizedFuture = canConsumeAsync(topicName, StringUtils.isBlank(originalRole) ? role : originalRole, authData, authData.getSubscription());
break;
default: isAuthorizedFuture = FutureUtil.failedFuture(
new IllegalStateException("TopicOperation is not supported."));
[pulsar] 02/07: [Issue] Fix get schemaName by partitioned topic
name (#7708)
Posted by rx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 84d5ab086d18812ae9ba29be602f6a347ef81564
Author: ran <ga...@126.com>
AuthorDate: Wed Aug 5 12:01:07 2020 +0800
[Issue] Fix get schemaName by partitioned topic name (#7708)
Fixes #7676
### Motivation
Currently, there is an error that gets schemaName by partitioned topic name.
### Modifications
Add partitioned check in method `topicName. getSchemaName()`.
(cherry picked from commit 7525ab90cb5c072237a3ae2bc665930cbc691f8a)
---
.../pulsar/schema/PartitionedTopicSchemaTest.java | 112 +++++++++++++++++++++
.../org/apache/pulsar/common/naming/TopicName.java | 2 +-
2 files changed, 113 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/PartitionedTopicSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/PartitionedTopicSchemaTest.java
new file mode 100644
index 0000000..c112c13
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/PartitionedTopicSchemaTest.java
@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.schema;
+
+import com.google.common.collect.Sets;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Test get partitioned topic schema.
+ */
+public class PartitionedTopicSchemaTest extends MockedPulsarServiceBaseTest {
+
+ private final static String PARTITIONED_TOPIC = "public/default/partitioned-schema-topic";
+ private final static int MESSAGE_COUNT_PER_PARTITION = 12;
+ private final static int TOPIC_PARTITION = 3;
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ isTcpLookup = true;
+ super.internalSetup();
+
+ admin.clusters().createCluster("test", new ClusterData(pulsar.getWebServiceAddress()));
+ admin.tenants().createTenant("my-property",
+ new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
+ admin.namespaces().createNamespace("my-property/my-ns");
+ admin.namespaces().setNamespaceReplicationClusters("my-property/my-ns", Sets.newHashSet("test"));
+
+ // so that clients can test short names
+ admin.tenants().createTenant("public",
+ new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
+ admin.namespaces().createNamespace("public/default");
+ admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test"));
+ admin.topics().createPartitionedTopic(PARTITIONED_TOPIC, TOPIC_PARTITION);
+ }
+
+ @AfterMethod
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void test() throws Exception {
+ Consumer<GenericRecord> consumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME())
+ .topic(PARTITIONED_TOPIC)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionName("test")
+ .subscribe();
+ consumer.close();
+
+ @Cleanup
+ Producer<Schemas.PersonFour> producer = pulsarClient.newProducer(Schema.JSON(Schemas.PersonFour.class))
+ .topic(PARTITIONED_TOPIC)
+ .enableBatching(false)
+ .roundRobinRouterBatchingPartitionSwitchFrequency(1)
+ .create();
+
+ for (int i = 0; i < MESSAGE_COUNT_PER_PARTITION * TOPIC_PARTITION; i++) {
+ Schemas.PersonFour person = new Schemas.PersonFour();
+ person.setId(i);
+ person.setName("user-" + i);
+ person.setAge(18);
+ producer.newMessage().value(person).send();
+ }
+
+ consumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME())
+ .topic(TopicName.get(PARTITIONED_TOPIC).getPartition(1).toString())
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionName("test")
+ .subscribe();
+
+ int receiveMsgCount = 0;
+ for (int i = 0; i < MESSAGE_COUNT_PER_PARTITION; i++) {
+ Message<GenericRecord> message = consumer.receive();
+ Assert.assertNotNull(message);
+ receiveMsgCount++;
+ }
+ Assert.assertEquals(MESSAGE_COUNT_PER_PARTITION, receiveMsgCount);
+ }
+
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index 5ec0d58..49eec35 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -342,7 +342,7 @@ public class TopicName implements ServiceUnitId {
public String getSchemaName() {
return getTenant()
+ "/" + getNamespacePortion()
- + "/" + getEncodedLocalName();
+ + "/" + TopicName.get(getPartitionedTopicName()).getEncodedLocalName();
}
@Override
[pulsar] 01/07: [Pulsar SQL] Make Pulsar SQL get correct offload
configurations (#7701)
Posted by rx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 069d72053774556e8873b0b81ad16319d4a7570c
Author: ran <ga...@126.com>
AuthorDate: Wed Aug 5 17:07:52 2020 +0800
[Pulsar SQL] Make Pulsar SQL get correct offload configurations (#7701)
### Motivation
Currently, Pulsar SQL can't get the correct offload configurations.
### Modifications
Make Pulsar SQL get the complete offload configurations.
### Verifying this change
Add a new integration test.
(cherry picked from commit 9f687d3f67d7534cb9ee77ddd2c7d0f24d34d0b4)
---
.github/workflows/ci-integration-sql.yaml | 9 ++
conf/presto/config.properties | 2 +
.../common/policies/data/OffloadPolicies.java | 5 +-
.../pulsar/sql/presto/PulsarConnectorCache.java | 6 +-
.../pulsar/sql/presto/PulsarConnectorConfig.java | 12 ++
.../sql/presto/TestPulsarConnectorConfig.java | 32 ++++
.../tests/integration/presto/TestBasicPresto.java | 1 +
...esto.java => TestPrestoQueryTieredStorage.java} | 173 +++++++++++++++++----
.../integration/topologies/PulsarCluster.java | 17 +-
9 files changed, 225 insertions(+), 32 deletions(-)
diff --git a/.github/workflows/ci-integration-sql.yaml b/.github/workflows/ci-integration-sql.yaml
index 5b4b071..f90f301 100644
--- a/.github/workflows/ci-integration-sql.yaml
+++ b/.github/workflows/ci-integration-sql.yaml
@@ -62,6 +62,15 @@ jobs:
if: steps.docs.outputs.changed_only == 'no'
run: mvn clean install -DskipTests
+# Flaky Test: https://github.com/apache/pulsar/issues/7750
+# - name: build pulsar image
+# if: steps.docs.outputs.changed_only == 'no'
+# run: mvn -B -f docker/pulsar/pom.xml install -am -Pdocker -DskipTests -Ddocker.nocache=true
+#
+# - name: build pulsar-all image
+# if: steps.docs.outputs.changed_only == 'no'
+# run: mvn -B -f docker/pulsar-all/pom.xml install -am -Pdocker -DskipTests -Ddocker.nocache=true
+#
- name: build artifacts and docker pulsar latest test image
if: steps.docs.outputs.changed_only == 'no'
run: mvn -B -f tests/docker-images/pom.xml install -am -Pdocker -DskipTests
diff --git a/conf/presto/config.properties b/conf/presto/config.properties
index 9f17135..2ca62ae 100644
--- a/conf/presto/config.properties
+++ b/conf/presto/config.properties
@@ -38,5 +38,7 @@ query.client.timeout=5m
query.min-expire-age=30m
presto.version=testversion
+
distributed-joins-enabled=true
+
node-scheduler.include-coordinator=true
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPolicies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPolicies.java
index 4936923..f377802 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPolicies.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPolicies.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.common.policies.data;
import static org.apache.pulsar.common.util.FieldParser.value;
import com.google.common.base.MoreObjects;
+import java.io.Serializable;
import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.Objects;
@@ -32,7 +33,9 @@ import org.apache.commons.lang3.StringUtils;
* Definition of the offload policies.
*/
@Data
-public class OffloadPolicies {
+public class OffloadPolicies implements Serializable {
+
+ private final static long serialVersionUID = 0L;
public final static int DEFAULT_MAX_BLOCK_SIZE_IN_BYTES = 64 * 1024 * 1024; // 64MB
public final static int DEFAULT_READ_BUFFER_SIZE_IN_BYTES = 1024 * 1024; // 1MB
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
index f4a7d74..091efe1 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
@@ -40,7 +40,6 @@ import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
import org.apache.bookkeeper.mledger.offload.OffloaderUtils;
import org.apache.bookkeeper.mledger.offload.Offloaders;
import org.apache.bookkeeper.stats.StatsProvider;
-import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.PulsarVersion;
import org.apache.pulsar.common.naming.NamespaceName;
@@ -81,9 +80,8 @@ public class PulsarConnectorCache {
this.statsProvider.start(clientConfiguration);
- OffloadPolicies offloadPolicies = new OffloadPolicies();
- BeanUtils.copyProperties(offloadPolicies, pulsarConnectorConfig);
- this.defaultOffloader = initManagedLedgerOffloader(offloadPolicies, pulsarConnectorConfig);
+ this.defaultOffloader = initManagedLedgerOffloader(
+ pulsarConnectorConfig.getOffloadPolices(), pulsarConnectorConfig);
}
public static PulsarConnectorCache getConnectorCache(PulsarConnectorConfig pulsarConnectorConfig) throws Exception {
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java
index 49d2ae3..0a172d9 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorConfig.java
@@ -23,6 +23,7 @@ import io.airlift.configuration.Config;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
import java.util.regex.Matcher;
import javax.validation.constraints.NotNull;
import org.apache.bookkeeper.stats.NullStatsProvider;
@@ -31,6 +32,7 @@ import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.naming.NamedEntity;
import org.apache.pulsar.common.nar.NarClassLoader;
+import org.apache.pulsar.common.policies.data.OffloadPolicies;
import org.apache.pulsar.common.protocol.Commands;
/**
@@ -399,6 +401,16 @@ public class PulsarConnectorConfig implements AutoCloseable {
return this.pulsarAdmin;
}
+ public OffloadPolicies getOffloadPolices() {
+ Properties offloadProperties = new Properties();
+ offloadProperties.putAll(getOffloaderProperties());
+ OffloadPolicies offloadPolicies = OffloadPolicies.create(offloadProperties);
+ offloadPolicies.setManagedLedgerOffloadDriver(getManagedLedgerOffloadDriver());
+ offloadPolicies.setManagedLedgerOffloadMaxThreads(getManagedLedgerOffloadMaxThreads());
+ offloadPolicies.setOffloadersDirectory(getOffloadersDirectory());
+ return offloadPolicies;
+ }
+
@Override
public void close() throws Exception {
this.pulsarAdmin.close();
diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnectorConfig.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnectorConfig.java
index faf2bbc..f3d2f7a 100644
--- a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnectorConfig.java
+++ b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnectorConfig.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.sql.presto;
+import org.apache.pulsar.common.policies.data.OffloadPolicies;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -68,4 +69,35 @@ public class TestPulsarConnectorConfig {
Assert.assertEquals(availableProcessors, connectorConfig.getManagedLedgerNumSchedulerThreads());
}
+ @Test
+ public void testGetOffloadPolices() throws Exception {
+ PulsarConnectorConfig connectorConfig = new PulsarConnectorConfig();
+
+ final String managedLedgerOffloadDriver = "s3";
+ final String offloaderDirectory = "/pulsar/offloaders";
+ final int managedLedgerOffloadMaxThreads = 5;
+ final String bucket = "offload-bucket";
+ final String region = "us-west-2";
+ final String endpoint = "http://s3.amazonaws.com";
+ final String offloadProperties = "{"
+ + "\"s3ManagedLedgerOffloadBucket\":\"" + bucket + "\","
+ + "\"s3ManagedLedgerOffloadRegion\":\"" + region + "\","
+ + "\"s3ManagedLedgerOffloadServiceEndpoint\":\"" + endpoint + "\""
+ + "}";
+
+ connectorConfig.setManagedLedgerOffloadDriver(managedLedgerOffloadDriver);
+ connectorConfig.setOffloadersDirectory(offloaderDirectory);
+ connectorConfig.setManagedLedgerOffloadMaxThreads(managedLedgerOffloadMaxThreads);
+ connectorConfig.setOffloaderProperties(offloadProperties);
+
+ OffloadPolicies offloadPolicies = connectorConfig.getOffloadPolices();
+ Assert.assertNotNull(offloadPolicies);
+ Assert.assertEquals(offloadPolicies.getManagedLedgerOffloadDriver(), managedLedgerOffloadDriver);
+ Assert.assertEquals(offloadPolicies.getOffloadersDirectory(), offloaderDirectory);
+ Assert.assertEquals(offloadPolicies.getManagedLedgerOffloadMaxThreads(), managedLedgerOffloadMaxThreads);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadBucket(), bucket);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadRegion(), region);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadServiceEndpoint(), endpoint);
+ }
+
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
index 093f3eb..e29d945 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
@@ -50,6 +50,7 @@ public class TestBasicPresto extends PulsarTestSuite {
@BeforeClass
public void setupPresto() throws Exception {
+ log.info("[setupPresto]");
pulsarCluster.startPrestoWorker();
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPrestoQueryTieredStorage.java
similarity index 53%
copy from tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
copy to tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPrestoQueryTieredStorage.java
index 093f3eb..8ba48aa 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPrestoQueryTieredStorage.java
@@ -20,17 +20,24 @@ package org.apache.pulsar.tests.integration.presto;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.tests.integration.containers.BrokerContainer;
+import org.apache.pulsar.tests.integration.containers.S3Container;
import org.apache.pulsar.tests.integration.docker.ContainerExecException;
import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
import org.apache.pulsar.tests.integration.suites.PulsarTestSuite;
import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
+import org.testcontainers.shaded.org.apache.commons.lang.StringUtils;
+import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterSuite;
import org.testng.annotations.BeforeClass;
-import org.testng.annotations.Test;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -38,38 +45,83 @@ import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Timestamp;
+import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
+import static com.google.common.base.Preconditions.checkNotNull;
import static org.assertj.core.api.Assertions.assertThat;
@Slf4j
-public class TestBasicPresto extends PulsarTestSuite {
+public class TestPrestoQueryTieredStorage extends PulsarTestSuite {
- private static final int NUM_OF_STOCKS = 10;
+ private final static int ENTRIES_PER_LEDGER = 1024;
+ private final static String OFFLOAD_DRIVER = "s3";
+ private final static String BUCKET = "pulsar-integtest";
+ private final static String ENDPOINT = "http://" + S3Container.NAME + ":9090";
+
+ private S3Container s3Container;
+
+ @Override
+ protected void beforeStartCluster() throws Exception {
+ for (BrokerContainer brokerContainer : pulsarCluster.getBrokers()) {
+ getEnv().forEach(brokerContainer::withEnv);
+ }
+ }
@BeforeClass
public void setupPresto() throws Exception {
- pulsarCluster.startPrestoWorker();
+ s3Container = new S3Container(
+ pulsarCluster.getClusterName(),
+ S3Container.NAME)
+ .withNetwork(pulsarCluster.getNetwork())
+ .withNetworkAliases(S3Container.NAME);
+ s3Container.start();
+
+ log.info("[setupPresto] prestoWorker: " + pulsarCluster.getPrestoWorkerContainer());
+ pulsarCluster.startPrestoWorker(OFFLOAD_DRIVER, getOffloadProperties(BUCKET, null, ENDPOINT));
}
+ public String getOffloadProperties(String bucket, String region, String endpoint) {
+ checkNotNull(bucket);
+ StringBuilder sb = new StringBuilder();
+ sb.append("{");
+ sb.append("\"s3ManagedLedgerOffloadBucket\":").append("\"").append(bucket).append("\",");
+ if (StringUtils.isNotEmpty(region)) {
+ sb.append("\"s3ManagedLedgerOffloadRegion\":").append("\"").append(region).append("\",");
+ }
+ if (StringUtils.isNotEmpty(endpoint)) {
+ sb.append("\"s3ManagedLedgerOffloadServiceEndpoint\":").append("\"").append(endpoint).append("\"");
+ }
+ sb.append("}");
+ return sb.toString();
+ }
+
+
@AfterClass
public void teardownPresto() {
log.info("tearing down...");
+ if (null != s3Container) {
+ s3Container.stop();
+ }
+
pulsarCluster.stopPrestoWorker();
}
- @Test
- public void testSimpleSQLQueryBatched() throws Exception {
- testSimpleSQLQuery(true);
+ // Flaky Test: https://github.com/apache/pulsar/issues/7750
+ // @Test
+ public void testQueryTieredStorage1() throws Exception {
+ testSimpleSQLQuery(false);
}
- @Test
- public void testSimpleSQLQueryNonBatched() throws Exception {
- testSimpleSQLQuery(false);
+ // Flaky Test: https://github.com/apache/pulsar/issues/7750
+ // @Test
+ public void testQueryTieredStorage2() throws Exception {
+ testSimpleSQLQuery(true);
}
-
- public void testSimpleSQLQuery(boolean isBatched) throws Exception {
+
+ public void testSimpleSQLQuery(boolean isNamespaceOffload) throws Exception {
// wait until presto worker started
ContainerExecResult result;
@@ -93,42 +145,51 @@ public class TestBasicPresto extends PulsarTestSuite {
.serviceUrl(pulsarCluster.getPlainTextServiceUrl())
.build();
- String stocksTopic;
- if (isBatched) {
- stocksTopic = "stocks_batched";
- } else {
- stocksTopic = "stocks_nonbatched";
- }
+ String stocksTopic = "stocks-" + randomName(5);
@Cleanup
Producer<Stock> producer = pulsarClient.newProducer(JSONSchema.of(Stock.class))
.topic(stocksTopic)
- .enableBatching(isBatched)
.create();
- for (int i = 0 ; i < NUM_OF_STOCKS; ++i) {
- final Stock stock = new Stock(i,"STOCK_" + i , 100.0 + i * 10);
- producer.send(stock);
+ long firstLedgerId = -1;
+ long currentLedgerId = -1;
+ int sendMessageCnt = 0;
+ while (currentLedgerId <= firstLedgerId) {
+ sendMessageCnt ++;
+ final Stock stock = new Stock(sendMessageCnt,"STOCK_" + sendMessageCnt , 100.0 + sendMessageCnt * 10);
+ MessageIdImpl messageId = (MessageIdImpl) producer.send(stock);
+ if (firstLedgerId == -1) {
+ firstLedgerId = messageId.getLedgerId();
+ }
+ currentLedgerId = messageId.getLedgerId();
+ log.info("firstLedgerId: {}, currentLedgerId: {}", firstLedgerId, currentLedgerId);
+ Thread.sleep(100);
}
producer.flush();
+ offloadAndDeleteFromBK(isNamespaceOffload, stocksTopic);
+
+ // check schema
result = execQuery("show schemas in pulsar;");
assertThat(result.getExitCode()).isEqualTo(0);
assertThat(result.getStdout()).contains("public/default");
+ // check table
result = execQuery("show tables in pulsar.\"public/default\";");
assertThat(result.getExitCode()).isEqualTo(0);
- assertThat(result.getStdout()).contains("stocks");
+ assertThat(result.getStdout()).contains(stocksTopic);
+ // check query
ContainerExecResult containerExecResult = execQuery(String.format("select * from pulsar.\"public/default\".%s order by entryid;", stocksTopic));
assertThat(containerExecResult.getExitCode()).isEqualTo(0);
log.info("select sql query output \n{}", containerExecResult.getStdout());
String[] split = containerExecResult.getStdout().split("\n");
- assertThat(split.length).isGreaterThan(NUM_OF_STOCKS - 2);
+ assertThat(split.length).isGreaterThan(sendMessageCnt - 2);
String[] split2 = containerExecResult.getStdout().split("\n|,");
- for (int i = 0; i < NUM_OF_STOCKS - 2; ++i) {
+ for (int i = 0; i < sendMessageCnt - 2; ++i) {
assertThat(split2).contains("\"" + i + "\"");
assertThat(split2).contains("\"" + "STOCK_" + i + "\"");
assertThat(split2).contains("\"" + (100.0 + i * 10) + "\"");
@@ -150,7 +211,7 @@ public class TestBasicPresto extends PulsarTestSuite {
timestamps.add(res.getTimestamp("__publish_time__"));
}
- assertThat(timestamps.size()).isGreaterThan(NUM_OF_STOCKS - 2);
+ assertThat(timestamps.size()).isGreaterThan(sendMessageCnt - 2);
query = String.format("select * from pulsar" +
".\"public/default\".%s where __publish_time__ > timestamp '%s' order by __publish_time__", stocksTopic, timestamps.get(timestamps.size() / 2));
@@ -225,4 +286,64 @@ public class TestBasicPresto extends PulsarTestSuite {
}
+ private void offloadAndDeleteFromBK(boolean isNamespaceOffload, String stocksTopic) {
+ String adminUrl = pulsarCluster.getHttpServiceUrl();
+ try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(adminUrl).build()) {
+ // read managed ledger info, check ledgers exist
+ long firstLedger = admin.topics().getInternalStats(stocksTopic).ledgers.get(0).ledgerId;
+
+ String output = "";
+
+ if (isNamespaceOffload) {
+ pulsarCluster.runAdminCommandOnAnyBroker(
+ "namespaces", "set-offload-policies",
+ "--bucket", "pulsar-integtest",
+ "--driver", "s3",
+ "--endpoint", "http://" + S3Container.NAME + ":9090",
+ "--offloadAfterElapsed", "1000",
+ "public/default");
+
+ output = pulsarCluster.runAdminCommandOnAnyBroker(
+ "namespaces", "get-offload-policies").getStdout();
+ Assert.assertTrue(output.contains("pulsar-integtest"));
+ Assert.assertTrue(output.contains("s3"));
+ }
+
+ // offload with a low threshold
+ output = pulsarCluster.runAdminCommandOnAnyBroker("topics",
+ "offload", "--size-threshold", "1M", stocksTopic).getStdout();
+ Assert.assertTrue(output.contains("Offload triggered"));
+
+ output = pulsarCluster.runAdminCommandOnAnyBroker("topics",
+ "offload-status", "-w", stocksTopic).getStdout();
+ Assert.assertTrue(output.contains("Offload was a success"));
+
+ // delete the first ledger, so that we cannot possibly read from it
+ ClientConfiguration bkConf = new ClientConfiguration();
+ bkConf.setZkServers(pulsarCluster.getZKConnString());
+ try (BookKeeper bk = new BookKeeper(bkConf)) {
+ bk.deleteLedger(firstLedger);
+ } catch (Exception e) {
+ log.error("Failed to delete from BookKeeper.", e);
+ Assert.fail("Failed to delete from BookKeeper.");
+ }
+
+ // Unload topic to clear all caches, open handles, etc
+ admin.topics().unload(stocksTopic);
+ } catch (Exception e) {
+ Assert.fail("Failed to deleteOffloadedDataFromBK.");
+ }
+ }
+
+ protected Map<String, String> getEnv() {
+ Map<String, String> result = new HashMap<>();
+ result.put("managedLedgerMaxEntriesPerLedger", String.valueOf(ENTRIES_PER_LEDGER));
+ result.put("managedLedgerMinLedgerRolloverTimeMinutes", "0");
+ result.put("managedLedgerOffloadDriver", OFFLOAD_DRIVER);
+ result.put("s3ManagedLedgerOffloadBucket", BUCKET);
+ result.put("s3ManagedLedgerOffloadServiceEndpoint", ENDPOINT);
+
+ return result;
+ }
+
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
index 6ffd38a..36175a6 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
@@ -328,6 +328,11 @@ public class PulsarCluster {
}
public void startPrestoWorker() {
+ startPrestoWorker(null, null);
+ }
+
+ public void startPrestoWorker(String offloadDriver, String offloadProperties) {
+ log.info("[startPrestoWorker] offloadDriver: {}, offloadProperties: {}", offloadDriver, offloadProperties);
if (null == prestoWorkerContainer) {
prestoWorkerContainer = new PrestoWorkerContainer(clusterName, PrestoWorkerContainer.NAME)
.withNetwork(network)
@@ -337,8 +342,18 @@ public class PulsarCluster {
.withEnv("zookeeperServers", ZKContainer.NAME + ":" + ZKContainer.ZK_PORT)
.withEnv("pulsar.zookeeper-uri", ZKContainer.NAME + ":" + ZKContainer.ZK_PORT)
.withEnv("pulsar.broker-service-url", "http://pulsar-broker-0:8080");
+ if (offloadDriver != null && offloadProperties != null) {
+ log.info("[startPrestoWorker] set offload env offloadDriver: {}, offloadProperties: {}",
+ offloadDriver, offloadProperties);
+ prestoWorkerContainer.withEnv("PULSAR_PREFIX_pulsar.managed-ledger-offload-driver", offloadDriver);
+ prestoWorkerContainer.withEnv("PULSAR_PREFIX_pulsar.offloader-properties", offloadProperties);
+ prestoWorkerContainer.withEnv("PULSAR_PREFIX_pulsar.offloaders-directory", "/pulsar/offloaders");
+ // used in s3 tests
+ prestoWorkerContainer.withEnv("AWS_ACCESS_KEY_ID", "accesskey");
+ prestoWorkerContainer.withEnv("AWS_SECRET_KEY", "secretkey");
+ }
}
- log.info("Starting Presto Worker");
+ log.info("[startPrestoWorker] Starting Presto Worker");
prestoWorkerContainer.start();
}
[pulsar] 06/07: Fix functions-worker typos (#7746)
Posted by rx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rxl pushed a commit to branch branch-2.6
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5dfaac58f24c93338f9030381b7495e7ee048802
Author: sijia-w <53...@users.noreply.github.com>
AuthorDate: Wed Aug 5 08:33:54 2020 +0200
Fix functions-worker typos (#7746)
* Update functions-worker.md
* Update functions-worker.md
* Update site2/docs/functions-worker.md
Co-authored-by: Jennifer Huang <47...@users.noreply.github.com>
(cherry picked from commit d3076f21b088a7062659d144b3f6b3efdacb848b)
---
site2/docs/functions-worker.md | 24 +++++++++++++-----------
1 file changed, 13 insertions(+), 11 deletions(-)
diff --git a/site2/docs/functions-worker.md b/site2/docs/functions-worker.md
index 35bfd9a..7970782 100644
--- a/site2/docs/functions-worker.md
+++ b/site2/docs/functions-worker.md
@@ -5,7 +5,7 @@ sidebar_label: "Setup: Pulsar 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.md).
-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.
+Pulsar `functions-worker` is a logic component to run Pulsar Functions in cluster mode. Two options are available, and you can select either based on your requirements.
- [run with brokers](#run-functions-worker-with-brokers)
- [run it separately](#run-functions-worker-separately) in a different broker
@@ -24,16 +24,16 @@ To enable functions-worker running as part of a broker, you need to set `functio
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.
+If the `functionsWorkerEnabled` is set to `true`, the functions-worker is started as part of a broker. You need to configure the `conf/functions_worker.yml` file to customize your functions_worker.
Before you run Functions-worker 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).
+In this mode, most of the settings are already inherited from your broker configuration (for example, configurationStore settings, authentication settings, and so on) since `functions-worker` is running as part of the broker.
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` .
+- `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 larger 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.
@@ -52,7 +52,7 @@ And then you can use the following command to verify if `functions-worker` is ru
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.
+After entering the command above, a list of active function workers in the cluster is returned. The output is similar to the following.
```json
[{"workerId":"<worker-id>","workerHostname":"<worker-hostname>","port":8080}]
@@ -65,7 +65,7 @@ This section illustrates how to run `functions-worker` as a separate process in
> Note
-In this mode, make sure `functionsWorkerEnabled` is set to `false`, so you won't start `functions-worker` with brokers by mistake.
+> 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
@@ -73,7 +73,7 @@ To run function-worker separately, you have to configure the following parameter
#### Worker parameters
-- `workerId`: The type is string. It is unique across clusters, used to identify a worker machine.
+- `workerId`: The type is string. It is unique across clusters, which is 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.
@@ -122,9 +122,10 @@ For details on TLS encryption, refer to [Transport Encryption using TLS](securit
##### Enable Authentication Provider
-To enable authentication on Functions Worker, configure the following settings.
+To enable authentication on Functions Worker, you need to configure the following settings.
+
> Note
-Substitute the *providers list* with the providers you want to enable.
+> Substitute the *providers list* with the providers you want to enable.
```
authenticationEnabled: true
@@ -133,6 +134,7 @@ authenticationProviders: [ provider1, provider2 ]
For *TLS Authentication* provider, follow the example below to add the necessary settings.
See [TLS Authentication](security-tls-authentication.md) for more details.
+
```
brokerClientAuthenticationPlugin: org.apache.pulsar.client.impl.auth.AuthenticationTls
brokerClientAuthenticationParameters: tlsCertFile:/path/to/admin.cert.pem,tlsKeyFile:/path/to/admin.key-pk8.pem
@@ -150,7 +152,7 @@ properties:
saslJaasBrokerSectionName: Broker
```
-For *Token Authentication* prodivder, add necessary settings under `properties` if needed.
+For *Token Authentication* provider, add necessary settings for `properties` if needed.
See [Token Authentication](security-jwt.md) for more details.
```
properties:
@@ -180,7 +182,7 @@ superUserRoles:
#### BookKeeper Authentication
-If authentication is enabled on the BookKeeper cluster, you should configure the BookKeeper authentication settings as follows:
+If authentication is enabled on the BookKeeper cluster, you need configure the BookKeeper authentication settings as follows:
- `bookkeeperClientAuthenticationPlugin`: the plugin name of BookKeeper client authentication.
- `bookkeeperClientAuthenticationParametersName`: the plugin parameters name of BookKeeper client authentication.