WIP: 128: produce using NATIVE_AVRO in SQL sink when choosing avro format to allow Pulsar consumer consume

.github/CODEOWNERS

@@ -0,0 +1 @@
+*                    @streamnative/compute 

.github/README.md

@@ -0,0 +1,11 @@
+# StreamNative's fork of Apache Flink
+
+This repository is maintained by StreamNative to host a StreamNative fork of the Flink Connector for Pulsar. Only the Flink Connector for Pulsar (including the Flink SQL Connector for Pulsar and the Flink DataStream Connector for Pulsar) diverges with the official Apache Flink repository.
+
+The following is some useful documentation and resources for the DataStream and SQL Connectors
+
+- [Flink SQL Connector for Pulsar documentation](../docs/content/docs/connectors/table/pulsar.md)
+- [flink-example: cookbooks, examples](https://github.com/streamnative/flink-example)
+- [StreamNative Hub documentation](https://hub.streamnative.io/data-processing/pulsar-flink/1.15.0.1)
+
+> The legacy [pulsar-flink connector](https://github.com/streamnative/pulsar-flink) is deprecated. It will not be actively maintained and users are recommended to use this repository.

.github/workflows/daily-end-to-end-test.yml

@@ -0,0 +1,29 @@
+name: Daily End to End Test
+
+on:
+  schedule:
+    - cron: 0 1 * * *
+  workflow_dispatch:
+jobs:
+  sync:
+    name: Run End to End test for pulsar
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        branch:
+          - develop
+          - release-sn-1.14
+          - release-sn-1.15
+    steps:
+      - name: Checkout the code
+        uses: actions/checkout@v2
+        with:
+          ref: ${{ matrix.branch }}
+      - name: Set up Maven
+        uses: apache/pulsar-test-infra/setup-maven@master
+        with:
+          maven-version: 3.6.2
+      - name: Run e2e Test
+        run: |
+          mvn clean install -DskipTests
+          mvn -pl 'flink-end-to-end-tests/flink-end-to-end-tests-pulsar' test

.github/workflows/docs.sh

@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+################################################################################
+#  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.
+################################################################################
+set -e
+
+mvn --version
+java -version
+javadoc -J-version
+
+# setup hugo
+HUGO_REPO=https://github.com/gohugoio/hugo/releases/download/v0.80.0/hugo_extended_0.80.0_Linux-64bit.tar.gz
+HUGO_ARTIFACT=hugo_extended_0.80.0_Linux-64bit.tar.gz
+if ! curl --fail -OL $HUGO_REPO ; then
+	echo "Failed to download Hugo binary"
+	exit 1
+fi
+tar -zxvf $HUGO_ARTIFACT
+git submodule update --init --recursive
+# generate docs into docs/target
+./hugo -v --source docs --destination target
+if [ $? -ne 0 ]; then
+	echo "Error building the docs"
+	exit 1
+fi
+
+# build Flink; required for Javadoc step
+mvn clean install -B -DskipTests -Dfast -Pskip-webui-build
+
+# build java/scala docs
+mkdir -p docs/target/api
+mvn javadoc:aggregate -B \
+    -Paggregate-scaladoc \
+    -DadditionalJOption="-Xdoclint:none --allow-script-in-comments" \
+    -Dmaven.javadoc.failOnError=false \
+    -Dcheckstyle.skip=true \
+    -Dspotless.check.skip=true \
+    -Denforcer.skip=true \
+    -Dheader="<a href=\"http://flink.apache.org/\" target=\"_top\"><h1>Back to Flink Website</h1></a> <script>var _paq=window._paq=window._paq||[];_paq.push([\"disableCookies\"]),_paq.push([\"setDomains\",[\"*.flink.apache.org\",\"*.nightlies.apache.org/flink\"]]),_paq.push([\"trackPageView\"]),_paq.push([\"enableLinkTracking\"]),function(){var u=\"//matomo.privacy.apache.org/\";_paq.push([\"setTrackerUrl\",u+\"matomo.php\"]),_paq.push([\"setSiteId\",\"1\"]);var d=document, g=d.createElement('script'), s=d.getElementsByTagName('script')[0];g.async=true; g.src=u+'matomo.js'; s.parentNode.insertBefore(g,s)}();</script>"
+mv target/site/apidocs docs/target/api/java
+pushd flink-scala
+mvn scala:doc -B
+mv target/site/scaladocs ../docs/target/api/scala
+popd
+
+# build python docs
+if [ -f  ./flink-python/dev/lint-python.sh ]; then
+    # Just completely ignore sudo in conda.
+    unset SUDO_UID SUDO_GID SUDO_USER
+
+    # build python docs
+    # disable the gateway, because otherwise it tries to find FLINK_HOME to access Java classes
+    PYFLINK_GATEWAY_DISABLED=1 ./flink-python/dev/lint-python.sh -i "sphinx"
+
+    # move python docs
+    mv flink-python/docs/_build/html docs/target/api/python
+fi

.github/workflows/run-pr-check.yml

@@ -0,0 +1,23 @@
+name: Run Test for Pull Request
+
+on:
+  pull_request:
+    branches: [ develop, release-sn-** ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout the code
+        uses: actions/checkout@v2
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+      - name: Set up JDK 8
+        uses: actions/setup-java@v2
+        with:
+          java-version: '8'
+          cache: 'maven'
+          distribution: 'zulu'
+      - name: Run Unit Test and Install
+        run: |
+          mvn -ntp -pl 'flink-connectors/flink-connector-pulsar,flink-connectors/flink-sql-connector-pulsar' clean install

docs/config.toml

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-baseURL = '//nightlies.apache.org/flink/flink-docs-master'
+baseURL = '//nightlies.apache.org/flink/flink-docs-release-1.15'
 languageCode = "en-us"
 title = "Apache Flink"
 enableGitInfo = false
@@ -24,7 +24,7 @@ pygmentsUseClasses = true
 [params]
   # Flag whether this is a stable version or not.
   # Used for the quickstart page.
-  IsStable = false
+  IsStable = true
 
   # Flag to indicate whether an outdated warning should be shown.
   ShowOutDatedWarning = false
@@ -34,14 +34,14 @@ pygmentsUseClasses = true
   # we change the version for the complete docs when forking of a release branch
   # etc.
   # The full version string as referenced in Maven (e.g. 1.2.1)
-  Version = "1.15-SNAPSHOT"
+  Version = "1.15.0"
 
   # For stable releases, leave the bugfix version out (e.g. 1.2). For snapshot
   # release this should be the same as the regular version
-  VersionTitle = "1.15-SNAPSHOT"
+  VersionTitle = "1.15"
 
   # The branch for this version of Apache Flink
-  Branch = "master"
+  Branch = "release-1.15"
 
   # The github repository for Apache Flink
   Repo = "//github.com/apache/flink"
@@ -60,22 +60,23 @@ pygmentsUseClasses = true
 
   ZhDownloadPage = "//flink.apache.org/zh/downloads.html"
 
-  JavaDocs = "//nightlies.apache.org/flink/flink-docs-master/api/java/"
+  JavaDocs = "//nightlies.apache.org/flink/flink-docs-release-1.15/api/java/"
 
-  ScalaDocs = "//nightlies.apache.org/flink/flink-docs-master/api/scala/index.html#org.apache.flink.api.scala.package"
+  ScalaDocs = "//nightlies.apache.org/flink/flink-docs-release-1.15/api/scala/index.html#org.apache.flink.api.scala.package"
 
-  PyDocs = "//nightlies.apache.org/flink/flink-docs-master/api/python/"
+  PyDocs = "//nightlies.apache.org/flink/flink-docs-release-1.15/api/python/"
 
   # External links at the bottom
   # of the menu
   MenuLinks = [
     ["Project Homepage", "//flink.apache.org"],
-    ["JavaDocs", "//nightlies.apache.org/flink/flink-docs-master/api/java/"],
-    ["ScalaDocs", "//nightlies.apache.org/flink/flink-docs-master/api/scala/index.html#org.apache.flink.api.scala.package"],
-    ["PyDocs", "//nightlies.apache.org/flink/flink-docs-master/api/python/"]
+    ["JavaDocs", "//nightlies.apache.org/flink/flink-docs-release-1.15/api/java/"],
+    ["ScalaDocs", "//nightlies.apache.org/flink/flink-docs-release-1.15/api/scala/index.html#org.apache.flink.api.scala.package/"],
+    ["PyDocs", "//nightlies.apache.org/flink/flink-docs-release-1.15/api/python/"]
   ]
 
   PreviousDocs = [
+    ["1.15", "http://nightlies.apache.org/flink/flink-docs-release-1.15"],
     ["1.14", "http://nightlies.apache.org/flink/flink-docs-release-1.14"],
     ["1.13", "http://nightlies.apache.org/flink/flink-docs-release-1.13"],
     ["1.12", "http://nightlies.apache.org/flink/flink-docs-release-1.12"],

docs/content.zh/docs/concepts/stateful-stream-processing.md

@@ -310,7 +310,10 @@ mechanism for this.
 
 Savepoints are similar to checkpoints except that they are
 **triggered by the user** and **don't automatically expire** when newer
-checkpoints are completed.
+checkpoints are completed. 
+To make proper use of savepoints, it's important to understand the differences between 
+[checkpoints]({{< ref "docs/ops/state/checkpoints" >}}) and [savepoints]({{< ref "docs/ops/state/savepoints" >}})
+which is described in [checkpoints vs. savepoints]({{< ref "docs/ops/state/checkpoints_vs_savepoints" >}}).
 
 {{< top >}}
 

docs/content.zh/docs/concepts/time.md

@@ -84,8 +84,8 @@ one can refer to different notions of *time*:
   out-of-order or late events, or when reprocessing historic data. For example,
   an hourly event time window will contain all records that carry an event
   timestamp that falls into that hour, regardless of the order in which they
-  arrive, or when they are processed. (See the section on [late
-  events](#late-elements) for more information.)
+  arrive, or when they are processed. (See the section on [lateness](#lateness)
+  for more information.)
 
   Note that sometimes when event time programs are processing live data in
   real-time, they will use some *processing time* operations in order to

docs/content.zh/docs/connectors/datastream/cassandra.md

@@ -37,7 +37,7 @@ To use this connector, add the following dependency to your project:
 
 {{< artifact flink-connector-cassandra withScalaVersion >}}
 
-Note that the streaming connectors are currently __NOT__ part of the binary distribution. See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+Note that the streaming connectors are currently __NOT__ part of the binary distribution. See how to link with them for cluster execution [here]({{< ref "docs/dev/configuration/overview" >}}).
 
 ## Installing Apache Cassandra
 There are multiple ways to bring up a Cassandra instance on local machine:

docs/content.zh/docs/connectors/datastream/elasticsearch.md

@@ -52,7 +52,7 @@ under the License.
 </table>
 
 请注意，流连接器目前不是二进制发行版的一部分。
-有关如何将程序和用于集群执行的库一起打包，参考[此文档]({{< ref "docs/dev/datastream/project-configuration" >}})
+有关如何将程序和用于集群执行的库一起打包，参考[此文档]({{< ref "docs/dev/configuration/overview" >}})。
 
 ## 安装 Elasticsearch
 
@@ -132,7 +132,6 @@ private static IndexRequest createIndexRequest(String element) {
 
     return Requests.indexRequest()
         .index("my-index")
-        .type("my-type")
         .id(element)
         .source(json);
 }
@@ -165,7 +164,7 @@ def createIndexRequest(element: (String)): IndexRequest = {
     "data" -> element.asInstanceOf[AnyRef]
   )
 
-  Requests.indexRequest.index("my-index").`type`("my-type").source(mapAsJavaMap(json))
+  Requests.indexRequest.index("my-index").source(mapAsJavaMap(json))
 }
 ```
 
@@ -210,16 +209,13 @@ def createIndexRequest(element: (String)): IndexRequest = {
 
 ### Elasticsearch Sinks 和容错
 
-默认情况下，Flink Elasticsearch Sink 不会提供任何传递健壮性的保障。
-用户可以选择启用 Elasticsearch sink 的 at-least-once 语义。
-
-通过启用 Flink checkpoint，Flink Elasticsearch Sink 可以保证至少一次将操作请求发送到 Elasticsearch 集群。
+通过启用 Flink checkpoint，Flink Elasticsearch Sink 保证至少一次将操作请求发送到 Elasticsearch 集群。
 这是通过在进行 checkpoint 时等待 `BulkProcessor` 中所有挂起的操作请求来实现。
 这有效地保证了在触发 checkpoint 之前所有的请求被 Elasticsearch 成功确认，然后继续处理发送到 sink 的记录。
 
 关于 checkpoint 和容错的更多详细信息，请参见[容错文档]({{< ref "docs/learn-flink/fault_tolerance" >}})。
 
-要使用具有容错特性的 Elasticsearch Sinks，需要配置启用 at-least-once 分发并且在执行环境中启用作业拓扑的 checkpoint：
+要使用具有容错特性的 Elasticsearch Sinks，需要在执行环境中启用作业拓扑的 checkpoint：
 
 {{< tabs "d00d1e93-4844-40d7-b0ec-9ec37e73145e" >}}
 {{< tab "Java" >}}
@@ -276,6 +272,10 @@ val sinkBuilder = new Elasticsearch7SinkBuilder[String]
 {{< /tab >}}
 {{< /tabs >}}
 
+<p style="border-radius: 5px; padding: 5px" class="bg-info">
+Using UpdateRequests with deterministic ids and the upsert method it is possible to achieve exactly-once semantics in Elasticsearch when AT_LEAST_ONCE delivery is configured for the connector.
+</p>
+
 ### 处理失败的 Elasticsearch 请求
 
 Elasticsearch 操作请求可能由于多种原因而失败，包括节点队列容量暂时已满或者要被索引的文档格式错误。
@@ -373,7 +373,7 @@ checkpoint 会进行等待，直到 Elasticsearch 节点队列有足够的容量
 ## 将 Elasticsearch 连接器打包到 Uber-Jar 中
 
 建议构建一个包含所有依赖的 uber-jar (可执行的 jar)，以便更好地执行你的 Flink 程序。
-(更多信息参见[此文档]({{< ref "docs/dev/datastream/project-configuration" >}}))。
+(更多信息参见[此文档]({{< ref "docs/dev/configuration/overview" >}}))。
 
 或者，你可以将连接器的 jar 文件放入 Flink 的 `lib/` 目录下，使其在全局范围内可用，即可用于所有的作业。
 

docs/content.zh/docs/connectors/datastream/filesystem.md

@@ -68,10 +68,10 @@ under the License.
 {{< tab "Java" >}}
 ```java
 // 从文件流中读取文件内容
-FileSource.forRecordStreamFormat(StreamFormat,Path...)
+FileSource.forRecordStreamFormat(StreamFormat,Path...);
 
 // 从文件中一次读取一批记录
-FileSource.forBulkFileFormat(BulkFormat,Path...)
+FileSource.forBulkFileFormat(BulkFormat,Path...);
 ```
 {{< /tab >}}
 {{< /tabs >}}

docs/content.zh/docs/connectors/datastream/formats/avro.md

@@ -52,7 +52,7 @@ DataStream<User> usersDS = env.createInput(users);
 注意，`User` 是一个通过 Avro schema生成的 POJO 类。Flink 还允许选择 POJO 中字符串类型的键。例如：
 
 ```java
-usersDS.keyBy("name")
+usersDS.keyBy("name");
 ```
 
 

docs/content.zh/docs/connectors/datastream/formats/hadoop.md

@@ -106,7 +106,7 @@ Flink 为 Hadoop `OutputFormats` 提供了一个兼容性包装器。支持任
 
 ```java
 // 获取我们希望发送的结果
-DataStream<Tuple2<Text, IntWritable>> hadoopResult = [...]
+DataStream<Tuple2<Text, IntWritable>> hadoopResult = [...];
 
 // 设置 the Hadoop TextOutputFormat。
 HadoopOutputFormat<Text, IntWritable> hadoopOF =