You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2025/01/24 06:53:39 UTC
(flink) 01/02: [FLINK-37148][docs] Add documents for DataStream V2
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch release-2.0
in repository https://gitbox.apache.org/repos/asf/flink.git
commit eb598844a3c1bec997d2d9e6e8a8d9f6d7ac50ca
Author: Xu Huang <hu...@gmail.com>
AuthorDate: Thu Jan 23 10:47:06 2025 +0800
[FLINK-37148][docs] Add documents for DataStream V2
---
docs/content.zh/_index.md | 1 +
docs/content.zh/docs/dev/datastream-v2/_index.md | 23 ++
.../docs/dev/datastream-v2/building_blocks.md | 183 +++++++++
.../docs/dev/datastream-v2/builtin-funcs/_index.md | 23 ++
.../dev/datastream-v2/builtin-funcs/joining.md | 126 ++++++
.../dev/datastream-v2/builtin-funcs/windows.md | 421 ++++++++++++++++++++
.../datastream-v2/context_and_state_processing.md | 284 ++++++++++++++
docs/content.zh/docs/dev/datastream-v2/overview.md | 197 ++++++++++
.../dev/datastream-v2/time-processing/_index.md | 23 ++
.../time-processing/event_timer_service.md | 429 +++++++++++++++++++++
.../time-processing/processing_timer_service.md | 110 ++++++
.../content.zh/docs/dev/datastream-v2/watermark.md | 203 ++++++++++
docs/content/_index.md | 1 +
docs/content/docs/dev/datastream-v2/_index.md | 23 ++
.../docs/dev/datastream-v2/building_blocks.md | 183 +++++++++
.../docs/dev/datastream-v2/builtin-funcs/_index.md | 23 ++
.../dev/datastream-v2/builtin-funcs/joining.md | 126 ++++++
.../dev/datastream-v2/builtin-funcs/windows.md | 421 ++++++++++++++++++++
.../datastream-v2/context_and_state_processing.md | 284 ++++++++++++++
docs/content/docs/dev/datastream-v2/overview.md | 197 ++++++++++
.../dev/datastream-v2/time-processing/_index.md | 23 ++
.../time-processing/event_timer_service.md | 429 +++++++++++++++++++++
.../time-processing/processing_timer_service.md | 110 ++++++
docs/content/docs/dev/datastream-v2/watermark.md | 203 ++++++++++
.../fig/datastream/one-input-partitioning.png | Bin 0 -> 210417 bytes
.../fig/datastream/one-input-state-access.png | Bin 0 -> 176175 bytes
.../fig/datastream/two-input-state-access.png | Bin 0 -> 393766 bytes
27 files changed, 4046 insertions(+)
diff --git a/docs/content.zh/_index.md b/docs/content.zh/_index.md
index 6ffda5a03da..9cd2844f564 100644
--- a/docs/content.zh/_index.md
+++ b/docs/content.zh/_index.md
@@ -62,6 +62,7 @@ under the License.
{{< columns >}}
* [DataStream API]({{< ref "docs/dev/datastream/overview" >}})
+* [DataStream API (V2)]({{< ref "docs/dev/datastream-v2/overview" >}})
* [Table API & SQL]({{< ref "docs/dev/table/overview" >}})
* [Stateful Functions](https://nightlies.apache.org/flink/flink-statefun-docs-stable/)
diff --git a/docs/content.zh/docs/dev/datastream-v2/_index.md b/docs/content.zh/docs/dev/datastream-v2/_index.md
new file mode 100644
index 00000000000..ed1bb51bebc
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/_index.md
@@ -0,0 +1,23 @@
+---
+title: DataStream API (V2)
+bookCollapseSection: true
+weight: 3
+---
+<!--
+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.
+-->
diff --git a/docs/content.zh/docs/dev/datastream-v2/building_blocks.md b/docs/content.zh/docs/dev/datastream-v2/building_blocks.md
new file mode 100644
index 00000000000..d9e2e65c267
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/building_blocks.md
@@ -0,0 +1,183 @@
+---
+title: Building Blocks
+weight: 3
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Building Blocks
+
+DataStream, Partitioning, ProcessFunction are the most fundamental elements of DataStream API and respectively represent:
+
+- What are the types of data streams
+
+- How data is partitioned
+
+- How to perform operations / processing on data streams
+
+They are also the core parts of the fundamental primitives provided by DataStream API.
+
+## DataStream
+
+Data flows on the stream may be divided into multiple partitions. According to how the data is partitioned on the stream, we divide it into the following categories:
+
+- Global Stream: Force single partition/parallelism, and the correctness of data depends on this.
+
+- Partition Stream:
+ - Divide data into multiple partitions. State is only available within the partition. One partition can only be processed by one task, but one task can handle one or multiple partitions.
+According to the partitioning approach, it can be further divided into the following two categories:
+
+ - Keyed Partition Stream: Each key is a partition, and the partition to which the data belongs is deterministic.
+
+ - Non-Keyed Partition Stream: Each parallelism is a partition, and the partition to which the data belongs is nondeterministic.
+
+- Broadcast Stream: Each partition contains the same data.
+
+## Partitioning
+
+Above we defined the stream and how it is partitioned. The next topic to discuss is how to convert
+between different partition types. We call these transformations partitioning.
+For example non-keyed partition stream can be transformed to keyed partition stream via a `KeyBy` partitioning.
+
+```java
+NonKeyedPartitionStream<Tuple<Integer, String>> stream = ...;
+KeyedPartitionStream<Integer, String> keyedStream = stream.keyBy(record -> record.f0);
+```
+
+Overall, we have the following four partitioning:
+
+- KeyBy: Let all data be repartitioned according to specified key.
+
+- Shuffle: Repartition and shuffle all data.
+
+- Global: Merge all partitions into one.
+
+- Broadcast: Force partitions broadcast data to downstream.
+
+The specific transformation relationship is shown in the following table:
+
+{{< img src="/fig/datastream/one-input-partitioning.png" alt="one-input-partitioning" width="70%" >}}
+
+(A crossed box indicates that it is not supported or not required)
+
+One thing to note is: broadcast can only be used in conjunction with other inputs and cannot be directly converted to other streams.
+
+## ProcessFunction
+Once we have the data stream, we can apply operations on it. The operations that can be performed over
+DataStream are collectively called Process Function. It is the only entrypoint for defining all kinds
+of processing on the data streams.
+
+### Classification of ProcessFunction
+According to the number of input / output, they are classified as follows:
+
+| Partitioning | Number of Inputs | Number of Outputs |
+|:-----------------------------------------:|:----------------------:|:-------------------------------:|
+| OneInputStreamProcessFunction | 1 | 1 |
+| TwoInputNonBroadcastStreamProcessFunction | 2 | 1 |
+| TwoInputBroadcastStreamProcessFunction | 2 | 1 |
+| TwoOutputStreamProcessFunction | 1 | 2 |
+
+(Processing for more inputs and outputs can be achieved by combining multiple process functions)
+
+We have two types of two-input process function, depending on whether one of the input is broadcast stream.
+
+DataStream has series of `process` and `connectAndProcess` methods to transform the input stream or connect and transform two input streams via ProcessFunction.
+
+### Requirements for input and output streams
+
+The following two tables list the input and output stream combinations supported by OneInputStreamProcessFunction and TwoOutputStreamProcessFunction respectively.
+
+For OneInputStreamProcessFunction:
+
+| Input Stream | Output Stream |
+|:------------:|:----------------:|
+| Global | Global |
+| Keyed | Keyed / NonKeyed |
+| NonKeyed | NonKeyed |
+| Broadcast | Not Supported |
+
+For TwoOutputStreamProcessFunction:
+
+| Input Stream | Output Stream |
+|:------------:|:-------------------------------------:|
+| Global | Global + Global |
+| Keyed | Keyed + Keyed / Non-Keyed + Non-Keyed |
+| NonKeyed | NonKeyed + NonKeyed |
+| Broadcast | Not Supported |
+
+Things with two inputs is a little more complicated. The following table lists which streams are compatible with each other and the types of streams they output.
+
+A cross(❎) indicates not supported.
+
+| Output | Global | Keyed | NonKeyed | Broadcast |
+|:---------:|:------:|:------------------:|:--------:|:-----------------:|
+| Global | Global | ❎ | ❎ | ❎ |
+| Keyed | ❎ | NonKeyed / Keyed | ❎ | NonKeyed / Keyed |
+| NonKeyed | ❎ | ❎ | NonKeyed | NonKeyed |
+| Broadcast | ❎ | NonKeyed / Keyed | NonKeyed | ❎ |
+
+## Config Process
+
+After defining the process functions, you may want to make some configurations for the properties of this processing.
+For example, set the parallelism and name of the process operation, etc.
+
+The return value of `process`/`connectAndProcess` is both a stream and a handle that allowing us to configure the previous processing.
+It has a number of methods called `withXXX` to do the configuration. For example:
+
+```java
+inputStream
+ .process(func1) // do process 1
+ .withName("my-process-func") // configure name for process 1
+ .withParallelism(2) // configure parallelism for process 1
+ .process(func2) // do further process 2
+```
+
+## Example
+
+Here is an example to show how to use those building blocks to write a flink job.
+
+```java
+// create environment
+ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+// create a stream from source
+env.fromSource(someSource)
+ // map every element x to x + 1.
+ .process(new OneInputStreamProcessFunction<Integer, Integer>() {
+ @Override
+ public void processRecord(
+ Integer x,
+ Collector<Integer> output)
+ throws Exception {
+ output.collect(x + 1);
+ }
+ })
+ // If the sink does not support concurrent writes, we can force the stream to one partition
+ .global()
+ // sink the stream to some sink
+ .toSink(someSink);
+// execute the job
+env.execute()
+```
+
+{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/_index.md b/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/_index.md
new file mode 100644
index 00000000000..6493a6c5e5d
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/_index.md
@@ -0,0 +1,23 @@
+---
+title: Builtin Functions
+bookCollapseSection: true
+weight: 7
+---
+<!--
+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.
+-->
diff --git a/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/joining.md b/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/joining.md
new file mode 100644
index 00000000000..1640d158b9e
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/joining.md
@@ -0,0 +1,126 @@
+---
+title: "Joining"
+weight: 9
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Joining
+
+Join is used to merge two data streams by matching elements from both streams based on a common key,
+and performing calculations on the matched elements.
+
+This section will introduce the Join operation in DataStream in detail.
+
+Please note that currently, DataStream supports only non-window INNER joins.
+Other types of joins, such as interval joins, lookup joins, and window joins, will be supported in the future.
+
+## Non-Window Join
+
+To use the Join, users must specify the following:
+
+1. The left and right streams that will be joined.
+2. The `KeySelector` of two streams, which determines the join key.
+3. The [`JoinFunction`](#joinfunction): the processing logic applied to the matched data.
+
+We will first introduce the APIs for performing Join, and then introduce the JoinFunction.
+
+### APIs for performing Join
+
+There are two approaches to performing a Join operation:
+
+1. Use `BuiltinFuncs.join` to connect two streams and execute the Join.
+2. Convert the `JoinFunction` into a `ProcessFunction` and utilize `KeyedPartitionStream#connectAndProcess` to execute the Join.
+
+In the first approach, users can choose between two options based on the types of the input data streams.
+
+- If both input data streams are already Keyed Partition Stream, users can directly combine the two Keyed Partition
+Stream along with the `JoinFunction` to produce the joined data stream.
+
+```java
+NonKeyedPartitionStream joinedStream = BuiltinFuncs.join(
+ keyedStream1,
+ keyedStream2,
+ new CustomJoinFunction()
+);
+```
+
+- If both input data streams are Non-Keyed Partition Stream, users need to incorporate the corresponding
+Join key `KeySelector` and `JoinFunction` to convert the two Non-Keyed Partition Stream into the joined data stream.
+
+```java
+NonKeyedPartitionStream joinedStream = BuiltinFuncs.join(
+ stream1,
+ new CustomJoinKeySelector1(),
+ stream2,
+ new CustomJoinKeySelector2(),
+ new CustomJoinFunction()
+);
+```
+
+In the second approach, users can convert the `JoinFunction` into a `ProcessFunction`, which can then be processed by
+the DataStream API. An example of this conversion and how to use the converted `ProcessFunction` is provided below:
+
+```java
+TwoInputNonBroadcastStreamProcessFunction wrappedJoinFunction = BuiltinFuncs.join(new CustomJoinFunction());
+NonKeyedPartitionStream joinedStream = keyedStream1.connectAndProcess(
+ keyedstream2,
+ wrappedJoinFunction
+);
+```
+
+Please note that in the second approach, it is essential that the two input data streams are Keyed Partition Stream.
+
+
+### JoinFunction
+
+`JoinFunction` is an interface used to describe how to calculate the matched data.
+It has only one method `processRecord`. Users can get the matched elements in `processRecord`
+to perform calculations and then output the calculation results.
+
+Below is an example demonstrating how to use a `JoinFunction` to connect student personal
+information with their exam scores.
+
+```java
+class JoinStudentInformationAndScore
+ implements JoinFunction<StudentInfo, ExamScore, EnrichedStudentExamScore> {
+
+ @Override
+ public void processRecord(
+ StudentInfo studentInfo,
+ ExamScore examScore,
+ Collector<EnrichedStudentExamScore> output,
+ RuntimeContext ctx)
+ throws Exception {
+ // do some calculation logic and emit joined result
+ EnrichedStudentExamScore studentExamScore =
+ new EnrichedStudentExamScore(studentInfo.getId(), studentInfo.getName(), examScore.getScore());
+ output.collect(studentExamScore);
+ }
+}
+```
+
+
+{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/windows.md b/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/windows.md
new file mode 100644
index 00000000000..008216c4919
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/builtin-funcs/windows.md
@@ -0,0 +1,421 @@
+---
+title: "Windows"
+weight: 8
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Windows
+
+Windows are at the heart of processing infinite streams. Windows split the stream into "buckets"
+of finite size, over which we can apply computations. This document focuses on how windowing is
+performed in Flink DataStream and how the programmer can benefit to the maximum from its offered
+functionality.
+
+To utilize the Window functionality provided by DataStream API, users need to complete the following three steps:
+1. Declare Window: Specify the type of window you wish to use.
+2. Define `WindowProcessFunction`: Outline the logic that should be executed at various stages of the Window's lifecycle.
+3. Combine the Window Declaration and `WindowProcessFunction` to build a `ProcessFunction`
+ - Encapsulate the window declaration and the `WindowProcessFunction` into a single
+ `ProcessFunction`, which can then be leveraged within the DataStream API.
+
+This section will provide a comprehensive overview of these three steps and include an example
+illustrating how to use Window in DataStream API.
+
+## Declare Window
+
+Users should first determine which type of window to use for their applications. We currently offer
+three built-in window types: Time Window, Session Window and Global Window.
+
+### Time Window
+
+Time Windows are divided into multiple windows based on time ranges, and data is allocated to the
+corresponding window according to its timestamp. We support two types of Time Windows:
+tumbling windows and sliding windows. The time semantics within these windows can be classified
+as event time or processing time.
+
+Please note that Time Windows are currently supported only in Keyed Partition Stream.
+
+#### Tumbling Window
+
+*Tumbling Window* assigns each element to a window of a specified *window size*.
+Tumbling windows have a fixed size and do not overlap. For example, if you specify a tumbling
+window with a size of 5 minutes, the current window will be evaluated and a new window will be
+started every five minutes as illustrated by the following figure.
+
+{{< img src="/fig/tumbling-windows.svg" alt="Tumbling Windows" >}}
+
+The following code snippets show how to use tumbling windows.
+
+```java
+// create a tumbling window strategy with a window size of 60 seconds
+WindowStrategy windowStrategy = WindowStrategy.tumbling(Duration.ofSeconds(60), WindowStrategy.EVENT_TIME);
+```
+
+Time intervals can be specified by using one of `Duration.ofMillis(x)`, `Duration.ofSeconds(x)`,
+`Duration.ofMinutes(x)`, and so on.
+
+#### Sliding Window
+
+*Sliding Window* assigns elements to windows of fixed length. Similar to a tumbling
+window, the size of the windows is configured by the *window size* parameter.
+An additional *window slide* parameter controls how frequently a sliding window is started. Hence,
+sliding windows can be overlapping if the slide is smaller than the window size.
+In this case elements are assigned to multiple windows.
+
+For example, you could have windows of size 10 minutes that slides by 5 minutes. With this you
+get every 5 minutes a window that contains the events that arrived during the last 10 minutes
+as depicted by the following figure.
+
+{{< img src="/fig/sliding-windows.svg" alt="sliding windows" >}}
+
+The following code snippets show how to use sliding windows.
+
+```java
+// create a sliding window strategy with a window size of 60 seconds and a slide of 30 seconds
+WindowStrategy windowStrategy = WindowStrategy.sliding(Duration.ofSeconds(60), Duration.ofSeconds(30), WindowStrategy.PROCESSING_TIME);
+```
+
+Time intervals can be specified by using one of `Duration.ofMillis(x)`, `Duration.ofSeconds(x)`,
+`Duration.ofMinutes(x)`, and so on.
+
+#### Allowed Lateness
+
+When working with *event-time* windowing, it can happen that elements arrive late, *i.e.* the
+event time watermark that Flink uses to keep track of the progress of event-time is already past
+the end timestamp of a window to which an element belongs. See
+[event time]({{< ref "docs/dev/datastream-v2/time-processing/event_timer_service" >}}) for
+a more thorough discussion of how Flink deals with event time.
+
+By default, late elements are dropped when the event time watermark is past the end of the window.
+However, Flink allows to specify a maximum *allowed lateness* for window operators. Allowed lateness
+specifies by how much time elements can be late before they are dropped, and its default value is 0.
+
+Elements that arrive after the event time watermark has passed the end of the window, but before
+it passes the end of the window plus the allowed lateness, are still added to the window.
+The late but not dropped element will cause the window to fire again, and the late and dropped
+element will be processed by the [WindowProcessFunction#onLateRecord](#introduction-to-windowprocessfunction).
+
+In order to make this work, Flink keeps the state of windows until their allowed lateness expires.
+Once this happens, Flink removes the window and deletes its state, as also described in
+the [Window Lifecycle](#window-lifecycle) section.
+
+By default, the allowed lateness is set to `0`. That is, elements that arrive behind the watermark
+will be dropped.
+
+You can specify an allowed lateness like this:
+
+```java
+// create a sliding window strategy with a window size of 60 seconds and a slide of 30 seconds and an allowed lateness of 10 seconds
+WindowStrategy windowStrategy = WindowStrategy.sliding(Duration.ofSeconds(60), Duration.ofSeconds(30), WindowStrategy.PROCESSING_TIME, Duration.ofSeconds(10));
+```
+
+### Session Window
+
+The *session windows* groups elements by sessions of activity. Session windows do not overlap and
+do not have a fixed start and end time, in contrast to *tumbling windows* and *sliding windows*.
+Instead, a session window closes when it does not receive elements for a certain period of
+time, *i.e.*, when a gap of inactivity occurred. A session window can be configured
+with *session gap*. When this period expires, the current session closes and subsequent elements
+are assigned to a new session window.
+
+{{< img src="/fig/session-windows.svg" alt="session windows" >}}
+
+The following code snippets show how to declare session windows.
+
+```java
+// create a session window strategy with a session gap of 60 seconds
+WindowStrategy windowStrategy = WindowStrategy.session(Duration.ofSeconds(60), WindowStrategy.EVENT_TIME);
+```
+
+Please note that Session Windows are supported only in Global Stream and Keyed Partition Stream.
+
+### Global Window
+
+*Global Window* refers to a scenario where all elements are assigned to a single, unified Window.
+This windowing scheme is particularly useful in bounded stream scenarios, as it can be triggered
+once all inputs have concluded. Note that it is compatible with Global Stream, Keyed Partition Stream,
+and Non-Keyed Partition Stream.
+
+{{< img src="/fig/non-windowed.svg" alt="global windows" >}}
+
+The following code snippets show how to declare a global window.
+
+```java
+// create a global window strategy
+WindowStrategy windowStrategy = WindowStrategy.global();
+```
+
+## Define WindowProcessFunction
+### Window Lifecycle
+In a nutshell, a window is **created** as soon as the first element that should belong to this
+window arrives, and the window is **completely removed** when the time (event or processing time)
+passes its end timestamp plus the user-specified `allowed lateness` (see [Allowed Lateness](#allowed-lateness)).
+For example, with an event-time-based windowing that creates non-overlapping (or tumbling) windows
+every 5 minutes and has an allowed lateness of 1 min, Flink will create a new window for the
+interval between `12:00` and `12:05` when the first element with a timestamp that falls into this
+interval arrives, and it will remove it when the event time watermark passes the `12:06` timestamp.
+
+Flink abstracts the essential actions within the window lifecycle into methods in
+the [WindowProcessFunction](#introduction-to-windowprocessfunction). Users are required to implement
+their own *WindowProcessFunction* to define their specific windowing computation logic.
+
+### Introduction to WindowProcessFunction
+
+The `WindowProcessFunction` is the main component that users need to implement to process the
+data in the window. After declaring a window, users need to define the operational logic for various
+stages of the window's lifecycle, called *WindowProcessFunction*.
+
+Here is the interface of `OneInputWindowStreamProcessFunction`:
+
+```java
+/**
+ * A type of {@link WindowProcessFunction} for one-input window processing.
+ *
+ * @param <IN> The type of the input value.
+ * @param <OUT> The type of the output value.
+ */
+@Experimental
+public interface OneInputWindowStreamProcessFunction<IN, OUT> extends WindowProcessFunction {
+
+ /**
+ * This method will be invoked when a record is received. Its default behaviors to store data in
+ * built-in window state by {@link OneInputWindowContext#putRecord}. If the user overrides this
+ * method, they have to take care of the input data themselves.
+ */
+ default void onRecord(
+ IN record,
+ Collector<OUT> output,
+ PartitionedContext<OUT> ctx,
+ OneInputWindowContext<IN> windowContext)
+ throws Exception {
+ windowContext.putRecord(record);
+ }
+
+ /**
+ * This method will be invoked when the Window is triggered, you can obtain all the input
+ * records in the Window by {@link OneInputWindowContext#getAllRecords()}.
+ */
+ void onTrigger(
+ Collector<OUT> output,
+ PartitionedContext<OUT> ctx,
+ OneInputWindowContext<IN> windowContext)
+ throws Exception;
+
+ /**
+ * Callback when a window is about to be cleaned up. It is the time to deletes any state in the
+ * {@code windowContext} when the Window expires (the event time or processing time passes its
+ * {@code maxTimestamp} + {@code allowedLateness}).
+ */
+ default void onClear(
+ Collector<OUT> output,
+ PartitionedContext<OUT> ctx,
+ OneInputWindowContext<IN> windowContext)
+ throws Exception {}
+
+ /** This method will be invoked when a record is received after the window has been cleaned. */
+ default void onLateRecord(IN record, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {}
+}
+```
+
+There are four key methods in the WindowProcessFunction. The names and meanings of these methods are as follows:
+1. `onRecord`: `onRecord` indicates that the window has received a record.
+2. `onTrigger`: `onTrigger` indicates that the window has been triggered.
+3. `onClear`: `onClear` indicates that the window has been cleared.
+4. `onLateRecord`: `onLateRecord` indicates that the window has received record after the window is cleared.
+
+There are some important points to consider regarding these methods.
+1. Windows can be triggered multiple times. Therefore, `onRecord` may be called after `onTrigger`.
+2. GlobalWindow is cleared when the data stream ends, while time/session windows are cleared after the window boundary is reached and the allowedLateness (see [Allowed Lateness](#allowed-lateness)) has elapsed.
+3. `onLateRecord` method is not possible to access the window state since the window has been cleared.
+
+Users should implement `WindowProcessFunction` to complete their own calculation logic.
+In this step, they can use [Window State](#declare-and-use-window-state) to store Window-related data and [access the built-in state](#access-built-state-of-window) used to store Window data.
+
+### Declare and Use Window State
+In a window, there are two types of states: partitioned state and window state.
+
+1. Partitioned State: We refer to the partition-related state as partitioned state.
+
+ For NonKeyedStream, this state is shared among a specific task. For KeyedStream, this state is shared among data with the same key.
+
+ User can declare partitioned state through `ProcessFunction#usesStates` and use partitioned state through `PartitionedContext#getStateManager`.
+It's users' responsibility to clear data in partitioned state that are no longer needed in `WindowProcessFunction#onClear`.
+2. Window State: We refer to the window-related state as window state.
+
+ Window state is bound to a specific window. For example, the window state declared and used for the same key in the 10:00-11:00 window is different from that in the 11:00-12:00 window.
+
+ User can declare window state through `WindowProcessFunction#usesWindowStates` and use window state through `WindowContext#getWindowState`.
+All window state will eventually be cleared by framework, whether or not the user clears it manually in `WindowProcessFunction#onClear`.
+
+### Access Built State of Window
+We provide built-in window state for each window to store the input data. Users can access this through `WindowContext#putRecord` and `WindowContext#getAllRecords`.
+This state will be cleared when the window is cleared.
+
+By default, `WindowProcessFunction#onRecord` stores the received data in the window's built-in state by `WindowContext#putRecord`, and users can retrieve all the data within the Window using `WindowContext#getAllRecords` when the window is triggered.
+
+Therefore, when overriding `WindowProcessFunction#onRecord`, users should consider whether they need to write the input data into the built-in state.
+A typical example is if users want to do pre-aggregation, they can declare a Window state, perform aggregation in `WindowProcessFunction#onRecord`, update the aggregated window state, and output the final result in `WindowProcessFunction#onTrigger`.
+Therefore, unnecessary cost of caching all data are eliminated.
+
+## Build a ProcessFunction
+After declaring the Window and defining the *WindowProcessFunction*, users should utilize the *BuiltinFuncs.window* method to encapsulate these two components into a *ProcessFunction*, which can then be integrated into the data processing stream.
+An example of this is as follows:
+
+```java
+KeyedPartitionStream stream = ...;
+OneInputStreamProcessFunction wrappedWindowProcessFunction = BuiltinFuncs.window(windowStrategy, new CustomWindowProcessFunction());
+stream.process(wrappedWindowProcessFunction)
+ .process(...);
+```
+
+By doing so, Flink will automatically manage the state and timers required for caching window data, relieving users of the burden of handling these details on their own.
+
+## Example: Count the sales of each product every hour
+
+The following is an example of how to use Window to count the sales of each product in each hour.
+
+In the following example, we first use the [event time extension]({{< ref "docs/dev/datastream-v2/time-processing/event_timer_service" >}}) to extract the event time from `orderSoucrce`, then declare a one-hour tumbling window and calculate the sales quantity of each product when the window is triggered.
+
+```java
+public class CountProductSalesEveryHour {
+
+ public static class Order {
+ public long orderId;
+ public long productId;
+ public long salesQuantity;
+ public long orderTime;
+ }
+
+ public static void main(String[] args) throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+ // create order source stream
+ NonKeyedPartitionStream<Order> orderSource = ...;
+
+ // extract and propagate event time from order
+ NonKeyedPartitionStream<Order> orderStream = orderSource.process(
+ EventTimeExtension
+ .<Order>newWatermarkGeneratorBuilder(order -> order.orderTime)
+ .periodicWatermark(Duration.ofMillis(200))
+ .buildAsProcessFunction()
+ );
+
+ NonKeyedPartitionStream<Tuple2<Long, Long>> productSalesQuantityStream = orderStream
+ // key by productId
+ .keyBy(order -> order.productId)
+ .process(BuiltinFuncs.window(
+ // declare tumbling window with window size 10 seconds
+ WindowStrategy.tumbling(
+ Duration.ofHours(1),
+ WindowStrategy.EVENT_TIME),
+ // define window process function to calculate total sales quantity per product per window
+ new CountSalesQuantity()
+ )
+ );
+
+ // print result
+ productSalesQuantityStream.toSink(new WrappedSink<>(new PrintSink<>()));
+
+ env.execute("CountSalesQuantifyOfEachProductEveryHour");
+ }
+
+ public static class CountSalesQuantity implements OneInputWindowStreamProcessFunction<Order, Tuple2<Long, Long>> {
+
+ @Override
+ public void onTrigger(
+ Collector<Tuple2<Long, Long>> output,
+ PartitionedContext<Tuple2<Long, Long>> ctx,
+ OneInputWindowContext<Order> windowContext) throws Exception {
+ // get current productId
+ long productId = ctx.getStateManager().getCurrentKey();
+ // calculate total sales quantity
+ long totalSalesQuantity = 0;
+ for (Order order : windowContext.getAllRecords()) {
+ totalSalesQuantity += order.salesQuantity;
+ }
+ // emit result
+ output.collect(Tuple2.of(productId, totalSalesQuantity));
+ }
+ }
+}
+```
+
+In the scenario of this example, users can reduce the cost of Window calculation by performing pre-aggregation in the Window.
+We rewrote the `CountSalesQuantity` WindowProcessFunction into the `CountSalesQuantityWithPreAggregation` WindowProcessFunction that can perform pre-aggregation when the input data arrives.
+
+```java
+
+public static class CountSalesQuantityWithPreAggregation implements OneInputWindowStreamProcessFunction<Order, Tuple2<Long, Long>> {
+
+ private final ValueStateDeclaration<Long> salesQuantityStateDeclaration =
+ StateDeclarations.valueState("totalSalesQuantity", TypeDescriptors.LONG);
+
+ @Override
+ public Set<StateDeclaration> useWindowStates() {
+ return Set.of(salesQuantityStateDeclaration);
+ }
+
+ @Override
+ public void onRecord(
+ Order record,
+ Collector<Tuple2<Long, Long>> output,
+ PartitionedContext<Tuple2<Long, Long>> ctx,
+ OneInputWindowContext<Order> windowContext) throws Exception {
+ // get sales quantity from state
+ ValueState<Long> salesQuantityState = windowContext.getWindowState(salesQuantityStateDeclaration).get();
+ long salesQuantity = 0;
+ if (salesQuantityState.value() != null) {
+ salesQuantity = salesQuantityState.value();
+ }
+
+ // update sales quantity in state
+ salesQuantity += record.salesQuantity;
+ salesQuantityState.update(salesQuantity);
+ }
+
+ @Override
+ public void onTrigger(
+ Collector<Tuple2<Long, Long>> output,
+ PartitionedContext<Tuple2<Long, Long>> ctx,
+ OneInputWindowContext<Order> windowContext) throws Exception {
+ // get current productId
+ long productId = ctx.getStateManager().getCurrentKey();
+ // get sales quantity from state
+ ValueState<Long> salesQuantityState = windowContext.getWindowState(salesQuantityStateDeclaration).get();
+ long salesQuantity = salesQuantityState.value() == null ? 0 : salesQuantityState.value();
+ // emit result
+ output.collect(Tuple2.of(productId, salesQuantity));
+ }
+}
+```
+
+In the `CountSalesQuantityWithPreAggregation` function, we first declare a `ValueState` to store the total sales quantity for each product within the window.
+When input data arrives, we update this state accordingly, and we output the final result once the window is triggered.
+This approach allows us to avoid storing all input data in the window; instead, we only need to maintain a `ValueState` for each product within the window.
+
+
+{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream-v2/context_and_state_processing.md b/docs/content.zh/docs/dev/datastream-v2/context_and_state_processing.md
new file mode 100644
index 00000000000..cd17fe26d86
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/context_and_state_processing.md
@@ -0,0 +1,284 @@
+---
+title: Context and State Processing
+weight: 4
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Context and State Processing
+
+## Context
+
+Unlike attributes like the name of process operation, some information(such as current key) can only be obtained when the process function is executed.
+In order to build a bridge between process functions and the execution engine, DataStream API provide a unified entrypoint called Runtime Context.
+
+We divide all contextual into multiple parts according to their functions:
+
+- JobInfo: Hold all job related information, such as job name, execution mode.
+
+- TaskInfo: Hold all task related information, such as parallelism.
+
+- MetricGroup: Manage context related to metrics, such as registering a metric.
+
+- State Manager: Manage context related to state, such as accessing a specific state.
+
+- Watermark Manager: Manage context related to watermark, such as triggering a watermark.
+
+- ProcessingTime Manager: Manage context related to processing timer, such as getting the current processing time.
+
+Those context information can be classified into two categories:
+
+* NonPartitionedContext: Include JobInfo, TaskInfo, MetricGroup, WatermarkManager.
+* PartitionedContext: Include StateManager, ProcessingTimeManager.
+
+Typically, `PartitionedContext` is provided to a `ProcessFunction` when processing of data within
+a specific partition is expected. E.g., upon receiving of new records or triggering of timers.
+On the other hand, `NonPartitionedContext` is provided when no specific partition is relevant.
+E.g., initiating or cleaning-up of the `ProcessFunction`.
+
+The following code snippet shows how to get the parallelism and execution mode of the process function:
+
+```java
+new OneInputStreamProcessFunction<String, String>(){
+ private transient int parallelism;
+
+ private transient ExecutionMode executionMode;
+ @Override
+ public void open(NonPartitionedContext<String> ctx) throws Exception {
+ parallelism = ctx.getTaskInfo().getParallelism();
+ executionMode = ctx.getJobInfo().getExecutionMode();
+ }
+}
+```
+
+## State Processing
+
+State is the foundation of stateful computation. DataStream API supports declaring state in Process Function and
+accessing and updating state during data processing.
+
+In general, you should follow the principle of "declare first, use later.". In general,
+writing a stateful process function is divided into three steps:
+
+1. Defining the state in the form of `StateDeclaration`.
+2. Declaring the state in `ProcessFunction#usesStates`.
+3. Getting and updating state via `StateManager`.
+
+Before we go any further, let's look at what a stateful process function looks like:
+
+```java
+private static class StatefulFunction implements OneInputStreamProcessFunction<Long, Long> {
+ // Step1: Defining the state in the form of `StateDeclaration`.
+ static final StateDeclaration.ListStateDeclaration<Long> LIST_STATE_DECLARATION =
+ StateDeclarations.listStateBuilder("example-list-state", TypeDescriptors.LONG).build();
+
+ // Step2: Declaring the state in `ProcessFunction#usesStates`
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ return Collections.singleton(LIST_STATE_DECLARATION);
+ }
+
+ @Override
+ public void processRecord(Long record, Collector<Long> output, RuntimeContext ctx)
+ throws Exception {
+ // Step3: Getting and updating state via `StateManager`.
+ Optional<ListState<Long>> stateOptional =
+ ctx.getStateManager().getState(LIST_STATE_DECLARATION);
+ ListState<Long> state = stateOptional.get();
+ // do something with this state. For example, update the state by this record.
+ state.update(Collections.singletonList(record));
+ }
+}
+```
+
+### Define State
+
+StateDeclaration is used to define a specific state, two types of information need to be provided for a state declaration:
+
+- Name: Used as the unique identifier of the state.
+
+- RedistributionMode: Defines how the state is redistributed between different partitions. For the keyed partition stream,
+since the state is bounded within the partition, no redistribution is required. But for the non-keyed partition stream,
+the partition will change with the parallelism, so the state must define how to redistribute.
+
+There are three RedistributionMode to choose:
+- NONE: Not supports redistribution.
+- REDISTRIBUTABLE: This state can be safely redistributed between different partitions, and the
+specific redistribution strategy is determined by the state itself.
+- IDENTICAL: States are guaranteed to be identical in different partitions, thus redistribution is
+not a problem.
+
+There are currently five type of `StateDeclaration`: `ValueStateDeclaration`, `ListStateDeclaration`, `MapStateDeclaration`, `ReducingStateDeclaration`, `AggregatingStateDeclaration` and `BroadcastStateDeclaration`
+that describe the `ValueState`, `ListState`, `MapState`, `ReducingState`, `AggregatingState` and `BroadcastState` respectively.
+
+* `ValueState<T>`: This keeps a value that can be updated and
+ retrieved. The value can be set using `update(T)` and retrieved using `T value()`.
+
+* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `Iterable`
+ over all currently stored elements. Elements are added using `add(T)` or `addAll(List<T>)`, the Iterable can
+ be retrieved using `Iterable<T> get()`. You can also override the existing list with `update(List<T>)`
+
+* `ReducingState<T>`: This keeps a single value that represents the aggregation of all values
+ added to the state. The interface is similar to `ListState` but elements added using
+ `add(T)` are reduced to an aggregate using a specified `ReduceFunction`.
+
+* `AggregatingState<IN, OUT>`: This keeps a single value that represents the aggregation of all values
+ added to the state. Contrary to `ReducingState`, the aggregate type may be different from the type
+ of elements that are added to the state. The interface is the same as for `ListState` but elements
+ added using `add(IN)` are aggregated using a specified `AggregateFunction`.
+
+* `MapState<UK, UV>`: This keeps a list of mappings. You can put key-value pairs into the state and
+ retrieve an `Iterable` over all currently stored mappings. Mappings are added using `put(UK, UV)` or
+ `putAll(Map<UK, UV>)`. The value associated with a user key can be retrieved using `get(UK)`. The iterable
+ views for mappings, keys and values can be retrieved using `entries()`, `keys()` and `values()` respectively.
+ You can also use `isEmpty()` to check whether this map contains any key-value mappings.
+
+* `BroadcastState<K, V>`: This can be created to store the state of a `BroadcastStream`. This state
+assumes that the same elements are sent to all instances of a process function. You can put key-value pairs into the state and
+retrieve an `Iterable` over all currently stored mappings.
+
+For ease of use, DataStream API provide an auxiliary class called `StateDeclarations`, which encapsulates a series of methods for creating various StateDeclaration instances.
+
+For example:
+
+```java
+// create a value state declaration
+ValueStateDeclaration<Integer> valueStateDeclaration = StateDeclarations.valueState("example-value-state", TypeDescriptors.LONG);
+
+// create a map state declaration
+MapStateDeclaration<Long, String> mapStateDeclaration = StateDeclarations.mapState(
+ "example-map-state", TypeDescriptors.LONG, TypeDescriptors.STRING);
+
+// create a reducing state declaration with a `sum` reduce function
+ReducingStateDeclaration<Long> reducingStateDeclaration =
+ StateDeclarations.reducingState("example-reducing-state", TypeDescriptors.LONG, Long::sum);
+```
+
+To provide the type information of state, in the above example, the `TypeDescriptors` class provides a set of pre-defined type descriptors for common types, such as `INT`, `LONG`, `BOOLEAN`, `STRING`, `LIST`, `MAP`, etc.
+If your type is not in the pre-defined list, you can provide a class implements `TypeDescriptor` interface to define your own type descriptor.
+
+### Declare State
+
+As we said earlier, when using state in the DataStream API, you must follow the principle of "declare first, use later.".
+`ProcessFunction` has a method called `useStates` to explicitly declares states upfront.
+
+```java
+default Set<StateDeclaration> usesStates() {
+ return Collections.emptySet();
+}
+```
+
+For stateful functions, this method must be overridden and each specific state must be declared in this method before it can be used. This gives
+flink more opportunities to optimize the execution of functions.
+
+### Get and Update State
+
+The `StateManager` is used to manage the state for the process function. It provides methods for getting state, as well as method for retrieving current key.
+
+```java
+public interface StateManager {
+ /**
+ * Get the key of current record.
+ *
+ * @return The key of current processed record.
+ * @throws UnsupportedOperationException if the key can not be extracted for this function, for
+ * instance, get the key from a non-keyed partition stream.
+ */
+ <K> K getCurrentKey() throws UnsupportedOperationException;
+
+ /**
+ * Get the specific list state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the list state corresponds to the state declaration.
+ */
+ <T> Optional<ListState<T>> getState(ListStateDeclaration<T> stateDeclaration) throws Exception;
+
+ /**
+ * Get the specific value state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the value state corresponds to the state declaration.
+ */
+ <T> Optional<ValueState<T>> getState(ValueStateDeclaration<T> stateDeclaration)
+ throws Exception;
+
+ /**
+ * Get the specific map state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the map state corresponds to the state declaration.
+ */
+ <K, V> Optional<MapState<K, V>> getState(MapStateDeclaration<K, V> stateDeclaration)
+ throws Exception;
+
+ /**
+ * Get the specific reducing state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the reducing state corresponds to the state declaration.
+ */
+ <T> Optional<ReducingState<T>> getState(ReducingStateDeclaration<T> stateDeclaration)
+ throws Exception;
+
+ /**
+ * Get the specific aggregating state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the aggregating state corresponds to the state declaration.
+ */
+ <IN, ACC, OUT> Optional<AggregatingState<IN, OUT>> getState(
+ AggregatingStateDeclaration<IN, ACC, OUT> stateDeclaration) throws Exception;
+
+ /**
+ * Get the specific broadcast state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the broadcast state corresponds to the state declaration.
+ */
+ <K, V> Optional<BroadcastState<K, V>> getState(BroadcastStateDeclaration<K, V> stateDeclaration)
+ throws Exception;
+}
+```
+
+You can get the `StateManager` from anywhere you have access to the `PartitionedContext` using `context#getStateManager()`.
+Next, using the `getState` method to obtain the state you declared, and then read and updated it.
+
+
+### The Legitimacy of State Declaration and Access
+
+In `ProcessFunction`, not all state declarations and accesses are legal, depending on the specific type of input streams.
+
+For `OneInputStreamProcessFunction` and `TwoOutputStreamProcessFunction`, the legality of state declaration and access are listed in the following table:
+
+{{< img src="/fig/datastream/one-input-state-access.png" alt="one-input-state-access" width="40%" >}}
+
+For the two inputs case, things get a little more complicated. Since different inputs may come from different type of streams, the legality of state access is also different for the data processing functions of each input edge.
+The specific rules are shown in the following table:
+
+{{< img src="/fig/datastream/two-input-state-access.png" alt="two-input-state-access" width="70%" >}}
+(K, NK, G, B denote the shorthand for Keyed, Non-Keyed, Global, Broadcast Stream, respectively)
+
+flink will check the legality of state declaration in advance. If an illegal state is declared in `usesStates` method, exception will be thrown at job compile time.
+
+{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream-v2/overview.md b/docs/content.zh/docs/dev/datastream-v2/overview.md
new file mode 100644
index 00000000000..d4f05c2a940
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/overview.md
@@ -0,0 +1,197 @@
+---
+title: Overview
+weight: 2
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Flink DataStream API Programming Guide
+
+DataStream programs in Flink are regular programs that implement transformations on data streams
+(e.g., filtering, updating state, defining windows, aggregating). The data streams are initially created from various
+sources (e.g., message queues, socket streams, files). Results are returned via sinks, which may for
+example write the data to files, or to standard output (for example the command line
+terminal). Flink programs run in a variety of contexts, standalone, or embedded in other programs.
+The execution can happen in a local JVM, or on clusters of many machines.
+
+## What is a DataStream?
+
+The DataStream API gets its name from the special `DataStream` class that is
+used to represent a collection of data in a Flink program. You can think of
+them as immutable collections of data that can contain duplicates. This data
+can either be finite or unbounded, the API that you use to work on them is the
+same.
+
+A `DataStream` is similar to a regular Java `Collection` in terms of usage but
+is quite different in some key ways. They are immutable, meaning that once they
+are created you cannot add or remove elements. You can also not simply inspect
+the elements inside but only work on them using the `DataStream` API
+operations, which are also called transformations.
+
+You can create an initial `DataStream` by adding a source in a Flink program.
+Then you can derive new streams from this and combine them by using API methods
+such as `process`, `connectAndProcess`, and so on.
+
+## Fundamental Primitives and Extensions
+
+Based on whether its functionality must be provided by flink, we divide the relevant concepts in DataStream API
+into two categories: fundamental primitives and high-level extensions.
+
+### Fundamental primitives
+
+Fundamental primitives are the basic and necessary semantics that flink need to provide in order to
+define a stateful stream processing application, which cannot be achieved by users if not provided by
+the framework. It includes data stream, partitioning, process function, state, processing timer service,
+watermark.
+
+More details can be found in:
+- [Building Blocks]({{< ref "docs/dev/datastream-v2/building_blocks" >}}): Given the most basic elements of DataStream API.
+- [State Processing]({{< ref "docs/dev/datastream-v2/context_and_state_processing" >}}): Explanation of how to develop stateful applications.
+- [Time Processing # Processing Timer Service]({{< ref "docs/dev/datastream-v2/time-processing/processing_timer_service" >}}): Explanation of how to handle processing time.
+- [Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}}): Explanation of how to define and handle user defined events.
+
+### High-Level Extensions
+
+High-Level extensions are like short-cuts / sugars, without which users can probably still achieve the same
+behavior by working with the fundamental APIs, but would be a lot easier with the builtin supports.
+It includes event timer service, window and join.
+
+More details can be found in:
+- [Time Processing # Event Timer Service]({{< ref "docs/dev/datastream-v2/time-processing/event_timer_service" >}}): Explanation of how to handle event time via extension.
+- [Builtin Functions]({{< ref "docs/dev/datastream-v2/builtin-funcs/windows" >}}): Explanation of how to do window aggregation and join via extension.
+
+## Anatomy of a Flink DataStream Program
+
+Flink programs look like regular programs that transform `DataStream`. Each
+program consists of the same basic parts:
+
+1. Obtain an `Execution Environment`,
+2. Load/create the initial data,
+3. Specify transformations on this data,
+4. Specify where to put the results of your computations,
+5. Trigger the program execution
+
+### Obtain an `Execution Environment`
+
+We will now give an overview of each of those steps, please refer to the
+respective sections for more details.
+
+The `ExecutionEnvironment` is the basis for all Flink programs. You can
+obtain one using these static methods on `ExecutionEnvironment`:
+
+```java
+ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+```
+
+If you are executing your program inside an IDE or as a regular Java program it will create a local environment
+that will execute your program on your local machine. If you created a JAR file
+from your program, and invoke it through the command line, the Flink cluster manager will execute your main method and
+`ExecutionEnvironment.getInstance()` will return an execution environment for executing
+your program on a cluster.
+
+
+### Load/create the Initial Data
+
+Sources are where your program reads its input from. You can attach a source to your program by
+using `ExecutionEnvironment.fromSource(source, sourceName)`. Flink comes with a number of pre-implemented
+source, you can use [FLIP-27](https://cwiki.apache.org/confluence/x/co_zBQ) based source via `DataStreamV2SourceUtils.wrapSource(source)` or use
+`DataStreamV2SourceUtils.fromData(collection)` for testing/debugging purpose.
+
+As an example, to just read data from a predefined collection, you can use:
+
+```java
+ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+NonKeyedPartitionStream<String> input =
+ env.fromSource(
+ DataStreamV2SourceUtils.fromData(Arrays.asList("1", "2", "3")),
+ "source"
+ );
+```
+
+Since data from source has no clear partitioning, this will give you a `NonKeyedPartitionStream` on which you can then apply transformations to create new
+derived DataStreams. For other types of DataStream, see [Building Blocks # DataStream]({{< ref "docs/dev/datastream-v2/building_blocks" >}}#datastream).
+
+### Specify Transformations on this Data
+
+You apply transformations by calling methods on DataStream with a
+`ProcssFunction`(using lambda expression here for simplicity). For example, a map transformation looks like this:
+
+```java
+NonKeyedPartitionStream<String> input = ...;
+
+NonKeyedPartitionStream<Integer> parsed = source.process(
+ (OneInputStreamProcessFunction<String, Integer>)
+ (record, output, ctx) -> {
+ output.collect(Integer.parseInt(record));
+ });
+```
+
+This will create a new DataStream by converting every String in the original
+collection to an Integer. For more details of processing, see [Building Blocks # Process Function]({{< ref "docs/dev/datastream-v2/building_blocks" >}}#processfunction).
+
+### Specify Where to Put the Results of Your Computations
+
+Once you have a DataStream containing your final results, you can write it to
+an external system by creating a sink.
+
+Data sinks consume DataStreams and forward them to files, sockets, external systems, or print them.
+Flink comes with a variety of built-in sink implementations, you can use SinkV2 based sink via `DataStreamV2SinkUtils.wrapSink(sink)`.
+
+These are just an example for creating a sink to print the results:
+
+```java
+parsed.toSink(DataStreamV2SinkUtils.wrapSink(new PrintSink<>()));
+```
+
+### Trigger the Program Execution
+
+Once you specified the complete program you need to **trigger the program
+execution** by calling `execute()` on the `ExecutionEnvironment`.
+Depending on the type of the `ExecutionEnvironment` the execution will be
+triggered on your local machine or submit your program for execution on a
+cluster. The `execute()` method will wait for the job to finish.
+
+That last part about program execution is crucial to understanding when and how
+Flink operations are executed. All Flink programs are executed lazily: When the
+program's main method is executed, the data loading and transformations do not
+happen directly. Rather, each operation is created and added to a dataflow
+graph. The operations are actually executed when the execution is explicitly
+triggered by an `execute()` call on the execution environment. Whether the
+program is executed locally or on a cluster depends on the type of execution
+environment.
+
+The lazy evaluation lets you construct sophisticated programs that Flink
+executes as one holistically planned unit.
+
+Where to go next?
+-----------------
+
+* [Building Blocks]({{< ref "docs/dev/datastream-v2/building_blocks" >}})
+* [Context and State Processing]({{< ref "docs/dev/datastream-v2/context_and_state_processing" >}})
+* [Time Processing]({{< ref "docs/dev/datastream-v2/time-processing/processing_timer_service" >}})
+* [Builtin Functions]({{< ref "docs/dev/datastream-v2/builtin-funcs/windows" >}})
+* [Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}})
+
+{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream-v2/time-processing/_index.md b/docs/content.zh/docs/dev/datastream-v2/time-processing/_index.md
new file mode 100644
index 00000000000..28f0422cdc4
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/time-processing/_index.md
@@ -0,0 +1,23 @@
+---
+title: Time Processing
+bookCollapseSection: true
+weight: 5
+---
+<!--
+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.
+-->
diff --git a/docs/content.zh/docs/dev/datastream-v2/time-processing/event_timer_service.md b/docs/content.zh/docs/dev/datastream-v2/time-processing/event_timer_service.md
new file mode 100644
index 00000000000..16b5dc64d30
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/time-processing/event_timer_service.md
@@ -0,0 +1,429 @@
+---
+title: "Event Timer Service"
+weight: 9
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Event Timer Service
+
+The event timer service is a high-level extension of the Flink DataStream API, provided by Flink.
+It enables users to register timers for executing calculations at specific event time points and helps determine when to trigger windows within the Flink framework.
+
+For a comprehensive explanation of event time, please refer to the section on [Notions of Time: Event Time and Processing Time]({{< ref "docs/concepts/time" >}}#notions-of-time-event-time-and-processing-time).
+
+In this section, we will introduce how to utilize the event timer service within the Flink DataStream API.
+
+We use a special type of [Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}}) to denote the progression of event time in the stream,
+which we refer to as the event time watermark. Additionally, to dealing idle inputs or sources, we implement another type of
+[Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}}) to indicate that the input or source
+is idle.
+This is known as the idle status watermark. For more information, please refer to [Dealing With Idle Inputs / Sources](#dealing-with-idle-inputs--sources).
+In this document, we collectively refer to the event time watermark and the idle status watermark as event-time related watermarks.
+
+The core of the event timer service lies in generating and propagating event-time related watermarks through the streams.
+To achieve this, we need to address two aspects:
+1. How to generate event-time related watermarks.
+2. How to handle event-time related watermarks.
+
+The following will introduce these two aspects.
+
+## Generate Event-Time related Watermarks
+
+In order to work with *event time*, Flink needs to know the events
+*timestamps*, meaning each element in the stream needs to have its event
+timestamp *assigned*. This is usually done by accessing/extracting the
+timestamp from some field in the element.
+
+Once the timestamps have been extracted, Flink generates event time watermarks.
+There are two methods for doing so: one is to use the `EventTimeWatermarkGeneratorBuilder` provided by Flink,
+and the other is to implement a custom `ProcessFunction`.
+
+Users should select one of these two approaches to suit their needs.
+
+### Generate Watermarks by WatermarkGeneratorBuilder
+
+Users can utilize the `EventTimeWatermarkGeneratorBuilder` to generate event-time related watermarks.
+There are four aspects of the `EventTimeWatermarkGeneratorBuilder` that users can configure:
+
+1. [Required] `EventTimeExtractor`
+
+ This function instructs Flink on how to extract the event time from each record.
+
+3. [Optional] Input Idle Timeout
+
+ If the input stream remains idle for a specified duration, Flink will ignore this input when
+combining event-time related watermarks to prevent stalling the progression of event time. The default value for this timeout is 0.
+
+ For more information, please refer to the [Dealing With Idle Inputs / Sources](#dealing-with-idle-inputs--sources).
+
+4. [Optional] Out-of-Order Time
+
+ To accommodate the disorder of input records, user can set a maximum out-of-order time for the event time watermark.
+The default value for this setting is also 0.
+
+ For more information, please refer to the [Fixed Amount of Lateness](#fixed-amount-of-lateness).
+
+5. [Optional] Generation Frequency
+
+ Flink offers three scenarios for generating event-time related watermarks:
+ - No EventTimeWatermarks are generated and emitted.
+ - EventTimeWatermarks are generated and emitted periodically.
+ - EventTimeWatermarks are generated and emitted for each event.
+
+ By default, Flink adopts the second approach, which involves periodically generating and emitting
+event-time related watermarks. The interval for this periodic generation is determined by the configuration
+"pipeline.auto-watermark-interval."
+
+After configuring and obtaining an instance of the `EventTimeWatermarkGeneratorBuilder`,
+users can utilize it to build a `ProcessFunction`. This `ProcessFunction` will extract the event time
+from each record and generates the corresponding event-time related watermark.
+
+A typical example of using `EventTimeWatermarkGeneratorBuilder` is shown in below:
+
+```java
+NonKeyedPartitionStream stream = ...;
+
+EventTimeWatermarkGeneratorBuilder<POJO> builder = EventTimeExtension
+ .newWatermarkGeneratorBuilder(pojo -> pojo.getTimeStamp()) // set event time extractor
+ .withIdleness(Duration.ofSeconds(10)) // set input idle timeout
+ .withMaxOutOfOrderTime(Duration.ofSeconds(30)) // set max out-of-order time
+ .periodicWatermark(Duration.ofMillis(200)); // set periodic watermark generation interval
+
+stream.process(builder.buildAsProcessFunction())
+ .process(...);
+```
+
+{{< hint warning >}}
+**Attention**: Both timestamps and event time watermarks
+are specified as milliseconds since the Java epoch of 1970-01-01T00:00:00Z.
+{{< /hint >}}
+
+#### Fixed Amount of Lateness
+
+There are some scenarios where the maximum lateness that can be
+encountered in a stream is known in advance, e.g. when creating a custom source
+containing elements with timestamps spread within a fixed period of time for
+testing.
+
+Users can deal these cases with `EventTimeWatermarkGeneratorBuilder#withMaxOutOfOrderTime`,
+i.e. the maximum amount of time an element is allowed to be late before being ignored when
+computing the final result for the given window. Lateness corresponds to the
+result of `t - t_w`, where `t` is the (event-time) timestamp of an element, and
+`t_w` that of the previous event time watermark. If `lateness > 0` then the element is
+considered late and is, by default, ignored when computing the result of the
+job for its corresponding window.
+
+#### Dealing With Idle Inputs / Sources
+
+If one of the input splits/partitions/shards does not carry events for a while
+this means that the `EventTimeWatermarkGeneratorBuilder` also does not get any new information
+on which to base a watermark. We call this an *idle input* or an *idle source*.
+This is a problem because it can happen that some of your partitions do still
+carry events. In that case, the event time watermark will be held back, because it is
+computed as the minimum over all the different parallel event time watermarks.
+
+To deal with this, you can configure an idleness timeout with
+`EventTimeWatermarkGeneratorBuilder#withIdleness` that will detect idleness
+and mark an input as idle.
+
+If an input remains idle, the `EventTimeWatermarkGeneratorBuilder` will emit an idle status watermark
+to indicate that the input is inactive. Consequently, downstream `ProcessFunction` instances will
+disregard this input when combining event time watermarks.
+
+### Generate Watermarks by Custom ProcessFunction
+
+Users can create and send event-time related watermarks by customizing the `ProcessFunction`.
+To do this, they need to follow these steps:
+1. Declare the `EventTimeWatermarkDeclaration` in the custom `ProcessFunction`, and the `IdleStatusWatermarkDeclaration` if support for input idle is required.
+2. Create the event-time related watermark using the `EventTimeWatermarkDeclaration`.
+3. (Optional) Create the idle status watermark using the `IdleStatusWatermarkDeclaration`.
+4. Send the watermarks through the `WatermarkManager`.
+
+The following shows an example of this approach.
+
+```java
+public static class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Integer, Integer> {
+
+
+ @Override
+ public Collection<? extends WatermarkDeclaration> watermarkDeclarations() {
+ return Set.of(
+ EventTimeExtension.EVENT_TIME_WATERMARK_DECLARATION,
+ EventTimeExtension.IDLE_STATUS_WATERMARK_DECLARATION
+ );
+ }
+
+ @Override
+ public void processRecord(Integer record, Collector<Integer> output, PartitionedContext ctx)
+ throws Exception {
+ // do something as needed
+
+ long eventTime = ... // get event time from record
+ // generate event time watermark and send to downstrea
+ LongWatermark eventTimeWatermark = EventTimeExtension.EVENT_TIME_WATERMARK_DECLARATION.newWatermark(eventTime);
+ ctx.getNonPartitionedContext()
+ .getWatermarkManager()
+ .emitWatermark(eventTimeWatermark);
+ }
+}
+```
+
+{{< hint warning >}}
+**Attention**: Both timestamps and event time watermarks
+are specified as milliseconds since the Java epoch of 1970-01-01T00:00:00Z.
+{{< /hint >}}
+
+## Handle Event-Time related Watermarks
+
+As described in [Generate Event-Time related Watermarks](#generate-event-time-related-watermarks),
+Flink provides abstractions that enable programmers to assign their own timestamps and emit event-time related watermarks.
+
+Once event-time related watermarks are generated and propagated, Flink can use them to determine the
+event time of the program, for example triggering a window when the event time watermark
+exceeds the end time of the Window.
+
+In addition, users can use event-time related watermarks to implement their own business logic.
+Currently, Flink provide users with two ways to use event-time related watermarks, one is to use the built-in
+`EventTimeProcessFunction`, and the other is to use the most basic `ProcessFunction` to
+process event-time related watermarks.
+
+It should be noted that the first method allows users to quickly register and unregister
+event timers, while the second method requires users to implement this function themselves.
+Therefore, when you need to use or perceive event time in your program, we recommend using
+the first method because it is simple and efficient.
+
+Each approach is described in detail below, users should select one of these to suit their needs.
+
+### Handle Event-Time related Watermarks by EventTimeProcessFunction
+
+`EventTimeProcessFunction` is provided by Flink and is a wrapper for `ProcessFunction` that needs to use event time.
+
+When utilizing the `EventTimeProcessFunction`, it's important for user's custom `ProcessFunction`
+to implement the `OneInputEventTimeStreamProcessFunction` / `TwoInputBroadcastEventTimeStreamProcessFunction`
+/ `TwoInputNonBroadcastEventTimeStreamProcessFunction` / `TwoOutputEventTimeStreamProcessFunction` interface
+instead of the `OneInputStreamProcessFunction` / `TwoInputBroadcastStreamProcessFunction`
+/ `TwoInputNonBroadcastStreamProcessFunction` / `TwoOutputStreamProcessFunction` interface.
+
+Below is the interface of the `EventTimeProcessFunction` and `OneInputEventTimeStreamProcessFunction`:
+
+```java
+@Experimental
+public interface EventTimeProcessFunction extends ProcessFunction {
+ /**
+ * Initialize the {@link EventTimeProcessFunction} with an instance of {@link EventTimeManager}.
+ * Note that this method should be invoked before the open method.
+ */
+ void initEventTimeProcessFunction(EventTimeManager eventTimeManager);
+}
+
+@Experimental
+public interface OneInputEventTimeStreamProcessFunction<IN, OUT>
+ extends EventTimeProcessFunction, OneInputStreamProcessFunction<IN, OUT> {
+
+ /**
+ * The {@code #onEventTimeWatermark} method signifies that the EventTimeProcessFunction has
+ * received an EventTimeWatermark. Other types of watermarks will be processed by the {@code
+ * ProcessFunction#onWatermark} method.
+ */
+ default void onEventTimeWatermark(
+ long watermarkTimestamp, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {}
+
+ /**
+ * Invoked when an event-time timer fires. Note that it is only used in {@link
+ * KeyedPartitionStream}.
+ */
+ default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+}
+```
+
+There are three methods that should be noted:
+- initEventTimeProcessFunction
+ - This method allows the `EventTimeProcessFunction` to obtain an instance of `EventTimeManager`.
+ Users can use this instance to access the current event time and to create or delete event timers as needed. Note that the event timer is only used in Keyed Partition Stream.
+
+- onEventTimeWatermark
+ - This method signifies that the `EventTimeProcessFunction` has received an event time watermark.
+ It is important to note in `EventTimeProcessFunction`, the event time watermarks will be processed
+ by `EventTimeProcessFunction#onEventTimeWatermark`, whereas other types of watermarks will be processed by the `EventTimeProcessFunction#onWatermark`.
+
+- onEventTimer
+ - This callback method is triggered by the event timer. Within this method, users can access
+ the key and event time associated with the event timer, perform necessary calculations, and output the results.
+
+Here is an example of how to implement `EventTimeProcessFunction`:
+
+```java
+class CustomEventTimeProcessFunction
+ implements OneInputEventTimeStreamProcessFunction<InputPojo, OutputPojo> {
+
+ private EventTimeManager eventTimeManager;
+
+ @Override
+ public void initEventTimeProcessFunction(EventTimeManager eventTimeManager) {
+ // get event time manager instance
+ this.eventTimeManager = eventTimeManager;
+ }
+
+ @Override
+ public void processRecord(
+ InputPojo record,
+ Collector<OutputPojo> output,
+ PartitionedContext<OutputPojo> ctx)
+ throws Exception {
+ ...
+
+ // register event timer
+ eventTimeManager.registerTimer(targetTimestamp);
+ }
+
+ @Override
+ public void onEventTimeWatermark(
+ long watermarkTimestamp,
+ Collector<OutputPojo> output,
+ NonPartitionedContext<OutputPojo> ctx)
+ throws Exception {
+ // sense event time watermark arrival
+ }
+
+ @Override
+ public void onEventTimer(
+ long timestamp,
+ Collector<OutputPojo> output,
+ PartitionedContext<OutputPojo> ctx) {
+ // write your event timer callback here
+ }
+}
+```
+
+After implementing the `EventTimeProcessFunction`, users should wrap their custom function using
+`EventTimeUtils#wrapProcessFunction`. This step is essential as it provides the necessary components,
+including an instance of `EventTimeManager`, and declares the built-in state required for timers
+and other functionalities.
+
+Here is an example of how to wrap the `EventTimeProcessFunction`:
+
+```java
+NonKeyedPartitionStream stream = ...;
+stream.keyBy(x -> x.getKey())
+ .process(EventTimeExtension.wrapProcessFunction(new CustomEventTimeProcessFunction()))
+ .process(...);
+```
+
+
+### Handle Event-Time related Watermarks by Custom ProcessFunction
+
+Similarly, users can handle event-time related watermarks by implementing `ProcessFunction` instead of `EventTimeProcessFunction`.
+
+In this approach, users must evaluate whether the current watermark received is an event time watermark
+or idle status watermark within the `ProcessFunction#onWatermark` method and execute the appropriate
+processing logic accordingly.
+
+An example is provided below.
+
+```java
+public static class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Integer, Integer> {
+
+ @Override
+ public WatermarkHandlingResult onWatermark(
+ Watermark watermark,
+ Collector<Integer> output,
+ NonPartitionedContext<Integer> ctx) throws Exception {
+ if (EventTimeExtension.isEventTimeWatermark(watermark)) {
+ // do something as needed
+ ...
+ return WatermarkHandlingResult.PEEK;
+ } else if (EventTimeExtension.isIdleStatusWatermark(watermark)) {
+ // do something as needed
+ ...
+ return WatermarkHandlingResult.PEEK;
+ } else {
+ // do something as needed
+ ...
+ }
+ }
+}
+```
+
+It is important to note that when `ProcessFunction#onWatermark` handles event-time related watermarks,
+it should return `WatermarkHandlingResult#PEEK`. This indicates that the Flink framework will choose
+the processing logic based on the watermark definition.
+For event time watermark and idle status watermark, the Flink framework will forward the watermark downstream.
+
+Conversely, if `ProcessFunction#onWatermark` returns `WatermarkHandlingResult#POP`, the watermarks
+will not be sent downstream by the Flink framework. Users should be aware that this may result in
+the loss of the watermark, or they may need to send the watermark manually.
+
+## Example
+Here is an example of how to use event timer service in Flink:
+
+```java
+NonKeyedPartitionStream<POJO> source = ...;
+
+source.process(
+ EventTimeExtension.<POJO>newWatermarkGeneratorBuilder(pojo -> pojo.getTimestamp())
+ .periodicWatermark(Duration.ofMillis(200))
+ .buildAsProcessFunction()
+ )
+ .keyBy(pojo -> pojo.getKey())
+ .process(EventTimeExtension.wrapProcessFunction(
+ new OneInputEventTimeStreamProcessFunction<POJO, String>() {
+
+ private EventTimeManager eventTimeManager;
+
+ @Override
+ public void initEventTimeProcessFunction(EventTimeManager eventTimeManager) {
+ // get event time manager instance
+ this.eventTimeManager = eventTimeManager;
+ }
+
+ @Override
+ public void processRecord(
+ POJO record,
+ Collector<String> output,
+ PartitionedContext<String> ctx) throws Exception {
+ ...
+
+ // register event timer
+ eventTimeManager.registerTimer(targetTimestamp);
+ }
+
+ @Override
+ public void onEventTimer(
+ long timestamp,
+ Collector<String> output,
+ PartitionedContext<String> ctx) {
+ // write your event timer callback here
+ }
+
+ }
+ )
+ )
+ .toSink(...);
+```
+{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream-v2/time-processing/processing_timer_service.md b/docs/content.zh/docs/dev/datastream-v2/time-processing/processing_timer_service.md
new file mode 100644
index 00000000000..d94d868d797
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/time-processing/processing_timer_service.md
@@ -0,0 +1,110 @@
+---
+title: "Processing Timer Service"
+weight: 8
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Processing Timer Service
+
+The process timer service is a fundamental primitive of the Flink DataStream API, provided by Flink.
+It allows users to register timers for performing calculations at specific processing time points.
+
+For a comprehensive explanation of processing time, please refer to the section on [Notions of Time: Event Time and Processing Time]({{< ref "docs/concepts/time" >}}#notions-of-time-event-time-and-processing-time).
+
+In this section, we will introduce how to utilize the processing timer service within the Flink DataStream API.
+
+## ProcessingTimerManager
+
+The `ProcessingTimerManager` is the key component for utilizing the processing timer service.
+You can obtain it by calling `PartitionedContext#getProcessingTimeManager` within a `ProcessFunction`.
+Below is the interface for the `ProcessingTimerManager`:
+
+```java
+@Experimental
+public interface ProcessingTimeManager {
+ /**
+ * Register a processing timer for this process function. `onProcessingTimer` method of this
+ * function will be invoked as callback if the timer expires.
+ *
+ * @param timestamp to trigger timer callback.
+ */
+ void registerTimer(long timestamp);
+
+ /**
+ * Deletes the processing-time timer with the given trigger timestamp. This method has only an
+ * effect if such a timer was previously registered and did not already expire.
+ *
+ * @param timestamp indicates the timestamp of the timer to delete.
+ */
+ void deleteTimer(long timestamp);
+
+ /**
+ * Get the current processing time.
+ *
+ * @return current processing time.
+ */
+ long currentTime();
+}
+```
+
+As shown above, `ProcessingTimerManager` has three methods: `registerTimer`, `deleteTimer`, and `currentTime`.
+When the target time of `registerTimer` arrives, Flink will call `ProcessFunction#onProcessingTimer`,
+and users should write their own calculation logic in `ProcessFunction#onProcessingTimer`.
+
+It should be noted that:
+1. for timers with the same target time, Flink will only retain one timer.
+This means that if multiple timers with the same target time are registered, `ProcessFunction#onProcessingTimer` will only be invoked once.
+2. the `ProcessingTimerManager` can only be used in Keyed Partitioned Stream.
+
+## Example
+Here is an example of using the processing timer service:
+
+```java
+public class CustomProcessFunction implements OneInputStreamProcessFunction<String, String> {
+
+ @Override
+ public void processRecord(
+ String record,
+ Collector<String> output,
+ PartitionedContext<String> ctx) throws Exception {
+ // do some calculation as needed
+
+ // register a timer with a target time of one minute after the current time
+ long currentTime = ctx.getProcessingTimeManager().currentTime();
+ ctx.getProcessingTimeManager().registerTimer(currentTime + Duration.ofMinutes(1L).toMillis());
+ }
+
+ @Override
+ public void onProcessingTimer(
+ long timestamp,
+ Collector<String> output,
+ PartitionedContext<String> ctx) {
+ // do some calculation and output result as needed
+ }
+}
+```
+
+{{< top >}}
diff --git a/docs/content.zh/docs/dev/datastream-v2/watermark.md b/docs/content.zh/docs/dev/datastream-v2/watermark.md
new file mode 100644
index 00000000000..da54198354b
--- /dev/null
+++ b/docs/content.zh/docs/dev/datastream-v2/watermark.md
@@ -0,0 +1,203 @@
+---
+title: Watermark
+weight: 11
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Watermark
+
+Before introducing Watermark, users should be aware that Watermark in DataStream V2 does not
+refer to the [original Watermark]({{< ref "docs/concepts/time" >}}#event-time-and-watermarks) that
+measure progress in event time, but is a special event that can be customized by the user and
+can be propagated along the streams.
+
+Using Watermark in Flink involves three key steps:
+1. Define and Declare the Watermark
+2. Emit the Watermark
+3. Handle the Watermark
+
+Let's follow these steps to understand how to use Watermark in Flink.
+
+## Define and Declare Watermark
+
+A watermark is a special event that carries data. It can be emitted from a `Source`
+or `ProcessFunction`, propagating along streams, received by downstream `ProcessFunction`.
+When defining a Watermark, there are four aspects need to consider:
+
+1. [Required] Watermark Identifier
+
+ Since multiple types of Watermarks can propagate in streams, it's essential to assign an
+identifier to each Watermark for differentiation.
+
+ Note that the identifier must be a String, is case-sensitive, and must be globally unique within the entire job.
+
+2. [Required] Watermark Data Type
+
+ It is important to specify the data type of the Watermark. Currently, Flink supports two types: Long and Bool.
+
+3. [Required] Combine Function and `combineWaitForAllChannels`
+
+ A `ProcessFunction` may receive multiple Watermarks from different input channels due to
+having multiple upstream inputs, each potentially with varying degrees of parallelism.
+In such cases, users often wish to combine Watermarks from input channels before outputting
+to the `ProcessFunction`.
+
+ Flink supports the following combination functions:
+ - For `Long` type Watermark:
+ - `MIN`: Retains and outputs the minimum value of all received watermarks.
+ - `MAX`: Retains and outputs the maximum value of all received watermarks.
+ - For `Bool` type Watermark:
+ - `AND`: Retains and outputs the logical AND result of all received watermarks.
+ - `OR`: Retains and outputs the logical OR result of all received watermarks.
+
+ Additionally, users can configure whether the combining process should wait until the
+`ProcessFunction` has received Watermarks from all upstream channels. This is particularly useful
+in some scenarios. For example, the event time watermark need to wait receives the watermarks from
+all inputs and then combine them. This ensures that the time carried by the event time watermark does not decrease.
+By default, the `combineWaitForAllChannels` setting is false.
+
+4. [Optional] `WatermarkHandlingStrategy` by Framework
+
+ The `WatermarkHandlingStrategy` determines whether the framework should send the watermark
+to downstream `ProcessFunction` when the user-defined `ProcessFunction` does pop the Watermark.
+There are two options of this setting:
+ - IGNORE: The framework shouldn't take any action.
+ - FORWARD: The framework should send the watermark to downstream.
+
+ This optional setting is useful in some cases. For example, setting it to `IGNORE` can indicate
+that the framework does not need to propagate this Watermark, but rather that it is up to the user
+to control its sending.
+
+To simplify the process of defining a Watermark, Flink offer a `WatermarkBuilder`.
+This builder ultimately creates a `WatermarkDeclaration` object.
+Below is an example demonstrating how to use it to defining a Watermark:
+
+```java
+LongWatermarkDeclaration watermarkDeclaration = WatermarkDeclarations
+ .newBuilder("MY_CUSTOM_WATERMARK_IDENTIFIER")
+ .typeLong()
+ .combineFunctionMax()
+ .combineWaitForAllChannels(true)
+ .defaultHandlingStrategyForward()
+ .build();
+```
+
+Once users have defined the Watermark, it is essential to declare it in the
+`ProcessFunction#declareWatermarks` or `Source#declareWatermarks`. This step allows the framework to recognize it properly.
+Here’s an example of how to declare the Watermark in the `ProcessFunction`:
+
+```java
+public class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Long, Long> {
+
+ LongWatermarkDeclaration watermarkDeclaration = ...;
+
+ @Override
+ public Set<? extends WatermarkDeclaration> declareWatermarks() {
+ return Set.of(watermarkDeclaration);
+ }
+}
+```
+
+Please note that each type of Watermark only needs to be declared once in a job.
+
+## Emit Watermark
+
+Users can utilize the Watermark declaration to create a watermark. Below is an example of how to generate a Long type watermark with a value of 1:
+
+```java
+LongWatermarkDeclaration watermarkDeclaration = ...;
+LongWatermark watermark = watermarkDeclaration.newWatermark(1);
+```
+
+And users can emit Watermark in `ProcessFunction` by calling `nonPartitionedContext.getWatermarkManager().emitWatermark(watermark)`,
+it is also support emit Watermark in `Source` by calling `sourceReaderContext.emitWatermark(watermark)`.
+Here is an example of how to emit a Watermark in a `ProcessFunction`:
+```java
+public class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Long, Long> {
+
+ LongWatermarkDeclaration watermarkDeclaration = ...;
+
+ @Override
+ public Set<? extends WatermarkDeclaration> declareWatermarks() {
+ return Set.of(watermarkDeclaration);
+ }
+
+ @Override
+ public void processRecord(Long record, Collector<Long> output, PartitionedContext<Long> ctx)
+ throws Exception {
+ // do something as needed
+
+ // generate and emit Watermark
+ LongWatermark watermark = watermarkDeclaration.newWatermark(1);
+ ctx.getNonPartitionedContext().getWatermarkManager().emitWatermark(watermark);
+ }
+}
+```
+
+## Handle Watermark
+
+Once the `ProcessFunction` receives a Watermark, the framework will invoke the
+`ProcessFunction#onWatermark` method to process it. Therefore, users will need to override
+`ProcessFunction#onWatermark` in order to handle the Watermark appropriately.
+
+The return value of the `ProcessFunction#onWatermark` method is of type `WatermarkHandlingResult`,
+which has two possible options:
+
+- PEEK: `ProcessFunction` only peek the Watermark, and it's framework's responsibility to handle this watermark.
+
+ The framework will forward/ignore the Watermark according to the `WatermarkHandlingStrategy` associated with it.
+
+- POLL: This Watermark should be sent to downstream by process function itself. The framework does no additional processing.
+
+By default, the `ProcessFunction#onWatermark` method returns `WatermarkHandlingResult.PEEK`.
+
+Here is an example of how to handle Watermarks in `ProcessFunction`:
+
+```java
+
+public static final String CUSTOM_WATERMARK_IDENTIFIER = "CUSTOM_WATERMARK_IDENTIFIER";
+
+public class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Long, Long> {
+
+ ...
+
+ @Override
+ public WatermarkHandlingResult onWatermark(
+ Watermark watermark, Collector<Long> output, NonPartitionedContext<Long> ctx) {
+ // For Watermark that this ProcessFunction is interested in, process the watermark
+ if (watermark.getIdentifier().equals(CUSTOM_WATERMARK_IDENTIFIER)) {
+ // do something as needed
+ }
+
+ // For other Watermarks, return PEEK
+ return WatermarkHandlingResult.PEEK;
+ }
+}
+```
+
+{{< top >}}
diff --git a/docs/content/_index.md b/docs/content/_index.md
index e084d72e0df..2e5ed5eccef 100644
--- a/docs/content/_index.md
+++ b/docs/content/_index.md
@@ -62,6 +62,7 @@ The reference documentation covers all the details. Some starting points:
{{< columns >}}
* [DataStream API]({{< ref "docs/dev/datastream/overview" >}})
+* [DataStream API (V2)]({{< ref "docs/dev/datastream-v2/overview" >}})
* [Table API & SQL]({{< ref "docs/dev/table/overview" >}})
* [Stateful Functions](https://nightlies.apache.org/flink/flink-statefun-docs-stable/)
diff --git a/docs/content/docs/dev/datastream-v2/_index.md b/docs/content/docs/dev/datastream-v2/_index.md
new file mode 100644
index 00000000000..ed1bb51bebc
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/_index.md
@@ -0,0 +1,23 @@
+---
+title: DataStream API (V2)
+bookCollapseSection: true
+weight: 3
+---
+<!--
+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.
+-->
diff --git a/docs/content/docs/dev/datastream-v2/building_blocks.md b/docs/content/docs/dev/datastream-v2/building_blocks.md
new file mode 100644
index 00000000000..d9e2e65c267
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/building_blocks.md
@@ -0,0 +1,183 @@
+---
+title: Building Blocks
+weight: 3
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Building Blocks
+
+DataStream, Partitioning, ProcessFunction are the most fundamental elements of DataStream API and respectively represent:
+
+- What are the types of data streams
+
+- How data is partitioned
+
+- How to perform operations / processing on data streams
+
+They are also the core parts of the fundamental primitives provided by DataStream API.
+
+## DataStream
+
+Data flows on the stream may be divided into multiple partitions. According to how the data is partitioned on the stream, we divide it into the following categories:
+
+- Global Stream: Force single partition/parallelism, and the correctness of data depends on this.
+
+- Partition Stream:
+ - Divide data into multiple partitions. State is only available within the partition. One partition can only be processed by one task, but one task can handle one or multiple partitions.
+According to the partitioning approach, it can be further divided into the following two categories:
+
+ - Keyed Partition Stream: Each key is a partition, and the partition to which the data belongs is deterministic.
+
+ - Non-Keyed Partition Stream: Each parallelism is a partition, and the partition to which the data belongs is nondeterministic.
+
+- Broadcast Stream: Each partition contains the same data.
+
+## Partitioning
+
+Above we defined the stream and how it is partitioned. The next topic to discuss is how to convert
+between different partition types. We call these transformations partitioning.
+For example non-keyed partition stream can be transformed to keyed partition stream via a `KeyBy` partitioning.
+
+```java
+NonKeyedPartitionStream<Tuple<Integer, String>> stream = ...;
+KeyedPartitionStream<Integer, String> keyedStream = stream.keyBy(record -> record.f0);
+```
+
+Overall, we have the following four partitioning:
+
+- KeyBy: Let all data be repartitioned according to specified key.
+
+- Shuffle: Repartition and shuffle all data.
+
+- Global: Merge all partitions into one.
+
+- Broadcast: Force partitions broadcast data to downstream.
+
+The specific transformation relationship is shown in the following table:
+
+{{< img src="/fig/datastream/one-input-partitioning.png" alt="one-input-partitioning" width="70%" >}}
+
+(A crossed box indicates that it is not supported or not required)
+
+One thing to note is: broadcast can only be used in conjunction with other inputs and cannot be directly converted to other streams.
+
+## ProcessFunction
+Once we have the data stream, we can apply operations on it. The operations that can be performed over
+DataStream are collectively called Process Function. It is the only entrypoint for defining all kinds
+of processing on the data streams.
+
+### Classification of ProcessFunction
+According to the number of input / output, they are classified as follows:
+
+| Partitioning | Number of Inputs | Number of Outputs |
+|:-----------------------------------------:|:----------------------:|:-------------------------------:|
+| OneInputStreamProcessFunction | 1 | 1 |
+| TwoInputNonBroadcastStreamProcessFunction | 2 | 1 |
+| TwoInputBroadcastStreamProcessFunction | 2 | 1 |
+| TwoOutputStreamProcessFunction | 1 | 2 |
+
+(Processing for more inputs and outputs can be achieved by combining multiple process functions)
+
+We have two types of two-input process function, depending on whether one of the input is broadcast stream.
+
+DataStream has series of `process` and `connectAndProcess` methods to transform the input stream or connect and transform two input streams via ProcessFunction.
+
+### Requirements for input and output streams
+
+The following two tables list the input and output stream combinations supported by OneInputStreamProcessFunction and TwoOutputStreamProcessFunction respectively.
+
+For OneInputStreamProcessFunction:
+
+| Input Stream | Output Stream |
+|:------------:|:----------------:|
+| Global | Global |
+| Keyed | Keyed / NonKeyed |
+| NonKeyed | NonKeyed |
+| Broadcast | Not Supported |
+
+For TwoOutputStreamProcessFunction:
+
+| Input Stream | Output Stream |
+|:------------:|:-------------------------------------:|
+| Global | Global + Global |
+| Keyed | Keyed + Keyed / Non-Keyed + Non-Keyed |
+| NonKeyed | NonKeyed + NonKeyed |
+| Broadcast | Not Supported |
+
+Things with two inputs is a little more complicated. The following table lists which streams are compatible with each other and the types of streams they output.
+
+A cross(❎) indicates not supported.
+
+| Output | Global | Keyed | NonKeyed | Broadcast |
+|:---------:|:------:|:------------------:|:--------:|:-----------------:|
+| Global | Global | ❎ | ❎ | ❎ |
+| Keyed | ❎ | NonKeyed / Keyed | ❎ | NonKeyed / Keyed |
+| NonKeyed | ❎ | ❎ | NonKeyed | NonKeyed |
+| Broadcast | ❎ | NonKeyed / Keyed | NonKeyed | ❎ |
+
+## Config Process
+
+After defining the process functions, you may want to make some configurations for the properties of this processing.
+For example, set the parallelism and name of the process operation, etc.
+
+The return value of `process`/`connectAndProcess` is both a stream and a handle that allowing us to configure the previous processing.
+It has a number of methods called `withXXX` to do the configuration. For example:
+
+```java
+inputStream
+ .process(func1) // do process 1
+ .withName("my-process-func") // configure name for process 1
+ .withParallelism(2) // configure parallelism for process 1
+ .process(func2) // do further process 2
+```
+
+## Example
+
+Here is an example to show how to use those building blocks to write a flink job.
+
+```java
+// create environment
+ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+// create a stream from source
+env.fromSource(someSource)
+ // map every element x to x + 1.
+ .process(new OneInputStreamProcessFunction<Integer, Integer>() {
+ @Override
+ public void processRecord(
+ Integer x,
+ Collector<Integer> output)
+ throws Exception {
+ output.collect(x + 1);
+ }
+ })
+ // If the sink does not support concurrent writes, we can force the stream to one partition
+ .global()
+ // sink the stream to some sink
+ .toSink(someSink);
+// execute the job
+env.execute()
+```
+
+{{< top >}}
diff --git a/docs/content/docs/dev/datastream-v2/builtin-funcs/_index.md b/docs/content/docs/dev/datastream-v2/builtin-funcs/_index.md
new file mode 100644
index 00000000000..6493a6c5e5d
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/builtin-funcs/_index.md
@@ -0,0 +1,23 @@
+---
+title: Builtin Functions
+bookCollapseSection: true
+weight: 7
+---
+<!--
+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.
+-->
diff --git a/docs/content/docs/dev/datastream-v2/builtin-funcs/joining.md b/docs/content/docs/dev/datastream-v2/builtin-funcs/joining.md
new file mode 100644
index 00000000000..1640d158b9e
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/builtin-funcs/joining.md
@@ -0,0 +1,126 @@
+---
+title: "Joining"
+weight: 9
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Joining
+
+Join is used to merge two data streams by matching elements from both streams based on a common key,
+and performing calculations on the matched elements.
+
+This section will introduce the Join operation in DataStream in detail.
+
+Please note that currently, DataStream supports only non-window INNER joins.
+Other types of joins, such as interval joins, lookup joins, and window joins, will be supported in the future.
+
+## Non-Window Join
+
+To use the Join, users must specify the following:
+
+1. The left and right streams that will be joined.
+2. The `KeySelector` of two streams, which determines the join key.
+3. The [`JoinFunction`](#joinfunction): the processing logic applied to the matched data.
+
+We will first introduce the APIs for performing Join, and then introduce the JoinFunction.
+
+### APIs for performing Join
+
+There are two approaches to performing a Join operation:
+
+1. Use `BuiltinFuncs.join` to connect two streams and execute the Join.
+2. Convert the `JoinFunction` into a `ProcessFunction` and utilize `KeyedPartitionStream#connectAndProcess` to execute the Join.
+
+In the first approach, users can choose between two options based on the types of the input data streams.
+
+- If both input data streams are already Keyed Partition Stream, users can directly combine the two Keyed Partition
+Stream along with the `JoinFunction` to produce the joined data stream.
+
+```java
+NonKeyedPartitionStream joinedStream = BuiltinFuncs.join(
+ keyedStream1,
+ keyedStream2,
+ new CustomJoinFunction()
+);
+```
+
+- If both input data streams are Non-Keyed Partition Stream, users need to incorporate the corresponding
+Join key `KeySelector` and `JoinFunction` to convert the two Non-Keyed Partition Stream into the joined data stream.
+
+```java
+NonKeyedPartitionStream joinedStream = BuiltinFuncs.join(
+ stream1,
+ new CustomJoinKeySelector1(),
+ stream2,
+ new CustomJoinKeySelector2(),
+ new CustomJoinFunction()
+);
+```
+
+In the second approach, users can convert the `JoinFunction` into a `ProcessFunction`, which can then be processed by
+the DataStream API. An example of this conversion and how to use the converted `ProcessFunction` is provided below:
+
+```java
+TwoInputNonBroadcastStreamProcessFunction wrappedJoinFunction = BuiltinFuncs.join(new CustomJoinFunction());
+NonKeyedPartitionStream joinedStream = keyedStream1.connectAndProcess(
+ keyedstream2,
+ wrappedJoinFunction
+);
+```
+
+Please note that in the second approach, it is essential that the two input data streams are Keyed Partition Stream.
+
+
+### JoinFunction
+
+`JoinFunction` is an interface used to describe how to calculate the matched data.
+It has only one method `processRecord`. Users can get the matched elements in `processRecord`
+to perform calculations and then output the calculation results.
+
+Below is an example demonstrating how to use a `JoinFunction` to connect student personal
+information with their exam scores.
+
+```java
+class JoinStudentInformationAndScore
+ implements JoinFunction<StudentInfo, ExamScore, EnrichedStudentExamScore> {
+
+ @Override
+ public void processRecord(
+ StudentInfo studentInfo,
+ ExamScore examScore,
+ Collector<EnrichedStudentExamScore> output,
+ RuntimeContext ctx)
+ throws Exception {
+ // do some calculation logic and emit joined result
+ EnrichedStudentExamScore studentExamScore =
+ new EnrichedStudentExamScore(studentInfo.getId(), studentInfo.getName(), examScore.getScore());
+ output.collect(studentExamScore);
+ }
+}
+```
+
+
+{{< top >}}
diff --git a/docs/content/docs/dev/datastream-v2/builtin-funcs/windows.md b/docs/content/docs/dev/datastream-v2/builtin-funcs/windows.md
new file mode 100644
index 00000000000..008216c4919
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/builtin-funcs/windows.md
@@ -0,0 +1,421 @@
+---
+title: "Windows"
+weight: 8
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Windows
+
+Windows are at the heart of processing infinite streams. Windows split the stream into "buckets"
+of finite size, over which we can apply computations. This document focuses on how windowing is
+performed in Flink DataStream and how the programmer can benefit to the maximum from its offered
+functionality.
+
+To utilize the Window functionality provided by DataStream API, users need to complete the following three steps:
+1. Declare Window: Specify the type of window you wish to use.
+2. Define `WindowProcessFunction`: Outline the logic that should be executed at various stages of the Window's lifecycle.
+3. Combine the Window Declaration and `WindowProcessFunction` to build a `ProcessFunction`
+ - Encapsulate the window declaration and the `WindowProcessFunction` into a single
+ `ProcessFunction`, which can then be leveraged within the DataStream API.
+
+This section will provide a comprehensive overview of these three steps and include an example
+illustrating how to use Window in DataStream API.
+
+## Declare Window
+
+Users should first determine which type of window to use for their applications. We currently offer
+three built-in window types: Time Window, Session Window and Global Window.
+
+### Time Window
+
+Time Windows are divided into multiple windows based on time ranges, and data is allocated to the
+corresponding window according to its timestamp. We support two types of Time Windows:
+tumbling windows and sliding windows. The time semantics within these windows can be classified
+as event time or processing time.
+
+Please note that Time Windows are currently supported only in Keyed Partition Stream.
+
+#### Tumbling Window
+
+*Tumbling Window* assigns each element to a window of a specified *window size*.
+Tumbling windows have a fixed size and do not overlap. For example, if you specify a tumbling
+window with a size of 5 minutes, the current window will be evaluated and a new window will be
+started every five minutes as illustrated by the following figure.
+
+{{< img src="/fig/tumbling-windows.svg" alt="Tumbling Windows" >}}
+
+The following code snippets show how to use tumbling windows.
+
+```java
+// create a tumbling window strategy with a window size of 60 seconds
+WindowStrategy windowStrategy = WindowStrategy.tumbling(Duration.ofSeconds(60), WindowStrategy.EVENT_TIME);
+```
+
+Time intervals can be specified by using one of `Duration.ofMillis(x)`, `Duration.ofSeconds(x)`,
+`Duration.ofMinutes(x)`, and so on.
+
+#### Sliding Window
+
+*Sliding Window* assigns elements to windows of fixed length. Similar to a tumbling
+window, the size of the windows is configured by the *window size* parameter.
+An additional *window slide* parameter controls how frequently a sliding window is started. Hence,
+sliding windows can be overlapping if the slide is smaller than the window size.
+In this case elements are assigned to multiple windows.
+
+For example, you could have windows of size 10 minutes that slides by 5 minutes. With this you
+get every 5 minutes a window that contains the events that arrived during the last 10 minutes
+as depicted by the following figure.
+
+{{< img src="/fig/sliding-windows.svg" alt="sliding windows" >}}
+
+The following code snippets show how to use sliding windows.
+
+```java
+// create a sliding window strategy with a window size of 60 seconds and a slide of 30 seconds
+WindowStrategy windowStrategy = WindowStrategy.sliding(Duration.ofSeconds(60), Duration.ofSeconds(30), WindowStrategy.PROCESSING_TIME);
+```
+
+Time intervals can be specified by using one of `Duration.ofMillis(x)`, `Duration.ofSeconds(x)`,
+`Duration.ofMinutes(x)`, and so on.
+
+#### Allowed Lateness
+
+When working with *event-time* windowing, it can happen that elements arrive late, *i.e.* the
+event time watermark that Flink uses to keep track of the progress of event-time is already past
+the end timestamp of a window to which an element belongs. See
+[event time]({{< ref "docs/dev/datastream-v2/time-processing/event_timer_service" >}}) for
+a more thorough discussion of how Flink deals with event time.
+
+By default, late elements are dropped when the event time watermark is past the end of the window.
+However, Flink allows to specify a maximum *allowed lateness* for window operators. Allowed lateness
+specifies by how much time elements can be late before they are dropped, and its default value is 0.
+
+Elements that arrive after the event time watermark has passed the end of the window, but before
+it passes the end of the window plus the allowed lateness, are still added to the window.
+The late but not dropped element will cause the window to fire again, and the late and dropped
+element will be processed by the [WindowProcessFunction#onLateRecord](#introduction-to-windowprocessfunction).
+
+In order to make this work, Flink keeps the state of windows until their allowed lateness expires.
+Once this happens, Flink removes the window and deletes its state, as also described in
+the [Window Lifecycle](#window-lifecycle) section.
+
+By default, the allowed lateness is set to `0`. That is, elements that arrive behind the watermark
+will be dropped.
+
+You can specify an allowed lateness like this:
+
+```java
+// create a sliding window strategy with a window size of 60 seconds and a slide of 30 seconds and an allowed lateness of 10 seconds
+WindowStrategy windowStrategy = WindowStrategy.sliding(Duration.ofSeconds(60), Duration.ofSeconds(30), WindowStrategy.PROCESSING_TIME, Duration.ofSeconds(10));
+```
+
+### Session Window
+
+The *session windows* groups elements by sessions of activity. Session windows do not overlap and
+do not have a fixed start and end time, in contrast to *tumbling windows* and *sliding windows*.
+Instead, a session window closes when it does not receive elements for a certain period of
+time, *i.e.*, when a gap of inactivity occurred. A session window can be configured
+with *session gap*. When this period expires, the current session closes and subsequent elements
+are assigned to a new session window.
+
+{{< img src="/fig/session-windows.svg" alt="session windows" >}}
+
+The following code snippets show how to declare session windows.
+
+```java
+// create a session window strategy with a session gap of 60 seconds
+WindowStrategy windowStrategy = WindowStrategy.session(Duration.ofSeconds(60), WindowStrategy.EVENT_TIME);
+```
+
+Please note that Session Windows are supported only in Global Stream and Keyed Partition Stream.
+
+### Global Window
+
+*Global Window* refers to a scenario where all elements are assigned to a single, unified Window.
+This windowing scheme is particularly useful in bounded stream scenarios, as it can be triggered
+once all inputs have concluded. Note that it is compatible with Global Stream, Keyed Partition Stream,
+and Non-Keyed Partition Stream.
+
+{{< img src="/fig/non-windowed.svg" alt="global windows" >}}
+
+The following code snippets show how to declare a global window.
+
+```java
+// create a global window strategy
+WindowStrategy windowStrategy = WindowStrategy.global();
+```
+
+## Define WindowProcessFunction
+### Window Lifecycle
+In a nutshell, a window is **created** as soon as the first element that should belong to this
+window arrives, and the window is **completely removed** when the time (event or processing time)
+passes its end timestamp plus the user-specified `allowed lateness` (see [Allowed Lateness](#allowed-lateness)).
+For example, with an event-time-based windowing that creates non-overlapping (or tumbling) windows
+every 5 minutes and has an allowed lateness of 1 min, Flink will create a new window for the
+interval between `12:00` and `12:05` when the first element with a timestamp that falls into this
+interval arrives, and it will remove it when the event time watermark passes the `12:06` timestamp.
+
+Flink abstracts the essential actions within the window lifecycle into methods in
+the [WindowProcessFunction](#introduction-to-windowprocessfunction). Users are required to implement
+their own *WindowProcessFunction* to define their specific windowing computation logic.
+
+### Introduction to WindowProcessFunction
+
+The `WindowProcessFunction` is the main component that users need to implement to process the
+data in the window. After declaring a window, users need to define the operational logic for various
+stages of the window's lifecycle, called *WindowProcessFunction*.
+
+Here is the interface of `OneInputWindowStreamProcessFunction`:
+
+```java
+/**
+ * A type of {@link WindowProcessFunction} for one-input window processing.
+ *
+ * @param <IN> The type of the input value.
+ * @param <OUT> The type of the output value.
+ */
+@Experimental
+public interface OneInputWindowStreamProcessFunction<IN, OUT> extends WindowProcessFunction {
+
+ /**
+ * This method will be invoked when a record is received. Its default behaviors to store data in
+ * built-in window state by {@link OneInputWindowContext#putRecord}. If the user overrides this
+ * method, they have to take care of the input data themselves.
+ */
+ default void onRecord(
+ IN record,
+ Collector<OUT> output,
+ PartitionedContext<OUT> ctx,
+ OneInputWindowContext<IN> windowContext)
+ throws Exception {
+ windowContext.putRecord(record);
+ }
+
+ /**
+ * This method will be invoked when the Window is triggered, you can obtain all the input
+ * records in the Window by {@link OneInputWindowContext#getAllRecords()}.
+ */
+ void onTrigger(
+ Collector<OUT> output,
+ PartitionedContext<OUT> ctx,
+ OneInputWindowContext<IN> windowContext)
+ throws Exception;
+
+ /**
+ * Callback when a window is about to be cleaned up. It is the time to deletes any state in the
+ * {@code windowContext} when the Window expires (the event time or processing time passes its
+ * {@code maxTimestamp} + {@code allowedLateness}).
+ */
+ default void onClear(
+ Collector<OUT> output,
+ PartitionedContext<OUT> ctx,
+ OneInputWindowContext<IN> windowContext)
+ throws Exception {}
+
+ /** This method will be invoked when a record is received after the window has been cleaned. */
+ default void onLateRecord(IN record, Collector<OUT> output, PartitionedContext<OUT> ctx)
+ throws Exception {}
+}
+```
+
+There are four key methods in the WindowProcessFunction. The names and meanings of these methods are as follows:
+1. `onRecord`: `onRecord` indicates that the window has received a record.
+2. `onTrigger`: `onTrigger` indicates that the window has been triggered.
+3. `onClear`: `onClear` indicates that the window has been cleared.
+4. `onLateRecord`: `onLateRecord` indicates that the window has received record after the window is cleared.
+
+There are some important points to consider regarding these methods.
+1. Windows can be triggered multiple times. Therefore, `onRecord` may be called after `onTrigger`.
+2. GlobalWindow is cleared when the data stream ends, while time/session windows are cleared after the window boundary is reached and the allowedLateness (see [Allowed Lateness](#allowed-lateness)) has elapsed.
+3. `onLateRecord` method is not possible to access the window state since the window has been cleared.
+
+Users should implement `WindowProcessFunction` to complete their own calculation logic.
+In this step, they can use [Window State](#declare-and-use-window-state) to store Window-related data and [access the built-in state](#access-built-state-of-window) used to store Window data.
+
+### Declare and Use Window State
+In a window, there are two types of states: partitioned state and window state.
+
+1. Partitioned State: We refer to the partition-related state as partitioned state.
+
+ For NonKeyedStream, this state is shared among a specific task. For KeyedStream, this state is shared among data with the same key.
+
+ User can declare partitioned state through `ProcessFunction#usesStates` and use partitioned state through `PartitionedContext#getStateManager`.
+It's users' responsibility to clear data in partitioned state that are no longer needed in `WindowProcessFunction#onClear`.
+2. Window State: We refer to the window-related state as window state.
+
+ Window state is bound to a specific window. For example, the window state declared and used for the same key in the 10:00-11:00 window is different from that in the 11:00-12:00 window.
+
+ User can declare window state through `WindowProcessFunction#usesWindowStates` and use window state through `WindowContext#getWindowState`.
+All window state will eventually be cleared by framework, whether or not the user clears it manually in `WindowProcessFunction#onClear`.
+
+### Access Built State of Window
+We provide built-in window state for each window to store the input data. Users can access this through `WindowContext#putRecord` and `WindowContext#getAllRecords`.
+This state will be cleared when the window is cleared.
+
+By default, `WindowProcessFunction#onRecord` stores the received data in the window's built-in state by `WindowContext#putRecord`, and users can retrieve all the data within the Window using `WindowContext#getAllRecords` when the window is triggered.
+
+Therefore, when overriding `WindowProcessFunction#onRecord`, users should consider whether they need to write the input data into the built-in state.
+A typical example is if users want to do pre-aggregation, they can declare a Window state, perform aggregation in `WindowProcessFunction#onRecord`, update the aggregated window state, and output the final result in `WindowProcessFunction#onTrigger`.
+Therefore, unnecessary cost of caching all data are eliminated.
+
+## Build a ProcessFunction
+After declaring the Window and defining the *WindowProcessFunction*, users should utilize the *BuiltinFuncs.window* method to encapsulate these two components into a *ProcessFunction*, which can then be integrated into the data processing stream.
+An example of this is as follows:
+
+```java
+KeyedPartitionStream stream = ...;
+OneInputStreamProcessFunction wrappedWindowProcessFunction = BuiltinFuncs.window(windowStrategy, new CustomWindowProcessFunction());
+stream.process(wrappedWindowProcessFunction)
+ .process(...);
+```
+
+By doing so, Flink will automatically manage the state and timers required for caching window data, relieving users of the burden of handling these details on their own.
+
+## Example: Count the sales of each product every hour
+
+The following is an example of how to use Window to count the sales of each product in each hour.
+
+In the following example, we first use the [event time extension]({{< ref "docs/dev/datastream-v2/time-processing/event_timer_service" >}}) to extract the event time from `orderSoucrce`, then declare a one-hour tumbling window and calculate the sales quantity of each product when the window is triggered.
+
+```java
+public class CountProductSalesEveryHour {
+
+ public static class Order {
+ public long orderId;
+ public long productId;
+ public long salesQuantity;
+ public long orderTime;
+ }
+
+ public static void main(String[] args) throws Exception {
+ ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+ // create order source stream
+ NonKeyedPartitionStream<Order> orderSource = ...;
+
+ // extract and propagate event time from order
+ NonKeyedPartitionStream<Order> orderStream = orderSource.process(
+ EventTimeExtension
+ .<Order>newWatermarkGeneratorBuilder(order -> order.orderTime)
+ .periodicWatermark(Duration.ofMillis(200))
+ .buildAsProcessFunction()
+ );
+
+ NonKeyedPartitionStream<Tuple2<Long, Long>> productSalesQuantityStream = orderStream
+ // key by productId
+ .keyBy(order -> order.productId)
+ .process(BuiltinFuncs.window(
+ // declare tumbling window with window size 10 seconds
+ WindowStrategy.tumbling(
+ Duration.ofHours(1),
+ WindowStrategy.EVENT_TIME),
+ // define window process function to calculate total sales quantity per product per window
+ new CountSalesQuantity()
+ )
+ );
+
+ // print result
+ productSalesQuantityStream.toSink(new WrappedSink<>(new PrintSink<>()));
+
+ env.execute("CountSalesQuantifyOfEachProductEveryHour");
+ }
+
+ public static class CountSalesQuantity implements OneInputWindowStreamProcessFunction<Order, Tuple2<Long, Long>> {
+
+ @Override
+ public void onTrigger(
+ Collector<Tuple2<Long, Long>> output,
+ PartitionedContext<Tuple2<Long, Long>> ctx,
+ OneInputWindowContext<Order> windowContext) throws Exception {
+ // get current productId
+ long productId = ctx.getStateManager().getCurrentKey();
+ // calculate total sales quantity
+ long totalSalesQuantity = 0;
+ for (Order order : windowContext.getAllRecords()) {
+ totalSalesQuantity += order.salesQuantity;
+ }
+ // emit result
+ output.collect(Tuple2.of(productId, totalSalesQuantity));
+ }
+ }
+}
+```
+
+In the scenario of this example, users can reduce the cost of Window calculation by performing pre-aggregation in the Window.
+We rewrote the `CountSalesQuantity` WindowProcessFunction into the `CountSalesQuantityWithPreAggregation` WindowProcessFunction that can perform pre-aggregation when the input data arrives.
+
+```java
+
+public static class CountSalesQuantityWithPreAggregation implements OneInputWindowStreamProcessFunction<Order, Tuple2<Long, Long>> {
+
+ private final ValueStateDeclaration<Long> salesQuantityStateDeclaration =
+ StateDeclarations.valueState("totalSalesQuantity", TypeDescriptors.LONG);
+
+ @Override
+ public Set<StateDeclaration> useWindowStates() {
+ return Set.of(salesQuantityStateDeclaration);
+ }
+
+ @Override
+ public void onRecord(
+ Order record,
+ Collector<Tuple2<Long, Long>> output,
+ PartitionedContext<Tuple2<Long, Long>> ctx,
+ OneInputWindowContext<Order> windowContext) throws Exception {
+ // get sales quantity from state
+ ValueState<Long> salesQuantityState = windowContext.getWindowState(salesQuantityStateDeclaration).get();
+ long salesQuantity = 0;
+ if (salesQuantityState.value() != null) {
+ salesQuantity = salesQuantityState.value();
+ }
+
+ // update sales quantity in state
+ salesQuantity += record.salesQuantity;
+ salesQuantityState.update(salesQuantity);
+ }
+
+ @Override
+ public void onTrigger(
+ Collector<Tuple2<Long, Long>> output,
+ PartitionedContext<Tuple2<Long, Long>> ctx,
+ OneInputWindowContext<Order> windowContext) throws Exception {
+ // get current productId
+ long productId = ctx.getStateManager().getCurrentKey();
+ // get sales quantity from state
+ ValueState<Long> salesQuantityState = windowContext.getWindowState(salesQuantityStateDeclaration).get();
+ long salesQuantity = salesQuantityState.value() == null ? 0 : salesQuantityState.value();
+ // emit result
+ output.collect(Tuple2.of(productId, salesQuantity));
+ }
+}
+```
+
+In the `CountSalesQuantityWithPreAggregation` function, we first declare a `ValueState` to store the total sales quantity for each product within the window.
+When input data arrives, we update this state accordingly, and we output the final result once the window is triggered.
+This approach allows us to avoid storing all input data in the window; instead, we only need to maintain a `ValueState` for each product within the window.
+
+
+{{< top >}}
diff --git a/docs/content/docs/dev/datastream-v2/context_and_state_processing.md b/docs/content/docs/dev/datastream-v2/context_and_state_processing.md
new file mode 100644
index 00000000000..cd17fe26d86
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/context_and_state_processing.md
@@ -0,0 +1,284 @@
+---
+title: Context and State Processing
+weight: 4
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Context and State Processing
+
+## Context
+
+Unlike attributes like the name of process operation, some information(such as current key) can only be obtained when the process function is executed.
+In order to build a bridge between process functions and the execution engine, DataStream API provide a unified entrypoint called Runtime Context.
+
+We divide all contextual into multiple parts according to their functions:
+
+- JobInfo: Hold all job related information, such as job name, execution mode.
+
+- TaskInfo: Hold all task related information, such as parallelism.
+
+- MetricGroup: Manage context related to metrics, such as registering a metric.
+
+- State Manager: Manage context related to state, such as accessing a specific state.
+
+- Watermark Manager: Manage context related to watermark, such as triggering a watermark.
+
+- ProcessingTime Manager: Manage context related to processing timer, such as getting the current processing time.
+
+Those context information can be classified into two categories:
+
+* NonPartitionedContext: Include JobInfo, TaskInfo, MetricGroup, WatermarkManager.
+* PartitionedContext: Include StateManager, ProcessingTimeManager.
+
+Typically, `PartitionedContext` is provided to a `ProcessFunction` when processing of data within
+a specific partition is expected. E.g., upon receiving of new records or triggering of timers.
+On the other hand, `NonPartitionedContext` is provided when no specific partition is relevant.
+E.g., initiating or cleaning-up of the `ProcessFunction`.
+
+The following code snippet shows how to get the parallelism and execution mode of the process function:
+
+```java
+new OneInputStreamProcessFunction<String, String>(){
+ private transient int parallelism;
+
+ private transient ExecutionMode executionMode;
+ @Override
+ public void open(NonPartitionedContext<String> ctx) throws Exception {
+ parallelism = ctx.getTaskInfo().getParallelism();
+ executionMode = ctx.getJobInfo().getExecutionMode();
+ }
+}
+```
+
+## State Processing
+
+State is the foundation of stateful computation. DataStream API supports declaring state in Process Function and
+accessing and updating state during data processing.
+
+In general, you should follow the principle of "declare first, use later.". In general,
+writing a stateful process function is divided into three steps:
+
+1. Defining the state in the form of `StateDeclaration`.
+2. Declaring the state in `ProcessFunction#usesStates`.
+3. Getting and updating state via `StateManager`.
+
+Before we go any further, let's look at what a stateful process function looks like:
+
+```java
+private static class StatefulFunction implements OneInputStreamProcessFunction<Long, Long> {
+ // Step1: Defining the state in the form of `StateDeclaration`.
+ static final StateDeclaration.ListStateDeclaration<Long> LIST_STATE_DECLARATION =
+ StateDeclarations.listStateBuilder("example-list-state", TypeDescriptors.LONG).build();
+
+ // Step2: Declaring the state in `ProcessFunction#usesStates`
+ @Override
+ public Set<StateDeclaration> usesStates() {
+ return Collections.singleton(LIST_STATE_DECLARATION);
+ }
+
+ @Override
+ public void processRecord(Long record, Collector<Long> output, RuntimeContext ctx)
+ throws Exception {
+ // Step3: Getting and updating state via `StateManager`.
+ Optional<ListState<Long>> stateOptional =
+ ctx.getStateManager().getState(LIST_STATE_DECLARATION);
+ ListState<Long> state = stateOptional.get();
+ // do something with this state. For example, update the state by this record.
+ state.update(Collections.singletonList(record));
+ }
+}
+```
+
+### Define State
+
+StateDeclaration is used to define a specific state, two types of information need to be provided for a state declaration:
+
+- Name: Used as the unique identifier of the state.
+
+- RedistributionMode: Defines how the state is redistributed between different partitions. For the keyed partition stream,
+since the state is bounded within the partition, no redistribution is required. But for the non-keyed partition stream,
+the partition will change with the parallelism, so the state must define how to redistribute.
+
+There are three RedistributionMode to choose:
+- NONE: Not supports redistribution.
+- REDISTRIBUTABLE: This state can be safely redistributed between different partitions, and the
+specific redistribution strategy is determined by the state itself.
+- IDENTICAL: States are guaranteed to be identical in different partitions, thus redistribution is
+not a problem.
+
+There are currently five type of `StateDeclaration`: `ValueStateDeclaration`, `ListStateDeclaration`, `MapStateDeclaration`, `ReducingStateDeclaration`, `AggregatingStateDeclaration` and `BroadcastStateDeclaration`
+that describe the `ValueState`, `ListState`, `MapState`, `ReducingState`, `AggregatingState` and `BroadcastState` respectively.
+
+* `ValueState<T>`: This keeps a value that can be updated and
+ retrieved. The value can be set using `update(T)` and retrieved using `T value()`.
+
+* `ListState<T>`: This keeps a list of elements. You can append elements and retrieve an `Iterable`
+ over all currently stored elements. Elements are added using `add(T)` or `addAll(List<T>)`, the Iterable can
+ be retrieved using `Iterable<T> get()`. You can also override the existing list with `update(List<T>)`
+
+* `ReducingState<T>`: This keeps a single value that represents the aggregation of all values
+ added to the state. The interface is similar to `ListState` but elements added using
+ `add(T)` are reduced to an aggregate using a specified `ReduceFunction`.
+
+* `AggregatingState<IN, OUT>`: This keeps a single value that represents the aggregation of all values
+ added to the state. Contrary to `ReducingState`, the aggregate type may be different from the type
+ of elements that are added to the state. The interface is the same as for `ListState` but elements
+ added using `add(IN)` are aggregated using a specified `AggregateFunction`.
+
+* `MapState<UK, UV>`: This keeps a list of mappings. You can put key-value pairs into the state and
+ retrieve an `Iterable` over all currently stored mappings. Mappings are added using `put(UK, UV)` or
+ `putAll(Map<UK, UV>)`. The value associated with a user key can be retrieved using `get(UK)`. The iterable
+ views for mappings, keys and values can be retrieved using `entries()`, `keys()` and `values()` respectively.
+ You can also use `isEmpty()` to check whether this map contains any key-value mappings.
+
+* `BroadcastState<K, V>`: This can be created to store the state of a `BroadcastStream`. This state
+assumes that the same elements are sent to all instances of a process function. You can put key-value pairs into the state and
+retrieve an `Iterable` over all currently stored mappings.
+
+For ease of use, DataStream API provide an auxiliary class called `StateDeclarations`, which encapsulates a series of methods for creating various StateDeclaration instances.
+
+For example:
+
+```java
+// create a value state declaration
+ValueStateDeclaration<Integer> valueStateDeclaration = StateDeclarations.valueState("example-value-state", TypeDescriptors.LONG);
+
+// create a map state declaration
+MapStateDeclaration<Long, String> mapStateDeclaration = StateDeclarations.mapState(
+ "example-map-state", TypeDescriptors.LONG, TypeDescriptors.STRING);
+
+// create a reducing state declaration with a `sum` reduce function
+ReducingStateDeclaration<Long> reducingStateDeclaration =
+ StateDeclarations.reducingState("example-reducing-state", TypeDescriptors.LONG, Long::sum);
+```
+
+To provide the type information of state, in the above example, the `TypeDescriptors` class provides a set of pre-defined type descriptors for common types, such as `INT`, `LONG`, `BOOLEAN`, `STRING`, `LIST`, `MAP`, etc.
+If your type is not in the pre-defined list, you can provide a class implements `TypeDescriptor` interface to define your own type descriptor.
+
+### Declare State
+
+As we said earlier, when using state in the DataStream API, you must follow the principle of "declare first, use later.".
+`ProcessFunction` has a method called `useStates` to explicitly declares states upfront.
+
+```java
+default Set<StateDeclaration> usesStates() {
+ return Collections.emptySet();
+}
+```
+
+For stateful functions, this method must be overridden and each specific state must be declared in this method before it can be used. This gives
+flink more opportunities to optimize the execution of functions.
+
+### Get and Update State
+
+The `StateManager` is used to manage the state for the process function. It provides methods for getting state, as well as method for retrieving current key.
+
+```java
+public interface StateManager {
+ /**
+ * Get the key of current record.
+ *
+ * @return The key of current processed record.
+ * @throws UnsupportedOperationException if the key can not be extracted for this function, for
+ * instance, get the key from a non-keyed partition stream.
+ */
+ <K> K getCurrentKey() throws UnsupportedOperationException;
+
+ /**
+ * Get the specific list state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the list state corresponds to the state declaration.
+ */
+ <T> Optional<ListState<T>> getState(ListStateDeclaration<T> stateDeclaration) throws Exception;
+
+ /**
+ * Get the specific value state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the value state corresponds to the state declaration.
+ */
+ <T> Optional<ValueState<T>> getState(ValueStateDeclaration<T> stateDeclaration)
+ throws Exception;
+
+ /**
+ * Get the specific map state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the map state corresponds to the state declaration.
+ */
+ <K, V> Optional<MapState<K, V>> getState(MapStateDeclaration<K, V> stateDeclaration)
+ throws Exception;
+
+ /**
+ * Get the specific reducing state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the reducing state corresponds to the state declaration.
+ */
+ <T> Optional<ReducingState<T>> getState(ReducingStateDeclaration<T> stateDeclaration)
+ throws Exception;
+
+ /**
+ * Get the specific aggregating state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the aggregating state corresponds to the state declaration.
+ */
+ <IN, ACC, OUT> Optional<AggregatingState<IN, OUT>> getState(
+ AggregatingStateDeclaration<IN, ACC, OUT> stateDeclaration) throws Exception;
+
+ /**
+ * Get the specific broadcast state.
+ *
+ * @param stateDeclaration of this state.
+ * @return the broadcast state corresponds to the state declaration.
+ */
+ <K, V> Optional<BroadcastState<K, V>> getState(BroadcastStateDeclaration<K, V> stateDeclaration)
+ throws Exception;
+}
+```
+
+You can get the `StateManager` from anywhere you have access to the `PartitionedContext` using `context#getStateManager()`.
+Next, using the `getState` method to obtain the state you declared, and then read and updated it.
+
+
+### The Legitimacy of State Declaration and Access
+
+In `ProcessFunction`, not all state declarations and accesses are legal, depending on the specific type of input streams.
+
+For `OneInputStreamProcessFunction` and `TwoOutputStreamProcessFunction`, the legality of state declaration and access are listed in the following table:
+
+{{< img src="/fig/datastream/one-input-state-access.png" alt="one-input-state-access" width="40%" >}}
+
+For the two inputs case, things get a little more complicated. Since different inputs may come from different type of streams, the legality of state access is also different for the data processing functions of each input edge.
+The specific rules are shown in the following table:
+
+{{< img src="/fig/datastream/two-input-state-access.png" alt="two-input-state-access" width="70%" >}}
+(K, NK, G, B denote the shorthand for Keyed, Non-Keyed, Global, Broadcast Stream, respectively)
+
+flink will check the legality of state declaration in advance. If an illegal state is declared in `usesStates` method, exception will be thrown at job compile time.
+
+{{< top >}}
diff --git a/docs/content/docs/dev/datastream-v2/overview.md b/docs/content/docs/dev/datastream-v2/overview.md
new file mode 100644
index 00000000000..d4f05c2a940
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/overview.md
@@ -0,0 +1,197 @@
+---
+title: Overview
+weight: 2
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Flink DataStream API Programming Guide
+
+DataStream programs in Flink are regular programs that implement transformations on data streams
+(e.g., filtering, updating state, defining windows, aggregating). The data streams are initially created from various
+sources (e.g., message queues, socket streams, files). Results are returned via sinks, which may for
+example write the data to files, or to standard output (for example the command line
+terminal). Flink programs run in a variety of contexts, standalone, or embedded in other programs.
+The execution can happen in a local JVM, or on clusters of many machines.
+
+## What is a DataStream?
+
+The DataStream API gets its name from the special `DataStream` class that is
+used to represent a collection of data in a Flink program. You can think of
+them as immutable collections of data that can contain duplicates. This data
+can either be finite or unbounded, the API that you use to work on them is the
+same.
+
+A `DataStream` is similar to a regular Java `Collection` in terms of usage but
+is quite different in some key ways. They are immutable, meaning that once they
+are created you cannot add or remove elements. You can also not simply inspect
+the elements inside but only work on them using the `DataStream` API
+operations, which are also called transformations.
+
+You can create an initial `DataStream` by adding a source in a Flink program.
+Then you can derive new streams from this and combine them by using API methods
+such as `process`, `connectAndProcess`, and so on.
+
+## Fundamental Primitives and Extensions
+
+Based on whether its functionality must be provided by flink, we divide the relevant concepts in DataStream API
+into two categories: fundamental primitives and high-level extensions.
+
+### Fundamental primitives
+
+Fundamental primitives are the basic and necessary semantics that flink need to provide in order to
+define a stateful stream processing application, which cannot be achieved by users if not provided by
+the framework. It includes data stream, partitioning, process function, state, processing timer service,
+watermark.
+
+More details can be found in:
+- [Building Blocks]({{< ref "docs/dev/datastream-v2/building_blocks" >}}): Given the most basic elements of DataStream API.
+- [State Processing]({{< ref "docs/dev/datastream-v2/context_and_state_processing" >}}): Explanation of how to develop stateful applications.
+- [Time Processing # Processing Timer Service]({{< ref "docs/dev/datastream-v2/time-processing/processing_timer_service" >}}): Explanation of how to handle processing time.
+- [Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}}): Explanation of how to define and handle user defined events.
+
+### High-Level Extensions
+
+High-Level extensions are like short-cuts / sugars, without which users can probably still achieve the same
+behavior by working with the fundamental APIs, but would be a lot easier with the builtin supports.
+It includes event timer service, window and join.
+
+More details can be found in:
+- [Time Processing # Event Timer Service]({{< ref "docs/dev/datastream-v2/time-processing/event_timer_service" >}}): Explanation of how to handle event time via extension.
+- [Builtin Functions]({{< ref "docs/dev/datastream-v2/builtin-funcs/windows" >}}): Explanation of how to do window aggregation and join via extension.
+
+## Anatomy of a Flink DataStream Program
+
+Flink programs look like regular programs that transform `DataStream`. Each
+program consists of the same basic parts:
+
+1. Obtain an `Execution Environment`,
+2. Load/create the initial data,
+3. Specify transformations on this data,
+4. Specify where to put the results of your computations,
+5. Trigger the program execution
+
+### Obtain an `Execution Environment`
+
+We will now give an overview of each of those steps, please refer to the
+respective sections for more details.
+
+The `ExecutionEnvironment` is the basis for all Flink programs. You can
+obtain one using these static methods on `ExecutionEnvironment`:
+
+```java
+ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+```
+
+If you are executing your program inside an IDE or as a regular Java program it will create a local environment
+that will execute your program on your local machine. If you created a JAR file
+from your program, and invoke it through the command line, the Flink cluster manager will execute your main method and
+`ExecutionEnvironment.getInstance()` will return an execution environment for executing
+your program on a cluster.
+
+
+### Load/create the Initial Data
+
+Sources are where your program reads its input from. You can attach a source to your program by
+using `ExecutionEnvironment.fromSource(source, sourceName)`. Flink comes with a number of pre-implemented
+source, you can use [FLIP-27](https://cwiki.apache.org/confluence/x/co_zBQ) based source via `DataStreamV2SourceUtils.wrapSource(source)` or use
+`DataStreamV2SourceUtils.fromData(collection)` for testing/debugging purpose.
+
+As an example, to just read data from a predefined collection, you can use:
+
+```java
+ExecutionEnvironment env = ExecutionEnvironment.getInstance();
+
+NonKeyedPartitionStream<String> input =
+ env.fromSource(
+ DataStreamV2SourceUtils.fromData(Arrays.asList("1", "2", "3")),
+ "source"
+ );
+```
+
+Since data from source has no clear partitioning, this will give you a `NonKeyedPartitionStream` on which you can then apply transformations to create new
+derived DataStreams. For other types of DataStream, see [Building Blocks # DataStream]({{< ref "docs/dev/datastream-v2/building_blocks" >}}#datastream).
+
+### Specify Transformations on this Data
+
+You apply transformations by calling methods on DataStream with a
+`ProcssFunction`(using lambda expression here for simplicity). For example, a map transformation looks like this:
+
+```java
+NonKeyedPartitionStream<String> input = ...;
+
+NonKeyedPartitionStream<Integer> parsed = source.process(
+ (OneInputStreamProcessFunction<String, Integer>)
+ (record, output, ctx) -> {
+ output.collect(Integer.parseInt(record));
+ });
+```
+
+This will create a new DataStream by converting every String in the original
+collection to an Integer. For more details of processing, see [Building Blocks # Process Function]({{< ref "docs/dev/datastream-v2/building_blocks" >}}#processfunction).
+
+### Specify Where to Put the Results of Your Computations
+
+Once you have a DataStream containing your final results, you can write it to
+an external system by creating a sink.
+
+Data sinks consume DataStreams and forward them to files, sockets, external systems, or print them.
+Flink comes with a variety of built-in sink implementations, you can use SinkV2 based sink via `DataStreamV2SinkUtils.wrapSink(sink)`.
+
+These are just an example for creating a sink to print the results:
+
+```java
+parsed.toSink(DataStreamV2SinkUtils.wrapSink(new PrintSink<>()));
+```
+
+### Trigger the Program Execution
+
+Once you specified the complete program you need to **trigger the program
+execution** by calling `execute()` on the `ExecutionEnvironment`.
+Depending on the type of the `ExecutionEnvironment` the execution will be
+triggered on your local machine or submit your program for execution on a
+cluster. The `execute()` method will wait for the job to finish.
+
+That last part about program execution is crucial to understanding when and how
+Flink operations are executed. All Flink programs are executed lazily: When the
+program's main method is executed, the data loading and transformations do not
+happen directly. Rather, each operation is created and added to a dataflow
+graph. The operations are actually executed when the execution is explicitly
+triggered by an `execute()` call on the execution environment. Whether the
+program is executed locally or on a cluster depends on the type of execution
+environment.
+
+The lazy evaluation lets you construct sophisticated programs that Flink
+executes as one holistically planned unit.
+
+Where to go next?
+-----------------
+
+* [Building Blocks]({{< ref "docs/dev/datastream-v2/building_blocks" >}})
+* [Context and State Processing]({{< ref "docs/dev/datastream-v2/context_and_state_processing" >}})
+* [Time Processing]({{< ref "docs/dev/datastream-v2/time-processing/processing_timer_service" >}})
+* [Builtin Functions]({{< ref "docs/dev/datastream-v2/builtin-funcs/windows" >}})
+* [Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}})
+
+{{< top >}}
diff --git a/docs/content/docs/dev/datastream-v2/time-processing/_index.md b/docs/content/docs/dev/datastream-v2/time-processing/_index.md
new file mode 100644
index 00000000000..28f0422cdc4
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/time-processing/_index.md
@@ -0,0 +1,23 @@
+---
+title: Time Processing
+bookCollapseSection: true
+weight: 5
+---
+<!--
+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.
+-->
diff --git a/docs/content/docs/dev/datastream-v2/time-processing/event_timer_service.md b/docs/content/docs/dev/datastream-v2/time-processing/event_timer_service.md
new file mode 100644
index 00000000000..16b5dc64d30
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/time-processing/event_timer_service.md
@@ -0,0 +1,429 @@
+---
+title: "Event Timer Service"
+weight: 9
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Event Timer Service
+
+The event timer service is a high-level extension of the Flink DataStream API, provided by Flink.
+It enables users to register timers for executing calculations at specific event time points and helps determine when to trigger windows within the Flink framework.
+
+For a comprehensive explanation of event time, please refer to the section on [Notions of Time: Event Time and Processing Time]({{< ref "docs/concepts/time" >}}#notions-of-time-event-time-and-processing-time).
+
+In this section, we will introduce how to utilize the event timer service within the Flink DataStream API.
+
+We use a special type of [Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}}) to denote the progression of event time in the stream,
+which we refer to as the event time watermark. Additionally, to dealing idle inputs or sources, we implement another type of
+[Watermark]({{< ref "docs/dev/datastream-v2/watermark" >}}) to indicate that the input or source
+is idle.
+This is known as the idle status watermark. For more information, please refer to [Dealing With Idle Inputs / Sources](#dealing-with-idle-inputs--sources).
+In this document, we collectively refer to the event time watermark and the idle status watermark as event-time related watermarks.
+
+The core of the event timer service lies in generating and propagating event-time related watermarks through the streams.
+To achieve this, we need to address two aspects:
+1. How to generate event-time related watermarks.
+2. How to handle event-time related watermarks.
+
+The following will introduce these two aspects.
+
+## Generate Event-Time related Watermarks
+
+In order to work with *event time*, Flink needs to know the events
+*timestamps*, meaning each element in the stream needs to have its event
+timestamp *assigned*. This is usually done by accessing/extracting the
+timestamp from some field in the element.
+
+Once the timestamps have been extracted, Flink generates event time watermarks.
+There are two methods for doing so: one is to use the `EventTimeWatermarkGeneratorBuilder` provided by Flink,
+and the other is to implement a custom `ProcessFunction`.
+
+Users should select one of these two approaches to suit their needs.
+
+### Generate Watermarks by WatermarkGeneratorBuilder
+
+Users can utilize the `EventTimeWatermarkGeneratorBuilder` to generate event-time related watermarks.
+There are four aspects of the `EventTimeWatermarkGeneratorBuilder` that users can configure:
+
+1. [Required] `EventTimeExtractor`
+
+ This function instructs Flink on how to extract the event time from each record.
+
+3. [Optional] Input Idle Timeout
+
+ If the input stream remains idle for a specified duration, Flink will ignore this input when
+combining event-time related watermarks to prevent stalling the progression of event time. The default value for this timeout is 0.
+
+ For more information, please refer to the [Dealing With Idle Inputs / Sources](#dealing-with-idle-inputs--sources).
+
+4. [Optional] Out-of-Order Time
+
+ To accommodate the disorder of input records, user can set a maximum out-of-order time for the event time watermark.
+The default value for this setting is also 0.
+
+ For more information, please refer to the [Fixed Amount of Lateness](#fixed-amount-of-lateness).
+
+5. [Optional] Generation Frequency
+
+ Flink offers three scenarios for generating event-time related watermarks:
+ - No EventTimeWatermarks are generated and emitted.
+ - EventTimeWatermarks are generated and emitted periodically.
+ - EventTimeWatermarks are generated and emitted for each event.
+
+ By default, Flink adopts the second approach, which involves periodically generating and emitting
+event-time related watermarks. The interval for this periodic generation is determined by the configuration
+"pipeline.auto-watermark-interval."
+
+After configuring and obtaining an instance of the `EventTimeWatermarkGeneratorBuilder`,
+users can utilize it to build a `ProcessFunction`. This `ProcessFunction` will extract the event time
+from each record and generates the corresponding event-time related watermark.
+
+A typical example of using `EventTimeWatermarkGeneratorBuilder` is shown in below:
+
+```java
+NonKeyedPartitionStream stream = ...;
+
+EventTimeWatermarkGeneratorBuilder<POJO> builder = EventTimeExtension
+ .newWatermarkGeneratorBuilder(pojo -> pojo.getTimeStamp()) // set event time extractor
+ .withIdleness(Duration.ofSeconds(10)) // set input idle timeout
+ .withMaxOutOfOrderTime(Duration.ofSeconds(30)) // set max out-of-order time
+ .periodicWatermark(Duration.ofMillis(200)); // set periodic watermark generation interval
+
+stream.process(builder.buildAsProcessFunction())
+ .process(...);
+```
+
+{{< hint warning >}}
+**Attention**: Both timestamps and event time watermarks
+are specified as milliseconds since the Java epoch of 1970-01-01T00:00:00Z.
+{{< /hint >}}
+
+#### Fixed Amount of Lateness
+
+There are some scenarios where the maximum lateness that can be
+encountered in a stream is known in advance, e.g. when creating a custom source
+containing elements with timestamps spread within a fixed period of time for
+testing.
+
+Users can deal these cases with `EventTimeWatermarkGeneratorBuilder#withMaxOutOfOrderTime`,
+i.e. the maximum amount of time an element is allowed to be late before being ignored when
+computing the final result for the given window. Lateness corresponds to the
+result of `t - t_w`, where `t` is the (event-time) timestamp of an element, and
+`t_w` that of the previous event time watermark. If `lateness > 0` then the element is
+considered late and is, by default, ignored when computing the result of the
+job for its corresponding window.
+
+#### Dealing With Idle Inputs / Sources
+
+If one of the input splits/partitions/shards does not carry events for a while
+this means that the `EventTimeWatermarkGeneratorBuilder` also does not get any new information
+on which to base a watermark. We call this an *idle input* or an *idle source*.
+This is a problem because it can happen that some of your partitions do still
+carry events. In that case, the event time watermark will be held back, because it is
+computed as the minimum over all the different parallel event time watermarks.
+
+To deal with this, you can configure an idleness timeout with
+`EventTimeWatermarkGeneratorBuilder#withIdleness` that will detect idleness
+and mark an input as idle.
+
+If an input remains idle, the `EventTimeWatermarkGeneratorBuilder` will emit an idle status watermark
+to indicate that the input is inactive. Consequently, downstream `ProcessFunction` instances will
+disregard this input when combining event time watermarks.
+
+### Generate Watermarks by Custom ProcessFunction
+
+Users can create and send event-time related watermarks by customizing the `ProcessFunction`.
+To do this, they need to follow these steps:
+1. Declare the `EventTimeWatermarkDeclaration` in the custom `ProcessFunction`, and the `IdleStatusWatermarkDeclaration` if support for input idle is required.
+2. Create the event-time related watermark using the `EventTimeWatermarkDeclaration`.
+3. (Optional) Create the idle status watermark using the `IdleStatusWatermarkDeclaration`.
+4. Send the watermarks through the `WatermarkManager`.
+
+The following shows an example of this approach.
+
+```java
+public static class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Integer, Integer> {
+
+
+ @Override
+ public Collection<? extends WatermarkDeclaration> watermarkDeclarations() {
+ return Set.of(
+ EventTimeExtension.EVENT_TIME_WATERMARK_DECLARATION,
+ EventTimeExtension.IDLE_STATUS_WATERMARK_DECLARATION
+ );
+ }
+
+ @Override
+ public void processRecord(Integer record, Collector<Integer> output, PartitionedContext ctx)
+ throws Exception {
+ // do something as needed
+
+ long eventTime = ... // get event time from record
+ // generate event time watermark and send to downstrea
+ LongWatermark eventTimeWatermark = EventTimeExtension.EVENT_TIME_WATERMARK_DECLARATION.newWatermark(eventTime);
+ ctx.getNonPartitionedContext()
+ .getWatermarkManager()
+ .emitWatermark(eventTimeWatermark);
+ }
+}
+```
+
+{{< hint warning >}}
+**Attention**: Both timestamps and event time watermarks
+are specified as milliseconds since the Java epoch of 1970-01-01T00:00:00Z.
+{{< /hint >}}
+
+## Handle Event-Time related Watermarks
+
+As described in [Generate Event-Time related Watermarks](#generate-event-time-related-watermarks),
+Flink provides abstractions that enable programmers to assign their own timestamps and emit event-time related watermarks.
+
+Once event-time related watermarks are generated and propagated, Flink can use them to determine the
+event time of the program, for example triggering a window when the event time watermark
+exceeds the end time of the Window.
+
+In addition, users can use event-time related watermarks to implement their own business logic.
+Currently, Flink provide users with two ways to use event-time related watermarks, one is to use the built-in
+`EventTimeProcessFunction`, and the other is to use the most basic `ProcessFunction` to
+process event-time related watermarks.
+
+It should be noted that the first method allows users to quickly register and unregister
+event timers, while the second method requires users to implement this function themselves.
+Therefore, when you need to use or perceive event time in your program, we recommend using
+the first method because it is simple and efficient.
+
+Each approach is described in detail below, users should select one of these to suit their needs.
+
+### Handle Event-Time related Watermarks by EventTimeProcessFunction
+
+`EventTimeProcessFunction` is provided by Flink and is a wrapper for `ProcessFunction` that needs to use event time.
+
+When utilizing the `EventTimeProcessFunction`, it's important for user's custom `ProcessFunction`
+to implement the `OneInputEventTimeStreamProcessFunction` / `TwoInputBroadcastEventTimeStreamProcessFunction`
+/ `TwoInputNonBroadcastEventTimeStreamProcessFunction` / `TwoOutputEventTimeStreamProcessFunction` interface
+instead of the `OneInputStreamProcessFunction` / `TwoInputBroadcastStreamProcessFunction`
+/ `TwoInputNonBroadcastStreamProcessFunction` / `TwoOutputStreamProcessFunction` interface.
+
+Below is the interface of the `EventTimeProcessFunction` and `OneInputEventTimeStreamProcessFunction`:
+
+```java
+@Experimental
+public interface EventTimeProcessFunction extends ProcessFunction {
+ /**
+ * Initialize the {@link EventTimeProcessFunction} with an instance of {@link EventTimeManager}.
+ * Note that this method should be invoked before the open method.
+ */
+ void initEventTimeProcessFunction(EventTimeManager eventTimeManager);
+}
+
+@Experimental
+public interface OneInputEventTimeStreamProcessFunction<IN, OUT>
+ extends EventTimeProcessFunction, OneInputStreamProcessFunction<IN, OUT> {
+
+ /**
+ * The {@code #onEventTimeWatermark} method signifies that the EventTimeProcessFunction has
+ * received an EventTimeWatermark. Other types of watermarks will be processed by the {@code
+ * ProcessFunction#onWatermark} method.
+ */
+ default void onEventTimeWatermark(
+ long watermarkTimestamp, Collector<OUT> output, NonPartitionedContext<OUT> ctx)
+ throws Exception {}
+
+ /**
+ * Invoked when an event-time timer fires. Note that it is only used in {@link
+ * KeyedPartitionStream}.
+ */
+ default void onEventTimer(long timestamp, Collector<OUT> output, PartitionedContext<OUT> ctx) {}
+}
+```
+
+There are three methods that should be noted:
+- initEventTimeProcessFunction
+ - This method allows the `EventTimeProcessFunction` to obtain an instance of `EventTimeManager`.
+ Users can use this instance to access the current event time and to create or delete event timers as needed. Note that the event timer is only used in Keyed Partition Stream.
+
+- onEventTimeWatermark
+ - This method signifies that the `EventTimeProcessFunction` has received an event time watermark.
+ It is important to note in `EventTimeProcessFunction`, the event time watermarks will be processed
+ by `EventTimeProcessFunction#onEventTimeWatermark`, whereas other types of watermarks will be processed by the `EventTimeProcessFunction#onWatermark`.
+
+- onEventTimer
+ - This callback method is triggered by the event timer. Within this method, users can access
+ the key and event time associated with the event timer, perform necessary calculations, and output the results.
+
+Here is an example of how to implement `EventTimeProcessFunction`:
+
+```java
+class CustomEventTimeProcessFunction
+ implements OneInputEventTimeStreamProcessFunction<InputPojo, OutputPojo> {
+
+ private EventTimeManager eventTimeManager;
+
+ @Override
+ public void initEventTimeProcessFunction(EventTimeManager eventTimeManager) {
+ // get event time manager instance
+ this.eventTimeManager = eventTimeManager;
+ }
+
+ @Override
+ public void processRecord(
+ InputPojo record,
+ Collector<OutputPojo> output,
+ PartitionedContext<OutputPojo> ctx)
+ throws Exception {
+ ...
+
+ // register event timer
+ eventTimeManager.registerTimer(targetTimestamp);
+ }
+
+ @Override
+ public void onEventTimeWatermark(
+ long watermarkTimestamp,
+ Collector<OutputPojo> output,
+ NonPartitionedContext<OutputPojo> ctx)
+ throws Exception {
+ // sense event time watermark arrival
+ }
+
+ @Override
+ public void onEventTimer(
+ long timestamp,
+ Collector<OutputPojo> output,
+ PartitionedContext<OutputPojo> ctx) {
+ // write your event timer callback here
+ }
+}
+```
+
+After implementing the `EventTimeProcessFunction`, users should wrap their custom function using
+`EventTimeUtils#wrapProcessFunction`. This step is essential as it provides the necessary components,
+including an instance of `EventTimeManager`, and declares the built-in state required for timers
+and other functionalities.
+
+Here is an example of how to wrap the `EventTimeProcessFunction`:
+
+```java
+NonKeyedPartitionStream stream = ...;
+stream.keyBy(x -> x.getKey())
+ .process(EventTimeExtension.wrapProcessFunction(new CustomEventTimeProcessFunction()))
+ .process(...);
+```
+
+
+### Handle Event-Time related Watermarks by Custom ProcessFunction
+
+Similarly, users can handle event-time related watermarks by implementing `ProcessFunction` instead of `EventTimeProcessFunction`.
+
+In this approach, users must evaluate whether the current watermark received is an event time watermark
+or idle status watermark within the `ProcessFunction#onWatermark` method and execute the appropriate
+processing logic accordingly.
+
+An example is provided below.
+
+```java
+public static class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Integer, Integer> {
+
+ @Override
+ public WatermarkHandlingResult onWatermark(
+ Watermark watermark,
+ Collector<Integer> output,
+ NonPartitionedContext<Integer> ctx) throws Exception {
+ if (EventTimeExtension.isEventTimeWatermark(watermark)) {
+ // do something as needed
+ ...
+ return WatermarkHandlingResult.PEEK;
+ } else if (EventTimeExtension.isIdleStatusWatermark(watermark)) {
+ // do something as needed
+ ...
+ return WatermarkHandlingResult.PEEK;
+ } else {
+ // do something as needed
+ ...
+ }
+ }
+}
+```
+
+It is important to note that when `ProcessFunction#onWatermark` handles event-time related watermarks,
+it should return `WatermarkHandlingResult#PEEK`. This indicates that the Flink framework will choose
+the processing logic based on the watermark definition.
+For event time watermark and idle status watermark, the Flink framework will forward the watermark downstream.
+
+Conversely, if `ProcessFunction#onWatermark` returns `WatermarkHandlingResult#POP`, the watermarks
+will not be sent downstream by the Flink framework. Users should be aware that this may result in
+the loss of the watermark, or they may need to send the watermark manually.
+
+## Example
+Here is an example of how to use event timer service in Flink:
+
+```java
+NonKeyedPartitionStream<POJO> source = ...;
+
+source.process(
+ EventTimeExtension.<POJO>newWatermarkGeneratorBuilder(pojo -> pojo.getTimestamp())
+ .periodicWatermark(Duration.ofMillis(200))
+ .buildAsProcessFunction()
+ )
+ .keyBy(pojo -> pojo.getKey())
+ .process(EventTimeExtension.wrapProcessFunction(
+ new OneInputEventTimeStreamProcessFunction<POJO, String>() {
+
+ private EventTimeManager eventTimeManager;
+
+ @Override
+ public void initEventTimeProcessFunction(EventTimeManager eventTimeManager) {
+ // get event time manager instance
+ this.eventTimeManager = eventTimeManager;
+ }
+
+ @Override
+ public void processRecord(
+ POJO record,
+ Collector<String> output,
+ PartitionedContext<String> ctx) throws Exception {
+ ...
+
+ // register event timer
+ eventTimeManager.registerTimer(targetTimestamp);
+ }
+
+ @Override
+ public void onEventTimer(
+ long timestamp,
+ Collector<String> output,
+ PartitionedContext<String> ctx) {
+ // write your event timer callback here
+ }
+
+ }
+ )
+ )
+ .toSink(...);
+```
+{{< top >}}
diff --git a/docs/content/docs/dev/datastream-v2/time-processing/processing_timer_service.md b/docs/content/docs/dev/datastream-v2/time-processing/processing_timer_service.md
new file mode 100644
index 00000000000..d94d868d797
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/time-processing/processing_timer_service.md
@@ -0,0 +1,110 @@
+---
+title: "Processing Timer Service"
+weight: 8
+type: docs
+aliases:
+---
+<!--
+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.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Processing Timer Service
+
+The process timer service is a fundamental primitive of the Flink DataStream API, provided by Flink.
+It allows users to register timers for performing calculations at specific processing time points.
+
+For a comprehensive explanation of processing time, please refer to the section on [Notions of Time: Event Time and Processing Time]({{< ref "docs/concepts/time" >}}#notions-of-time-event-time-and-processing-time).
+
+In this section, we will introduce how to utilize the processing timer service within the Flink DataStream API.
+
+## ProcessingTimerManager
+
+The `ProcessingTimerManager` is the key component for utilizing the processing timer service.
+You can obtain it by calling `PartitionedContext#getProcessingTimeManager` within a `ProcessFunction`.
+Below is the interface for the `ProcessingTimerManager`:
+
+```java
+@Experimental
+public interface ProcessingTimeManager {
+ /**
+ * Register a processing timer for this process function. `onProcessingTimer` method of this
+ * function will be invoked as callback if the timer expires.
+ *
+ * @param timestamp to trigger timer callback.
+ */
+ void registerTimer(long timestamp);
+
+ /**
+ * Deletes the processing-time timer with the given trigger timestamp. This method has only an
+ * effect if such a timer was previously registered and did not already expire.
+ *
+ * @param timestamp indicates the timestamp of the timer to delete.
+ */
+ void deleteTimer(long timestamp);
+
+ /**
+ * Get the current processing time.
+ *
+ * @return current processing time.
+ */
+ long currentTime();
+}
+```
+
+As shown above, `ProcessingTimerManager` has three methods: `registerTimer`, `deleteTimer`, and `currentTime`.
+When the target time of `registerTimer` arrives, Flink will call `ProcessFunction#onProcessingTimer`,
+and users should write their own calculation logic in `ProcessFunction#onProcessingTimer`.
+
+It should be noted that:
+1. for timers with the same target time, Flink will only retain one timer.
+This means that if multiple timers with the same target time are registered, `ProcessFunction#onProcessingTimer` will only be invoked once.
+2. the `ProcessingTimerManager` can only be used in Keyed Partitioned Stream.
+
+## Example
+Here is an example of using the processing timer service:
+
+```java
+public class CustomProcessFunction implements OneInputStreamProcessFunction<String, String> {
+
+ @Override
+ public void processRecord(
+ String record,
+ Collector<String> output,
+ PartitionedContext<String> ctx) throws Exception {
+ // do some calculation as needed
+
+ // register a timer with a target time of one minute after the current time
+ long currentTime = ctx.getProcessingTimeManager().currentTime();
+ ctx.getProcessingTimeManager().registerTimer(currentTime + Duration.ofMinutes(1L).toMillis());
+ }
+
+ @Override
+ public void onProcessingTimer(
+ long timestamp,
+ Collector<String> output,
+ PartitionedContext<String> ctx) {
+ // do some calculation and output result as needed
+ }
+}
+```
+
+{{< top >}}
diff --git a/docs/content/docs/dev/datastream-v2/watermark.md b/docs/content/docs/dev/datastream-v2/watermark.md
new file mode 100644
index 00000000000..da54198354b
--- /dev/null
+++ b/docs/content/docs/dev/datastream-v2/watermark.md
@@ -0,0 +1,203 @@
+---
+title: Watermark
+weight: 11
+type: docs
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+{{< hint warning >}}
+**Note:** DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production.
+{{< /hint >}}
+
+# Watermark
+
+Before introducing Watermark, users should be aware that Watermark in DataStream V2 does not
+refer to the [original Watermark]({{< ref "docs/concepts/time" >}}#event-time-and-watermarks) that
+measure progress in event time, but is a special event that can be customized by the user and
+can be propagated along the streams.
+
+Using Watermark in Flink involves three key steps:
+1. Define and Declare the Watermark
+2. Emit the Watermark
+3. Handle the Watermark
+
+Let's follow these steps to understand how to use Watermark in Flink.
+
+## Define and Declare Watermark
+
+A watermark is a special event that carries data. It can be emitted from a `Source`
+or `ProcessFunction`, propagating along streams, received by downstream `ProcessFunction`.
+When defining a Watermark, there are four aspects need to consider:
+
+1. [Required] Watermark Identifier
+
+ Since multiple types of Watermarks can propagate in streams, it's essential to assign an
+identifier to each Watermark for differentiation.
+
+ Note that the identifier must be a String, is case-sensitive, and must be globally unique within the entire job.
+
+2. [Required] Watermark Data Type
+
+ It is important to specify the data type of the Watermark. Currently, Flink supports two types: Long and Bool.
+
+3. [Required] Combine Function and `combineWaitForAllChannels`
+
+ A `ProcessFunction` may receive multiple Watermarks from different input channels due to
+having multiple upstream inputs, each potentially with varying degrees of parallelism.
+In such cases, users often wish to combine Watermarks from input channels before outputting
+to the `ProcessFunction`.
+
+ Flink supports the following combination functions:
+ - For `Long` type Watermark:
+ - `MIN`: Retains and outputs the minimum value of all received watermarks.
+ - `MAX`: Retains and outputs the maximum value of all received watermarks.
+ - For `Bool` type Watermark:
+ - `AND`: Retains and outputs the logical AND result of all received watermarks.
+ - `OR`: Retains and outputs the logical OR result of all received watermarks.
+
+ Additionally, users can configure whether the combining process should wait until the
+`ProcessFunction` has received Watermarks from all upstream channels. This is particularly useful
+in some scenarios. For example, the event time watermark need to wait receives the watermarks from
+all inputs and then combine them. This ensures that the time carried by the event time watermark does not decrease.
+By default, the `combineWaitForAllChannels` setting is false.
+
+4. [Optional] `WatermarkHandlingStrategy` by Framework
+
+ The `WatermarkHandlingStrategy` determines whether the framework should send the watermark
+to downstream `ProcessFunction` when the user-defined `ProcessFunction` does pop the Watermark.
+There are two options of this setting:
+ - IGNORE: The framework shouldn't take any action.
+ - FORWARD: The framework should send the watermark to downstream.
+
+ This optional setting is useful in some cases. For example, setting it to `IGNORE` can indicate
+that the framework does not need to propagate this Watermark, but rather that it is up to the user
+to control its sending.
+
+To simplify the process of defining a Watermark, Flink offer a `WatermarkBuilder`.
+This builder ultimately creates a `WatermarkDeclaration` object.
+Below is an example demonstrating how to use it to defining a Watermark:
+
+```java
+LongWatermarkDeclaration watermarkDeclaration = WatermarkDeclarations
+ .newBuilder("MY_CUSTOM_WATERMARK_IDENTIFIER")
+ .typeLong()
+ .combineFunctionMax()
+ .combineWaitForAllChannels(true)
+ .defaultHandlingStrategyForward()
+ .build();
+```
+
+Once users have defined the Watermark, it is essential to declare it in the
+`ProcessFunction#declareWatermarks` or `Source#declareWatermarks`. This step allows the framework to recognize it properly.
+Here’s an example of how to declare the Watermark in the `ProcessFunction`:
+
+```java
+public class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Long, Long> {
+
+ LongWatermarkDeclaration watermarkDeclaration = ...;
+
+ @Override
+ public Set<? extends WatermarkDeclaration> declareWatermarks() {
+ return Set.of(watermarkDeclaration);
+ }
+}
+```
+
+Please note that each type of Watermark only needs to be declared once in a job.
+
+## Emit Watermark
+
+Users can utilize the Watermark declaration to create a watermark. Below is an example of how to generate a Long type watermark with a value of 1:
+
+```java
+LongWatermarkDeclaration watermarkDeclaration = ...;
+LongWatermark watermark = watermarkDeclaration.newWatermark(1);
+```
+
+And users can emit Watermark in `ProcessFunction` by calling `nonPartitionedContext.getWatermarkManager().emitWatermark(watermark)`,
+it is also support emit Watermark in `Source` by calling `sourceReaderContext.emitWatermark(watermark)`.
+Here is an example of how to emit a Watermark in a `ProcessFunction`:
+```java
+public class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Long, Long> {
+
+ LongWatermarkDeclaration watermarkDeclaration = ...;
+
+ @Override
+ public Set<? extends WatermarkDeclaration> declareWatermarks() {
+ return Set.of(watermarkDeclaration);
+ }
+
+ @Override
+ public void processRecord(Long record, Collector<Long> output, PartitionedContext<Long> ctx)
+ throws Exception {
+ // do something as needed
+
+ // generate and emit Watermark
+ LongWatermark watermark = watermarkDeclaration.newWatermark(1);
+ ctx.getNonPartitionedContext().getWatermarkManager().emitWatermark(watermark);
+ }
+}
+```
+
+## Handle Watermark
+
+Once the `ProcessFunction` receives a Watermark, the framework will invoke the
+`ProcessFunction#onWatermark` method to process it. Therefore, users will need to override
+`ProcessFunction#onWatermark` in order to handle the Watermark appropriately.
+
+The return value of the `ProcessFunction#onWatermark` method is of type `WatermarkHandlingResult`,
+which has two possible options:
+
+- PEEK: `ProcessFunction` only peek the Watermark, and it's framework's responsibility to handle this watermark.
+
+ The framework will forward/ignore the Watermark according to the `WatermarkHandlingStrategy` associated with it.
+
+- POLL: This Watermark should be sent to downstream by process function itself. The framework does no additional processing.
+
+By default, the `ProcessFunction#onWatermark` method returns `WatermarkHandlingResult.PEEK`.
+
+Here is an example of how to handle Watermarks in `ProcessFunction`:
+
+```java
+
+public static final String CUSTOM_WATERMARK_IDENTIFIER = "CUSTOM_WATERMARK_IDENTIFIER";
+
+public class CustomProcessFunction
+ implements OneInputStreamProcessFunction<Long, Long> {
+
+ ...
+
+ @Override
+ public WatermarkHandlingResult onWatermark(
+ Watermark watermark, Collector<Long> output, NonPartitionedContext<Long> ctx) {
+ // For Watermark that this ProcessFunction is interested in, process the watermark
+ if (watermark.getIdentifier().equals(CUSTOM_WATERMARK_IDENTIFIER)) {
+ // do something as needed
+ }
+
+ // For other Watermarks, return PEEK
+ return WatermarkHandlingResult.PEEK;
+ }
+}
+```
+
+{{< top >}}
diff --git a/docs/static/fig/datastream/one-input-partitioning.png b/docs/static/fig/datastream/one-input-partitioning.png
new file mode 100644
index 00000000000..10e96e8a30f
Binary files /dev/null and b/docs/static/fig/datastream/one-input-partitioning.png differ
diff --git a/docs/static/fig/datastream/one-input-state-access.png b/docs/static/fig/datastream/one-input-state-access.png
new file mode 100644
index 00000000000..a021ec91d3a
Binary files /dev/null and b/docs/static/fig/datastream/one-input-state-access.png differ
diff --git a/docs/static/fig/datastream/two-input-state-access.png b/docs/static/fig/datastream/two-input-state-access.png
new file mode 100644
index 00000000000..9b5480c4c1b
Binary files /dev/null and b/docs/static/fig/datastream/two-input-state-access.png differ