[IOTDB-4829] Let NoMoreTsBlockEvent RPC is called in async way (apache#7911)

Author: lancelly

docs/UserGuide/Reference/Syntax-Conventions.md

@@ -20,11 +20,11 @@
 -->
 
 
-# Literal Values
+## Literal Values
 
 This section describes how to write literal values in IoTDB. These include strings, numbers, timestamp values, boolean values, and NULL.
 
-## String Literals
+### String Literals
 
 > We refer to MySQL's definition of string：A string is a sequence of bytes or characters, enclosed within either single quote (`'`) or double quote (`"`) characters.
 
@@ -111,7 +111,7 @@ Usages of string literals:
 - The key/value of an attribute can be String Literal and identifier, more details can be found at **key-value pair** part. 
 
 
-### How to use quotation marks in String Literals
+#### How to use quotation marks in String Literals
 
 There are several ways to include quote characters within a string:
 
@@ -132,7 +132,7 @@ The following examples demonstrate how quoting and escaping work:
 """string"  // "string
 ```
 
-## Numeric Literals
+### Numeric Literals
 
 Number literals include integer (exact-value) literals and floating-point (approximate-value) literals.
 
@@ -146,27 +146,27 @@ The `FLOAT` and `DOUBLE` data types are floating-point types and calculations ar
 
 An integer may be used in floating-point context; it is interpreted as the equivalent floating-point number.
 
-## Timestamp Literals
+### Timestamp Literals
 
 The timestamp is the time point at which data is produced. It includes absolute timestamps and relative timestamps in IoTDB. For information about timestamp support in IoTDB, see [Data Type Doc](../Data-Concept/Data-Type.md).
 
 Specially, `NOW()` represents a constant timestamp that indicates the system time at which the statement began to execute.
 
-## Boolean Literals
+### Boolean Literals
 
 The constants `TRUE` and `FALSE` evaluate to 1 and 0, respectively. The constant names can be written in any lettercase.
 
-## NULL Values
+### NULL Values
 
 The `NULL` value means “no data.” `NULL` can be written in any lettercase.
 
-# Identifiers
+## Identifiers
 
-## Usage scenarios
+### Usage scenarios
 
 Certain objects within IoTDB, including `TRIGGER`, `FUNCTION`(UDF), `CONTINUOUS QUERY`, `SCHEMA TEMPLATE`, `USER`, `ROLE`,`Pipe`,`PipeSink`,`alias` and other object names are known as identifiers.
 
-## Constraints
+### Constraints
 
 Below are basic constraints of identifiers, specific identifiers may have other constraints, for example, `user` should consists of more than 4 characters. 
 
@@ -182,7 +182,7 @@ Below are basic constraints of identifiers, specific identifiers may have other
 - Identifier contains special characters.
 - Identifier that is a real number.
 
-## How to use quotations marks in quoted identifiers
+### How to use quotations marks in quoted identifiers
 
 `'` and `"` can be used directly in quoted identifiers.
 
@@ -198,7 +198,7 @@ create schema template `t1``t`
 (temperature FLOAT encoding=RLE, status BOOLEAN encoding=PLAIN compression=SNAPPY)
 ```
 
-## Examples
+### Examples
 
 Examples of case in which quoted identifier is used ：
 
@@ -279,11 +279,11 @@ Examples of case in which quoted identifier is used ：
 - The key/value of an attribute can be String Literal and identifier, more details can be found at **key-value pair** part. 
 
 
-# Node Names in Path
+## Node Names in Path
 
 Node name is a special identifier, it can also be wildcard `*` and `**`. When creating timeseries, node name can not be wildcard. In query statment, you can use wildcard to match one or more nodes of path.
 
-## Wildcard
+### Wildcard
 
 `*` represents one node. For example, `root.vehicle.*.sensor1` represents a 4-node path which is prefixed with `root.vehicle` and suffixed with `sensor1`.
 
@@ -315,7 +315,7 @@ select a*b from root.sg
 |Time|root.sg.a * root.sg.b|
 ```
 
-## Identifier
+### Identifier
 
 When node name is not wildcard, it is a identifier, which means the constraints on it is the same as described in Identifier part.
 
@@ -378,7 +378,7 @@ Results:
 +-----------------------------+-----------+
 ```
 
-# Key-Value Pair
+## Key-Value Pair
 
 **The key/value of an attribute can be constant(including string) and identifier. **
 
@@ -477,13 +477,13 @@ CREATE PIPE my_pipe TO my_iotdb FROM
 (select ** from root WHERE time>=yyyy-mm-dd HH:MM:SS) WITH 'SyncDelOp' = 'true'
 ```
 
-# Keywords and Reserved Words
+## Keywords and Reserved Words
 
 Keywords are words that have significance in SQL. Keywords can be used as an identifier. Certain keywords, such as TIME/TIMESTAMP and ROOT, are reserved and cannot use as identifiers.
 
 [Keywords and Reserved Words](Keywords.md) shows the keywords and reserved words in IoTDB.
 
-# Session、TsFile API
+## Session、TsFile API
 
 When using the Session and TsFile APIs, if the method you call requires parameters such as measurement, device, storage group, path in the form of String, **please ensure that the parameters passed in the input string is the same as when using the SQL statement**, here are some examples to help you understand. Code example could be found at: `example/session/src/main/java/org/apache/iotdb/SyntaxConventionRelatedExample.java`
 
@@ -582,7 +582,7 @@ String[] paths = new String[]{"root.sg.a", "root.sg.`a.``\"b`", "root.sg.`111`"}
 List<String> pathList = Arrays.asList(paths);
 ```
 
-# Detailed Definitions of Lexical and Grammar
+## Detailed Definitions of Lexical and Grammar
 
 Please read the lexical and grammar description files in our code repository:
 

docs/zh/UserGuide/Reference/Syntax-Conventions.md

@@ -20,11 +20,11 @@
 -->
 
 
-# 字面值常量
+## 字面值常量
 
 该部分对 IoTDB 中支持的字面值常量进行说明，包括字符串常量、数值型常量、时间戳常量、布尔型常量和空值。
 
-## 字符串常量
+### 字符串常量
 
 > 我们参照了 MySQL 对 字符串的定义：A string is a sequence of bytes or characters, enclosed within either single quote (`'`) or double quote (`"`) characters.
 
@@ -37,7 +37,7 @@ MySQL 对字符串的定义可以参考：[MySQL :: MySQL 8.0 Reference Manual :
 "another string"
 ```
 
-### 使用场景
+#### 使用场景
 
 - `INSERT` 或者 `SELECT` 中用于表达 `TEXT` 类型数据的场景。
 
@@ -108,7 +108,7 @@ MySQL 对字符串的定义可以参考：[MySQL :: MySQL 8.0 Reference Manual :
 
 - 用于表示键值对，键值对的键和值可以被定义成常量（包括字符串）或者标识符，具体请参考键值对章节。
 
-### 如何在字符串内使用引号
+#### 如何在字符串内使用引号
 
 - 在单引号引起的字符串内，双引号无需特殊处理。同理，在双引号引起的字符串内，单引号无需特殊处理。
 - 在单引号引起的字符串里，可以通过双写单引号来表示一个单引号，即单引号 ' 可以表示为 ''。
@@ -128,7 +128,7 @@ MySQL 对字符串的定义可以参考：[MySQL :: MySQL 8.0 Reference Manual :
 """string"  // "string
 ```
 
-## 数值型常量
+### 数值型常量
 
 数值型常量包括整型和浮点型。
 
@@ -140,27 +140,27 @@ MySQL 对字符串的定义可以参考：[MySQL :: MySQL 8.0 Reference Manual :
 
 在浮点上下文中可以使用整数，它会被解释为等效的浮点数。
 
-## 时间戳常量
+### 时间戳常量
 
 时间戳是一个数据到来的时间点，在 IoTDB 中分为绝对时间戳和相对时间戳。详细信息可参考 [数据类型文档](https://iotdb.apache.org/zh/UserGuide/Master/Data-Concept/Data-Type.html)。
 
 特别地，`NOW()`表示语句开始执行时的服务端系统时间戳。
 
-## 布尔型常量
+### 布尔型常量
 
 布尔值常量 `TRUE` 和 `FALSE` 分别等价于 `1` 和 `0`，它们对大小写不敏感。
 
-## 空值
+### 空值
 
 `NULL`值表示没有数据。`NULL`对大小写不敏感。
 
-# 标识符
+## 标识符
 
-## 使用场景
+### 使用场景
 
 在 IoTDB 中，触发器名称、UDF函数名、元数据模板名称、用户与角色名、连续查询标识、Pipe、PipeSink、键值对中的键和值、别名等可以作为标识符。
 
-## 约束
+### 约束
 
 请注意，此处约束是标识符的通用约束，具体标识符可能还附带其它约束条件，如用户名限制字符数大于等于4，更严格的约束请参考具体标识符相关的说明文档。
 
@@ -181,7 +181,7 @@ MySQL 对字符串的定义可以参考：[MySQL :: MySQL 8.0 Reference Manual :
 - 标识符包含不允许的特殊字符。
 - 标识符为实数。
 
-## 如何在反引号引起的标识符中使用引号
+### 如何在反引号引起的标识符中使用引号
 
 **在反引号引起的标识符中可以直接使用单引号和双引号。**
 
@@ -197,7 +197,7 @@ create schema template `t1't"t`
 (temperature FLOAT encoding=RLE, status BOOLEAN encoding=PLAIN compression=SNAPPY)
 ```
 
-## 特殊情况示例
+### 特殊情况示例
 
 需要使用反引号进行引用的部分情况示例：
 
@@ -277,11 +277,11 @@ create schema template `t1't"t`
 - 用于表示键值对，键值对的键和值可以被定义成常量（包括字符串）或者标识符，具体请参考键值对章节。
 
 
-# 路径结点名
+## 路径结点名
 
 路径结点名是特殊的标识符，其还可以是通配符 \* 或 \*\*。在创建时间序列时，各层级的路径结点名不能为通配符 \* 或 \*\*。在查询语句中，可以用通配符 \* 或 \*\* 来表示路径结点名，以匹配一层或多层路径。
 
-## 通配符
+### 通配符
 
 `*`在路径中表示一层。例如`root.vehicle.*.sensor1`代表的是以`root.vehicle`为前缀，以`sensor1`为后缀，层次等于 4 层的路径。
 
@@ -312,7 +312,7 @@ select a*b from root.sg
 |Time|root.sg.a * root.sg.b|
 ```
 
-## 标识符
+### 标识符
 
 路径结点名不为通配符时，使用方法和标识符一致。**在 SQL 中需要使用反引号引用的路径结点，在结果集中也会用反引号引起。**
 
@@ -377,7 +377,7 @@ select `111` from root.sg
 +-----------------------------+-------------+
 ```
 
-# 键值对
+## 键值对
 
 **键值对的键和值可以被定义为标识符或者常量。**
 
@@ -476,13 +476,13 @@ CREATE PIPE my_pipe TO my_iotdb FROM
 (select ** from root WHERE time>=yyyy-mm-dd HH:MM:SS) WITH 'SyncDelOp' = 'true'
 ```
 
-# 关键字和保留字
+## 关键字和保留字
 
 关键字是在 SQL 具有特定含义的词，可以作为标识符。保留字是关键字的一个子集，保留字不能用于标识符。
 
 关于 IoTDB 的关键字和保留字列表，可以查看 [关键字和保留字](https://iotdb.apache.org/zh/UserGuide/Master/Reference/Keywords.html) 。
 
-# Session、TsFile API
+## Session、TsFile API
 
 在使用Session、TsFIle API时，如果您调用的方法需要以字符串形式传入物理量（measurement）、设备（device）、存储组（storage group）、路径（path）等参数，**请保证所传入字符串与使用 SQL 语句时的写法一致**，下面是一些帮助您理解的例子。具体代码示例可以参考：`example/session/src/main/java/org/apache/iotdb/SyntaxConventionRelatedExample.java`
 
@@ -581,7 +581,7 @@ String[] paths = new String[]{"root.sg.a", "root.sg.`a.``\"b`", "root.sg.`111`"}
 List<String> pathList = Arrays.asList(paths);
 ```
 
-# 词法与文法详细定义
+## 词法与文法详细定义
 
 请阅读代码仓库中的词法和语法描述文件：
 

server/src/main/java/org/apache/iotdb/db/mpp/execution/exchange/SinkHandle.java

@@ -170,48 +170,13 @@ public synchronized void send(int partition, List<TsBlock> tsBlocks) {
     throw new UnsupportedOperationException();
   }
 
-  private void sendEndOfDataBlockEvent() throws Exception {
-    logger.debug("[NotifyNoMoreTsBlock]");
-    int attempt = 0;
-    TEndOfDataBlockEvent endOfDataBlockEvent =
-        new TEndOfDataBlockEvent(
-            remoteFragmentInstanceId,
-            remotePlanNodeId,
-            localFragmentInstanceId,
-            nextSequenceId - 1);
-    while (attempt < MAX_ATTEMPT_TIMES) {
-      attempt += 1;
-      try (SyncDataNodeMPPDataExchangeServiceClient client =
-          mppDataExchangeServiceClientManager.borrowClient(remoteEndpoint)) {
-        client.onEndOfDataBlockEvent(endOfDataBlockEvent);
-        break;
-      } catch (Throwable e) {
-        logger.error("Failed to send end of data block event, attempt times: {}", attempt, e);
-        if (attempt == MAX_ATTEMPT_TIMES) {
-          throw e;
-        }
-        Thread.sleep(retryIntervalInMs);
-      }
-    }
-  }
-
   @Override
   public synchronized void setNoMoreTsBlocks() {
     logger.debug("[StartSetNoMoreTsBlocks]");
     if (aborted || closed) {
       return;
     }
-    try {
-      sendEndOfDataBlockEvent();
-    } catch (Exception e) {
-      throw new RuntimeException("Send EndOfDataBlockEvent failed", e);
-    }
-    noMoreTsBlocks = true;
-
-    if (isFinished()) {
-      sinkHandleListener.onFinish(this);
-    }
-    sinkHandleListener.onEndOfBlocks(this);
+    executorService.submit(new SendEndOfDataBlockEventTask());
   }
 
   @Override
@@ -413,4 +378,51 @@ public void run() {
       }
     }
   }
+
+  /**
+   * Send a {@link org.apache.iotdb.mpp.rpc.thrift.TEndOfDataBlockEvent} to downstream fragment
+   * instance.
+   */
+  class SendEndOfDataBlockEventTask implements Runnable {
+
+    @Override
+    public void run() {
+      try (SetThreadName sinkHandleName = new SetThreadName(threadName)) {
+        logger.debug("[NotifyNoMoreTsBlock]");
+        int attempt = 0;
+        TEndOfDataBlockEvent endOfDataBlockEvent =
+            new TEndOfDataBlockEvent(
+                remoteFragmentInstanceId,
+                remotePlanNodeId,
+                localFragmentInstanceId,
+                nextSequenceId - 1);
+        while (attempt < MAX_ATTEMPT_TIMES) {
+          attempt += 1;
+          try (SyncDataNodeMPPDataExchangeServiceClient client =
+              mppDataExchangeServiceClientManager.borrowClient(remoteEndpoint)) {
+            client.onEndOfDataBlockEvent(endOfDataBlockEvent);
+            break;
+          } catch (Throwable e) {
+            logger.error("Failed to send end of data block event, attempt times: {}", attempt, e);
+            if (attempt == MAX_ATTEMPT_TIMES) {
+              logger.error("Failed to send end of data block event after all retry", e);
+              sinkHandleListener.onFailure(SinkHandle.this, e);
+              return;
+            }
+            try {
+              Thread.sleep(retryIntervalInMs);
+            } catch (InterruptedException ex) {
+              Thread.currentThread().interrupt();
+              sinkHandleListener.onFailure(SinkHandle.this, e);
+            }
+          }
+        }
+        noMoreTsBlocks = true;
+        if (isFinished()) {
+          sinkHandleListener.onFinish(SinkHandle.this);
+        }
+        sinkHandleListener.onEndOfBlocks(SinkHandle.this);
+      }
+    }
+  }
 }