Flink Clients

Flink SQL知其然知其所以然: Flink SQL + calcite

一条 flink sql 的执行过程

先了解整个流程，有了全局视角之后，后续会详述细节。

先来看看 flink datastream 任务的执行过程:

• DataStream：
  使用时要在 flink datastream api 提供的各种 udf（比如 flatmap，keyedProcessFunction 等）中自定义处理逻辑，具体的业务执行逻辑都是敲代码、 java 文件写的，然后编译在 jvm 中执行，就和一个普通的 main 函数应用一模一样的流程。因为代码执行逻辑都是自己写的，所以这一部分相对好理解。

• SQL：
  Java 编译器不能识别和编译一条 SQL 进行执行，那么一条 SQL 是咋执行的呢？

2.1.先发挥自己的想象力

我们逆向思维进行考虑，如果想让一条 Flink SQL 按照我们的预期在 jvm 中执行，需要哪些过程。

1. 整体来说：参考 datastream，如果 jvm 能执行 datastream java code 编译后的 class 文件，那么加一个 sql 解析层，能将 sql 逻辑解析为 datastream 的各种算子，然后编译执行不就 vans 了。
2. sql parser：首先得有一个 sql parser 吧，得先能识别 sql 语法，将 sql 语法转化为 AST、具体的关系代数。
3. 关系代数到 datastream 算子的映射：sql 逻辑解析为 datastream，需要有一个解析的映射逻辑吧。sql 是基于关系代数的，可以维护一个 sql 中的每个关系代数到具体 datastream 接口的映射关系，有了这些映射关系我们就可以将 sql 映射成一段可执行的 datastream 代码。举个例子：其可以将：

• sql select xxx 解析为类似 datastream 中的 map
• where xxx 解析为 filter
• group by 解析成 keyby
• sum（xx），count（xxx）可以解析为 datastream 中的 aggregate function
• etc…
• 代码生成：有了 sql AST，sql 到 datasretam 算子的映射关系之后，就要进行具体的代码生成了。比如去解析 sql AST 中具体哪些字段用作 where 逻辑，哪些字段用作 group by，都需要生成对应具体的 datastream 代码。
• 运行：经过上述流程之后，就可以将一个 sql 翻译成一个 datastream 作业了，happy 的执行。

如下图所示，描绘了上述逻辑：

12

那么这个和 flink 实际实现有啥异同呢？

flink 大致是这样做的，虽在 flink 本身的中间还有一些其他的流程，后来的版本也不是基于 datastream，但是整体的处理逻辑还是和上述一致的。

所以不了解整体流程的同学可以先按照上述流程进行理解。

按照 博主的脑洞 来总结一条 sql 的使命就是：sql -> AST -> codegen(java code) -> 让我们 run 起来好吗

2.2.看看 flink 的实现

26

上面手绘可能看不清，下面这张图更清楚。

28

标准的一条 flink sql 运行起来的流程如下：

Notes：刚开始对其中的 SqlNode，RelNode 概念可能比较模糊。先理解整个流程，后续会详细介绍这些概念。

1. sql 解析阶段：calcite parser 解析（sql -> AST，AST 即 SqlNode Tree）

2. SqlNode 验证阶段：calcite validator 校验（SqlNode -> SqlNode，语法、表达式、表信息）

3. 语义分析阶段：SqlNode 转换为 RelNode，RelNode 即 Logical Plan（SqlNode -> RelNode）

4. 优化阶段：calcite optimizer 优化（RelNode -> RelNode，剪枝、谓词下推等）

5. 物理计划生成阶段：Logical Plan 转换为 Physical Plan（等同于 RelNode 转换成 DataSet\DataStream API）

6. 后续的运行逻辑与 datastream 一致

可以发现 flink 的实现 比 博主的脑洞 整体主要框架上面是一致的。多出来的部分主要是 SqlNode 验证阶段，优化阶段。

3.简介篇-calcite 在 flink sql 中的角色

大致了解了 一条 flink sql 的运行流程 之后，我们来看看 calcite 这玩意到底在 flink 里干了些啥。

根据上文总结来说 calcite 在 flink sql 中担当了 sql 解析、验证、优化功能。

30

看着 calcite 干了这么多事，那 calcite 是个啥东东，它的定位是啥？

3.1.calcite 是啥？

calcite 是一个动态数据的管理框架，它可以用来构建数据库系统的不同的解析的模块，但是它不包含数据存储数据处理等功能。

calcite 的目标是一种方案，适应所有的需求场景，希望能为不同计算平台和数据源提供统一的 sql 解析引擎，但是它只是提供查询引擎，而没有真正的去存储这些数据。

61

下图是目前使用了 calcite 能力的其他组件，也可见官网 https://calcite.apache.org/docs/powered_by.html 。

4

简单来说的话，可以先理解为 calcite 具有这几个功能（当然还有其他很牛逼的功能，感兴趣可以自查官网）。

1. 自定义 sql 解析器：比如说我们新发明了一个引擎，然后我们要在这个引擎上来创造一套基于 sql 的接口，那么我们就可以使用直接 calcite，不用自己去写一套专门的 sql 的解析器，以及执行以及优化引擎，calcite 人都有。
2. sql parser（extends SqlAbstractParserImpl）：将 sql 的各种关系代数解析为具体的 AST，这些 AST 都能对应到具体的 java model，在 java 的世界里面，对象很重要，有了这些对象（SqlSelect、SqlNode），就可以根据这些对象做具体逻辑处理了。举个例子，如下图，一条简单的 select c,d from source where a = '6' sql，经过 calcite 的解析之后，就可以得到 AST model（SqlNode）。可以看到有 SqlSelect、SqlIdentifier、SqlIdentifier、SqlCharStringLiteral。
3. sql validator（extends SqlValidatorImpl）：根据语法、表达式、表信息进行 SqlNode 正确性校验。
4. sql optimizer：剪枝、谓词下推等优化

上面的这些能力整体组成如下图所示：

29

实际使用 calcite 解析一条 sql，跑起来看看。

2

3.2.flink sql 为什么选择 calcite？

1. 不用重复造轮子。有限的精力应该放在有价值的事情上。
2. calcite 有针对 stream 表的解决方案。具体可见 https://calcite.apache.org/docs/stream.html。

4.案例篇-calcite 的能力、案例

4.1.先用用 calcite

1

重中之重，在了解原理之前，先跑起来是王道，也会帮助我们逐步理解。

官网已经有一个 csv 的案例了。感兴趣的可以直达 https://calcite.apache.org/docs/tutorial.html 。

跑完一个 csv demo，在详细了解 calcite 之前还需要了解下 sql，calcite 的支柱：关系代数。

4.2.关系代数

sql 是基于关系代数的查询语言，是关系代数在工程上的一种很好的实现方案。在工程中，关系代数难表达，但是 sql 就易于理解。关系代数和 sql 的关系如下。

1. 可以将一条 sql 解析为一个关系代数表达式的组合。在 sql 中的操作都可以转化成关系代数的表达式。

2. sql 的执行优化（所有的优化的前提都是优化前和优化后最终执行结果相同，即等价交换）是基于关系代数运算的。

4.2.1.常用关系代数

总结下，有哪些常用的关系代数：

50

4.2.2.sql 优化支柱之关系代数等价变换

关系代数等价变换是 calcite optimizer 的基础理论。

下面是一些等价变换的例子。

1.连接（⋈），笛卡尔积（×）的交换律

51

2.连接（⋈），笛卡尔积（×）的结合律

3.投影（Π）的串接定律

4.选择（σ）的串接定律

5.选择（σ）与投影（Π）的交换

6.选择（σ）与笛卡尔积（×）的交换

7.选择（σ）与并（∪）的交换

8.选择（σ）与差（-）的交换

9.投影（Π）与笛卡尔积（×）的交换

10.投影（Π）与并（∪）的交换

然后看一个基于关系代数优化的实际 sql 案例：

有三个关系 A（a1,a2,a3,…）、B（b1,b2,b3, … ）、C（a1,b1,c1,c2, … ）

有一个查询请求如下：

1

SELECT A.a1 FROM A，B，C WHERE A.a1 = C.a1 AND B.b1 = C.b1 AND f(c1)

1.首先将 sql 转为关系代数的语法树。

36

2.优化：选择（σ）的串接定律。

47

37

3.优化：选择（σ）与笛卡尔积（×）的交换。

48

38

4.优化：投影（π）与笛卡尔积（×）的交换。

49

关于关系代数我们就有了大致的了解。

除此之外，对于更深入了解 flink sql，calcite 而言，我们还需要了解一下在 calcite 代码体系中有哪些重要 model。

4.3.calcite 必知的基础 model

calcite 中有两个最最基础、重要的 model 在我们理解 flink sql 解析流程时需要知道的。

1. SqlNode：sql 转化而成，可以理解为直观表达 sql 层次结构的的 model
2. RelNode：SqlNode 转化而成，可以理解为将 SqlNode 转化为关系代数，表达关系代数层次结构的 model

举个例子来说明下，下面这条 flink sql，经过解析之后的 SqlNode，RelNode 如下图：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

SELECT
  sum(part_pv) as pv,
  window_start
FROM (
 SELECT
   count(1) as part_pv,
   cast(tumble_start(rowtime, INTERVAL '60' SECOND) as bigint) * 1000 as window_start
 FROM
   source_db.source_table
 GROUP BY
   tumble(rowtime, INTERVAL '60' SECOND)
   , mod(id, 1024)
)
GROUP BY
  window_start

62

可以看到 SqlNode 包含的内容是 sql 的层次结构，包括 selectList，from，where，group by 等。

RelNode 包含的是关系代数的层次结构，每一层都有一个 input 来承接。结合上面优化案例的树状结构一样。

63

4.4.calcite 的处理流程（以 flink sql 为例）

29

如上图所示，此处我们结合上节介绍的 calcite 的 model，以及 flink sql 的实现来走一遍其处理流程：

1. sql 解析阶段（sql –> SqlNode）
2. SqlNode 验证（SqlNode –> SqlNode）
3. 语义分析（SqlNode –> RelNode）
4. 优化阶段（RelNode –> RelNode）

4.4.1.flink sql demo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

SELECT
  sum(part_pv) as pv,
  window_start
FROM (
 SELECT
   count(1) as part_pv,
   cast(tumble_start(rowtime, INTERVAL '60' SECOND) as bigint) * 1000 as window_start
 FROM
   source_db.source_table
 GROUP BY
   tumble(rowtime, INTERVAL '60' SECOND)
   , mod(id, 1024)
)
GROUP BY
  window_start

其中前三步解析和转化，都在 在执行 TableEnvironment#sqlQuery 进行。

最后一步优化，在执行 sink 操作时进行，即在这个例子中是 tEnv.toRetractStream(result, Row.class)。

源码公众号后台回复flink sql 知其所以然（六）| flink sql 约会 calcite获取。

4.4.2.sql 解析阶段（sql –> SqlNode）

sql 解析阶段使用 Sql Parser 将 sql 解析为 SqlNode。这一步在执行 TableEnvironment#sqlQuery 进行。

可以从上图看到 flink sql 具体实现类是 FlinkSqlParserImpl。

68

具体 parse 得到 SqlNode 如上图。

4.4.3.SqlNode 验证（SqlNode –> SqlNode）

上面的第一步生产的 SqlNode 对象是一个未经验证的，这一步就是语法检查阶段，语法检查前需要知道元数据信息，这个检查会包括表名、字段名、函数名、数据类型的检查。进行语法检查的实现如下：

可以从上图看到 flink sql 校验器的具体实现类是 FlinkCalciteSqlValidator，其中包含了元数据信息，从而可以进行元数据信息检查。

4.4.4.语义分析（SqlNode –> RelNode）

这一步就是将 SqlNode 转换成 RelNode，也就是生成相应的关系代数层面的逻辑（这里一般都叫做逻辑计划：Logical Plan）。

4.4.5.优化阶段（RelNode –> RelNode）

这一步就是优化阶段。详细内容可以自己 debug 代码查看，此处不赘述。

4.5.calcite 怎么做到这么通用？

此处以 calcite parser 举例说明，其模块为什么这通用？其他的模块都是类似的方式。

先说结论：因为 calcite parser 模块提供了接口，具体的 parse 逻辑、规则是可以根据用户自定义进行配置的。大家可以看下图，博主画出了一张图进行详述。

5

如上图，引擎 sql 解析器的生成是有一个输入的，就是 用户自定义语法分析规则变量，具体引擎的 sql 解析器其实也是根据用户自定义的 解析规则 去生成的 解析器。其 解析器 的动态生成依赖 javacc 这样的组件。calcite 提供的是统一的 sql AST 模型、优化模型接口等，而具体的解析实现交给了用户自己去决定。

javacc 会根据 calcite 中定义的 Parser.jj 文件，生成具体的 sql parser 代码（如上图），这个 sql parser 的能力就是将 sql 转换成 AST （SqlNode）。关于 calcite 能力的更详细内容见 https://matt33.com/2019/03/07/apache-calcite-process-flow/ 。

上图涉及到的文件大家可以下载 calcite 源码 https://github.com/apache/calcite.git 之后，切换到 coremodule 之后查看。

31

4.5.1.javacc 是啥？

javacc 是一个用 java 开发的最受欢迎的语法分析生成器。这个分析生成器工具可以读取上下文无关且有着特殊意义的语法并把它转换成可以识别且匹配该语法的 java 程序。它是 100% 的纯 java 代码，可以在多种平台上运行。

简单解释 javacc 就是它是一个通用的语法分析生产器，用户可以使用 javacc 任意定义一套 DSL 及解析器。

举个例子，如果哪天你觉得 sql 也不够简洁通用，你可以使用 javacc 自己定义一套更简洁的 user-define-ql。然后使用 javacc 作为你的 user-define-ql 的解析器。是不是很流批，可以自己去搞编译器了。

4.5.2.跑跑 javacc

这里不介绍具体的 javacc 语法，直接以官网的 Simple1.jj 为案例。详细语法和功能可以参考官网（https://javacc.github.io/javacc/） 或者一下博客。

1. https://www.cnblogs.com/Gavin_Liu/archive/2009/03/07/1405029.html
2. https://www.yangguo.info/2014/12/13/%E7%BC%96%E8%AF%91%E5%8E%9F%E7%90%86-Javacc%E4%BD%BF%E7%94%A8/
3. https://www.engr.mun.ca/~theo/JavaCC-Tutorial/javacc-tutorial.pdf

Simple1.jj 是用于识别一系列的 {相同数量的花括号}，之后跟着 0 个或多个行终结符。

7

下面是合法的字符串例子：

{}，{{{{{}}}}}，etc.

下面是不合法的字符串例子：

{{{{，{}{}，{}}，{{}{}}，etc.

接下来让我们实际将 Simple1.jj 编译生成具体的规则代码。

在 pom 中加入 javacc build 插件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

<plugin>
    <!-- This must be run AFTER the fmpp-maven-plugin -->
    <groupId>org.codehaus.mojo</groupId>
    <artifactId>javacc-maven-plugin</artifactId>
    <version>2.4</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <id>javacc</id>
            <goals>
                <goal>javacc</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.build.directory}/generated-sources/</sourceDirectory>
                <includes>
                    <include>**/Simple1.jj</include>
                </includes>
                <!-- This must be kept synced with Apache Calcite. -->
                <lookAhead>1</lookAhead>
                <isStatic>false</isStatic>
                <outputDirectory>${project.build.directory}/generated-sources/</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

在 compile 之后，就会在 generated-sources 下生成代码：

8

然后把代码 copy 到 Sources 路径下：

33

执行下代码，可以看到 {}，{{}} 都可以校验通过，一旦出现不符合规则的 {{ 输入，就会抛出异常。

关于 javacc 基本上就了解个大概了。

感兴趣的可以尝试自定义一个编译器。

4.5.3.fmpp 是啥？

5

fmpp 就是一个基于 freemarker 的模板生产器。用户可以统一管理自己的变量，然后用 ftl 模板 + 变量 生成对应的最终文件。在 calcite 中使用 fmpp 作为变量 + 模板的统一管理器。然后基于 fmpp 来生成对应的 Parser.jj 文件。

5.原理剖析篇-calcite 在 flink sql 中大展身手

博主画了一张图，包含了其中重要组件之间的依赖关系。

3

你没猜错，还是上面那些流程，fmpp（Parser.jj 模板生成） -> javacc（Parser 生成） -> calcite。

在介绍 Parser 生成流程之前，先看看 flink 最终生成的 Parser：FlinkSqlParserImpl （此处使用 Blink Planner）。

5.1.FlinkSqlParserImpl

以下面这个案例出发（代码基于 flink 1.13.1 版本）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

public class ParserTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(10);

        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        DataStream<Tuple3<String, Long, Long>> tuple3DataStream =
                env.fromCollection(Arrays.asList(
                        Tuple3.of("2", 1L, 1627254000000L),
                        Tuple3.of("2", 1L, 1627218000000L + 5000L),
                        Tuple3.of("2", 101L, 1627218000000L + 6000L),
                        Tuple3.of("2", 201L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 7000L),
                        Tuple3.of("2", 301L, 1627218000000L + 86400000 + 7000L)))
                        .assignTimestampsAndWatermarks(
                                new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Long>>(Time.seconds(0L)) {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, Long, Long> element) {
                                        return element.f2;
                                    }
                                });

        tEnv.registerFunction("mod", new Mod_UDF());

        tEnv.registerFunction("status_mapper", new StatusMapper_UDF());

        tEnv.createTemporaryView("source_db.source_table", tuple3DataStream,
                "status, id, timestamp, rowtime.rowtime");

        String sql = "SELECT\n"
                + "  count(1),\n"
                + "  cast(tumble_start(rowtime, INTERVAL '1' DAY) as string)\n"
                + "FROM\n"
                + "  source_db.source_table\n"
                + "GROUP BY\n"
                + "  tumble(rowtime, INTERVAL '1' DAY)";

        Table result = tEnv.sqlQuery(sql);

        tEnv.toAppendStream(result, Row.class).print();

        env.execute();

    }

}

debug 过程如之前分析 sql -> SqlNode 过程所示，如下图直接定位到 SqlParser：

21

如上图可以看到具体的 Parser 就是 FlinkSqlParserImpl。

定位到具体的代码如下图所示（flink-table-palnner-blink-2.11-1.13.1.jar）。

34

最终 parse 的结果 SqlNode 如下图。

22

再来看看 FlinkSqlParserImpl 是怎么使用 calcite 生成的。

具体到 flink 中的实现，位于源码中的 flink-table.flink-sql-parser 模块（源码基于 flink 1.13.1）。

flink 是依赖 maven 插件实现的上面的整体流程。

5.2.FlinkSqlParserImpl 的生成

14

接下来看看整个 Parser 生成流程。

5.2.1.flink 引入 calcite

使用 maven-dependency-plugin 将 calcite 解压到 flink 项目 build 目录下。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

<plugin>
    <!-- Extract parser grammar template from calcite-core.jar and put
         it under ${project.build.directory} where all freemarker templates are. -->
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-dependency-plugin</artifactId>
    <executions>
        <execution>
            <id>unpack-parser-template</id>
            <phase>initialize</phase>
            <goals>
                <goal>unpack</goal>
            </goals>
            <configuration>
                <artifactItems>
                    <artifactItem>
                        <groupId>org.apache.calcite</groupId>
                        <artifactId>calcite-core</artifactId>
                        <type>jar</type>
                        <overWrite>true</overWrite>
                        <outputDirectory>${project.build.directory}/</outputDirectory>
                        <includes>**/Parser.jj</includes>
                    </artifactItem>
                </artifactItems>
            </configuration>
        </execution>
    </executions>
</plugin>

15

5.2.2.fmpp 生成 Parser.jj

使用 maven-resources-plugin 将 Parser.jj 代码生成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

<plugin>
    <artifactId>maven-resources-plugin</artifactId>
    <executions>
        <execution>
            <id>copy-fmpp-resources</id>
            <phase>initialize</phase>
            <goals>
                <goal>copy-resources</goal>
            </goals>
            <configuration>
                <outputDirectory>${project.build.directory}/codegen</outputDirectory>
                <resources>
                    <resource>
                        <directory>src/main/codegen</directory>
                        <filtering>false</filtering>
                    </resource>
                </resources>
            </configuration>
        </execution>
    </executions>
</plugin>
<plugin>
    <groupId>com.googlecode.fmpp-maven-plugin</groupId>
    <artifactId>fmpp-maven-plugin</artifactId>
    <version>1.0</version>
    <dependencies>
        <dependency>
            <groupId>org.freemarker</groupId>
            <artifactId>freemarker</artifactId>
            <version>2.3.28</version>
        </dependency>
    </dependencies>
    <executions>
        <execution>
            <id>generate-fmpp-sources</id>
            <phase>generate-sources</phase>
            <goals>
                <goal>generate</goal>
            </goals>
            <configuration>
                <cfgFile>${project.build.directory}/codegen/config.fmpp</cfgFile>
                <outputDirectory>target/generated-sources</outputDirectory>
                <templateDirectory>${project.build.directory}/codegen/templates</templateDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

16

5.2.3.javacc 生成 parser

使用 javacc 将根据 Parser.jj 文件生成 Parser。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

<plugin>
    <!-- This must be run AFTER the fmpp-maven-plugin -->
    <groupId>org.codehaus.mojo</groupId>
    <artifactId>javacc-maven-plugin</artifactId>
    <version>2.4</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <id>javacc</id>
            <goals>
                <goal>javacc</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.build.directory}/generated-sources/</sourceDirectory>
                <includes>
                    <include>**/Parser.jj</include>
                </includes>
                <!-- This must be kept synced with Apache Calcite. -->
                <lookAhead>1</lookAhead>
                <isStatic>false</isStatic>
                <outputDirectory>${project.build.directory}/generated-sources/</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

17

5.2.4.看看 Parser

最终生成的 Parser 就是 FlinkSqlParserImpl。

18

5.2.5.blink planner 引入 flink-sql-parser

blink planner（flink-table-planner-blink） 在打包时将 flink-sql-parser、flink-sql-parser-hive 打包进去。

35

从TimeoutException看Flink的心跳机制

0x00 摘要

本文从一个调试时候常见的异常 “TimeoutException: Heartbeat of TaskManager timed out”切入，为大家剖析Flink的心跳机制。文中代码基于Flink 1.10。

0x01 缘由

大家如果经常调试Flink，当进入断点看到了堆栈和变量内容之后，你容易陷入了沉思。当你发现了问题可能所在，高兴的让程序Resume的时候，你发现程序无法运行，有如下提示：

1

Caused by: java.util.concurrent.TimeoutException: Heartbeat of TaskManager with id 93aa1740-cd2c-4032-b74a-5f256edb3217 timed out.

这实在是很郁闷的事情。作为程序猿不能忍啊，既然异常提示中有 Heartbeat 字样，于是我们就来一起看看Flink的心跳机制，看看有没有可以修改的途径。

0x02 背景概念

2.1 四大模块

Flink有核心四大组件：Dispatcher，JobMaster，ResourceManager，TaskExecutor。

• Dispatcher（Application Master）用于接收client提交的任务和启动相应的JobManager。其提供REST接口来接收client的application提交，负责启动JM和提交application，同时运行Web UI。

• ResourceManager：主要用于资源的申请和分配。当TM有空闲的slot就会告诉JM，没有足够的slot也会启动新的TM。kill掉长时间空闲的TM。

• JobMaster

  ：功能主要包括（旧版本中JobManager的功能在新版本中以JobMaster形式出现，可能本文中会混淆这两个词，请大家谅解）：

  • 将JobGraph转化为ExecutionGraph（physical dataflow graph，并行化）。
  • 向RM申请资源、schedule tasks、保存作业的元数据。

• TaskManager：类似Spark的executor，会跑多个线程的task、数据缓存与交换。Flink 架构遵循 Master - Slave 架构设计原则，JobMaster 为 Master 节点，TaskManager 为Slave节点。

这四大组件彼此之间的通信需要依赖RPC实现。

2.2 Akka

Flink底层RPC基于Akka实现。Akka是一个开发并发、容错和可伸缩应用的框架。它是Actor Model的一个实现，和Erlang的并发模型很像。在Actor模型中，所有的实体被认为是独立的actors。actors和其他actors通过发送异步消息通信。

Actor模型的强大来自于异步。它也可以显式等待响应，这使得可以执行同步操作。但是强烈不建议同步消息，因为它们限制了系统的伸缩性。

2.3 RPC机制

RPC作用是：让异步调用看起来像同步调用。

Flink基于Akka构建了其底层通信系统，引入了RPC调用，各节点通过GateWay方式回调，隐藏通信组件的细节，实现解耦。Flink整个通信框架的组件主要由RpcEndpoint、RpcService、RpcServer、AkkaInvocationHandler、AkkaRpcActor等构成。

RPC相关的主要接口如下：

• RpcEndpoint
• RpcService
• RpcGateway

2.3.1 RpcEndpoint：RPC的基类

RpcEndpoint是Flink RPC终端的基类，所有提供远程过程调用的分布式组件必须扩展RpcEndpoint，其功能由RpcService支持。

RpcEndpoint的子类只有四类组件：Dispatcher，JobMaster，ResourceManager，TaskExecutor，即Flink中只有这四个组件有RPC的能力，换句话说只有这四个组件有RPC的这个需求。

每个RpcEndpoint对应了一个路径（endpointId和actorSystem共同确定），每个路径对应一个Actor，其实现了RpcGateway接口，

RpcService：RPC服务提供者

RpcServer是RpcEndpoint的成员变量，为RpcService提供RPC服务/连接远程Server，其只有一个子类实现：AkkaRpcService（可见目前Flink的通信方式依然是Akka）。

RpcServer用于启动和连接到RpcEndpoint， 连接到rpc服务器将返回一个RpcGateway，可用于调用远程过程。

Flink四大组件Dispatcher，JobMaster，ResourceManager，TaskExecutor，都是RpcEndpoint的实现，所以构建四大组件时，同步需要初始化RpcServer。如JobManager的构造方式，第一个参数就是需要知道RpcService。

RpcGateway：RPC调用的网关

Flink的RPC协议通过RpcGateway来定义；由前面可知，若想与远端Actor通信，则必须提供地址（ip和port），如在Flink-on-Yarn模式下，JobMaster会先启动ActorSystem，此时TaskExecutor的Container还未分配，后面与TaskExecutor通信时，必须让其提供对应地址。

Dispatcher，JobMaster，ResourceManager，TaskExecutor 这四大组件通过各种方式实现了Gateway。以JobMaster为例，JobMaster实现JobMasterGateway接口。各组件类的成员变量都有需要通信的其他组件的GateWay实现类，这样可通过各自的Gateway实现RPC调用。

2.4 常见心跳机制

常见的心跳检测有两种：

• socket 套接字SO_KEEPALIVE本身带有的心跳机制，定期向对方发送心跳包，对方收到心跳包后会自动回复；
• 应用自身实现心跳机制，同样也是使用定期发送请求的方式；

Flink实现的是第二种方案。

0x03 Flink心跳机制

3.1 代码和机制

Flink的心跳机制代码在：

1

Flink-master/flink-runtime/src/main/java/org/apache/flink/runtime/heartbeat

四个接口：

1

HeartbeatListener.java          HeartbeatManager.java      HeartbeatTarget.java  HeartbeatMonitor.java

以及如下几个类：

1
2

HeartbeatManagerImpl.java   HeartbeatManagerSenderImpl.java   HeartbeatMonitorImpl.java
HeartbeatServices.java    NoOpHeartbeatManager.java

Flink集群有多种业务流程，比如Resource Manager, Task Manager, Job Manager。每种业务流程都有自己的心跳机制。Flink的心跳机制只是提供接口和基本功能，具体业务功能由各业务流程自己实现。

我们首先设定 心跳系统中有两种节点：sender和receiver。心跳机制是sender和receivers彼此相互检测。但是检测动作是Sender主动发起，即Sender主动发送请求探测receiver是否存活，因为Sender已经发送过来了探测心跳请求，所以这样receiver同时也知道Sender是存活的，然后Reciver给Sender回应一个心跳表示自己也是活着的。

因为Flink的几个名词和我们常见概念有所差别，所以流程上需要大家仔细甄别，即：

• Flink Sender 主动发送Request请求给Receiver，要求Receiver回应一个心跳；
• Flink Receiver 收到Request之后，通过Receive函数回应一个心跳请求给Sender；

3.2 静态架构

3.2.1 HeartbeatTarget ：监控目标抽象

HeartbeatTarget是对监控目标的抽象。心跳机制在行为上而言有两种动作：

• 向某个节点发送请求。
• 处理某个节点发来的请求。

HeartbeatTarget的函数就是这两个动作：

• receiveHeartbeat ：向某个节点（Sender）发送心跳回应，其参数heartbeatOrigin 就是 Receiver。
• requestHeartbeat ：向某个节点（Receiver）要求其回应一个心跳，其参数requestOrigin 就是 Sender。requestHeartbeat这个函数是Sender的函数，其中Sender通过RPC直接调用到Receiver。

这两个函数的参数也很简单：分别是请求的发送放和接收方，还有Payload载荷。对于一个确定节点而言，接收的和发送的载荷是同一类型的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public interface HeartbeatTarget<I> {
	/**
	 * Sends a heartbeat response to the target.
	 * @param heartbeatOrigin Resource ID identifying the machine for which a heartbeat shall be reported.
	 */
  // heartbeatOrigin 就是 Receiver
	void receiveHeartbeat(ResourceID heartbeatOrigin, I heartbeatPayload);

	/**
	 * Requests a heartbeat from the target. 
	 * @param requestOrigin Resource ID identifying the machine issuing the heartbeat request.
	 */
  // requestOrigin 就是 Sender
	void requestHeartbeat(ResourceID requestOrigin, I heartbeatPayload);
}

3.2.2 HeartbeatMonitor : 管理heartbeat target的心跳状态

对HeartbeatTarget的封装，这样Manager对Target的操作是通过对Monitor完成，后续会在其继承类中详细说明。

1
2
3
4
5
6
7
8
9
10
11
12

public interface HeartbeatMonitor<O> {
  // Gets heartbeat target.
	HeartbeatTarget<O> getHeartbeatTarget();
	// Gets heartbeat target id.
	ResourceID getHeartbeatTargetId();
	// Report heartbeat from the monitored target.
	void reportHeartbeat();
	//Cancel this monitor.
	void cancel();
  //Gets the last heartbeat.
	long getLastHeartbeat();
}

3.2.3 HeartbeatManager ：心跳管理者

HeartbeatManager负责管理心跳机制，比如启动/停止/报告一个HeartbeatTarget。此接口继承HeartbeatTarget。

除了HeartbeatTarget的函数之外，这接口有4个函数：

• monitorTarget，把和某资源对应的节点加入到心跳监控列表;
• unmonitorTarget，从心跳监控列表删除某资源对应的节点;
• stop，停止心跳管理服务，释放资源;
• getLastHeartbeatFrom，获取某节点的最后一次心跳数据。

1
2
3
4
5
6

public interface HeartbeatManager<I, O> extends HeartbeatTarget<I> {
	void monitorTarget(ResourceID resourceID, HeartbeatTarget<O> heartbeatTarget);
	void unmonitorTarget(ResourceID resourceID);
	void stop();
	long getLastHeartbeatFrom(ResourceID resourceId);
}

3.2.4 HearbeatListener 处理心跳结果

用户业务逻辑需要继承这个接口以处理心跳结果。其可以看做服务的输出，实现了三个回调函数。

• notifyHeartbeatTimeout，处理节点心跳超时
• reportPayload，处理节点发来的Payload载荷
• retrievePayLoad。获取对某节点发下一次心跳请求的Payload载荷

1
2
3
4
5

public interface HeartbeatListener<I, O> {
	void notifyHeartbeatTimeout(ResourceID resourceID);
	void reportPayload(ResourceID resourceID, I payload);
	O retrievePayload(ResourceID resourceID);
}

3.3 动态运行机制

之前提到Sender和Receiver，下面两个类就对应上述概念。

• HeartbeatManagerImpl ：Receiver，存在于JobMaster与TaskExecutor中；
• HeartbeatManagerSenderImpl ：Sender，继承 HeartbeatManagerImpl类，用于周期发送心跳要求，存在于JobMaster、ResourceManager中。

几个关键问题：

1. 如何判定心跳超时？ 心跳服务启动后，Flink在Monitor中通过 ScheduledFuture 会启动一个线程来处理心跳超时事件。在设定的心跳超时时间到达后才执行线程。 如果在设定的心跳超时时间内接收到组件的心跳消息，会先将该线程取消而后重新开启，重置心跳超时事件的触发。 如果在设定的心跳超时时间内没有收到组件的心跳，则会通知组件：你超时了。
2. 何时”调用双方”发起心跳检查？ 心跳检查是双向的，一方（Sender）会主动发起心跳请求，而另一方（Receiver）则是对心跳做出响应，两者通过RPC相互调用，重置对方的 Monitor 超时线程。 以JobMaster和TaskManager为例，JM在启动时会开启周期调度，向已经注册到JM中的TM发起心跳检查，通过RPC调用TM的requestHeartbeat方法，重置TM中对JM超时线程的调用，表示当前JM状态正常。在TM的requestHeartbeat方法被调用后，通过RPC调用JM的receiveHeartbeat，重置 JM 中对TM超时线程的调用，表示TM状态正常。
3. 如何处理心跳超时？ 心跳服务依赖 HeartbeatListener，当在timeout时间范围内未接收到心跳响应，则会触发超时处理线程，该线程通过调用HeartbeatListener.notifyHeartbeatTimeout方法做后续重连操作或者直接断开。

下面是一个概要（以RM & TM为例）：

• RM : 实现了ResourceManagerGateway (可以直接被RPC调用)
• TM : 实现了TaskExecutorGateway (可以直接被RPC调用)
• RM ：有一个Sender HM ： taskManagerHeartbeatManager，Sender HM 拥有用户定义的 TaskManagerHeartbeatListener
• TM ：有一个Receiver HM ：resourceManagerHeartbeatManager，Receiver HM 拥有用户定义的ResourceManagerHeartbeatListener。
• HeartbeatManager 有一个ConcurrentHashMap<ResourceID, HeartbeatMonitor> heartbeatTargets，这个Map是它监控的所有Target。
• 对于RM的每一个需要监控的TM, 其生成一个HeartbeatTarget，进而被构造成一个HeartbeatMonitor，放置到ResourceManager.taskManagerHeartbeatManager中。
• 每一个Target对应的Monitor中，有自己的异步任务ScheduledFuture，这个ScheduledFuture不停的被取消/重新生成。如果在某个期间内没有被取消，则通知用户定义的listener出现了timeout。

3.3.1 HearbeatManagerImpl : Receiver

HearbeatManagerImpl是receiver的具体实现。它由 心跳 被发起方（就是Receiver，例如TM） 创建，接收 **发起方（就是Sender，例如 JM）**的心跳发送请求。心跳超时 会触发 heartbeatListener.notifyHeartbeatTimeout方法。

注意：被发起方监控线程（Monitor）的开启是在接收到请求心跳（requestHeartbeat被调用后）以后才触发的，属于被动触发。

HearbeatManagerImpl主要维护了

• 一个心跳监控列表 map : <ResourceID, HeartbeatMonitor<O>> heartbeatTargets;。这是一个KV关联。 key代表要发送心跳组件（例如：TM）的ID，value则是为当前组件创建的触发心跳超时的线程HeartbeatMonitor，两者一一对应。 当一个从所联系的machine发过来的心跳被收到时候，对应的monitor的状态会被更新（重启一个新ScheduledFuture）。当一个monitor发现了一个 heartbeat timed out，它会通知自己的HeartbeatListener。
• 一个 ScheduledExecutor mainThreadExecutor 负责heartbeat timeout notifications。
• heartbeatListener ：处理心跳结果。

HearbeatManagerImpl 数据结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

@ThreadSafe
public class HeartbeatManagerImpl<I, O> implements HeartbeatManager<I, O> {

	/** Heartbeat timeout interval in milli seconds. */
	private final long heartbeatTimeoutIntervalMs;

	/** Resource ID which is used to mark one own's heartbeat signals. */
	private final ResourceID ownResourceID;

	/** Heartbeat listener with which the heartbeat manager has been associated. */
	private final HeartbeatListener<I, O> heartbeatListener;

	/** Executor service used to run heartbeat timeout notifications. */
	private final ScheduledExecutor mainThreadExecutor;

	/** Map containing the heartbeat monitors associated with the respective resource ID. */
	private final ConcurrentHashMap<ResourceID, HeartbeatMonitor<O>> heartbeatTargets;

	/** Running state of the heartbeat manager. */
	protected volatile boolean stopped;  
}

HearbeatManagerImpl实现的主要函数有：

• monitorTarget ：把一个节点加入到心跳监控列表。
  • 传入参数有：ResourceId和HearbeatTarget，monitorTarget根据这两个参数，生成一个HeartbeatMonitor对象，然后把这个对象跟ResrouceId做kv关联，存入到heartbeatTargets。 一个节点可能参与多个业务流程，因此一个节点参与多个心跳流程，一个节点上运行多个不同类型的HearbeatTarget。所以一个ResourceID可能会跟不同类型的HearbeatTarget对象关联，分别加入到多个HeartbeatManager，进行不同类型的心跳监控。也因此这个函数入参是两个参数。
• requestHeartbeat ：Sender通过RPC异步调用到Receiver的这个函数 以要求receiver向requestOrigin节点（就是Sender）发起一次心跳响应，载荷是heartbeatPayLoad。其内部流程如下：
  • 首先会调用reportHeartbeat函数，作用是 通过Monitor 记录发起请求的这个时间点，然后创建一个ScheduleFuture。如果到期后，requestOrigin没有作出响应，那么就将requestOrigin节点对应的HeartbeatMonitor的state设置成TIMEOUT状态，如果到期内requestOrigin响应了，ScheduleFuture会被取消，HeartbeatMonitor的state仍然是RUNNING。
  • 其次调用reportPayload函数，把requestOrigin节点的最新的heartbeatPayload通知给heartbeatListener。heartbeatListener是外部传入的，它根据所有节点的心跳记录做监听管理。
  • 最后调用receiveHearbeat函数，响应一个心跳给Sender。

3.3.2 HeartbeatManagerSenderImpl : Sender

继承HearbeatManagerImpl，由**心跳管理的一方(例如JM)**创建，实现了run函数（即它可以作为一个单独线程运行），创建后立即开启周期调度线程，每次遍历自己管理的heartbeatTarget，触发heartbeatTarget.requestHeartbeat，要求 Target 返回一个心跳响应。属于主动触发心跳请求。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

public class HeartbeatManagerSenderImpl<I, O> extends HeartbeatManagerImpl<I, O> implements Runnable {
	public void run() {
		if (!stopped) {
			for (HeartbeatMonitor<O> heartbeatMonitor : getHeartbeatTargets().values()) {
				requestHeartbeat(heartbeatMonitor);
			}
      // 周期调度
			getMainThreadExecutor().schedule(this, heartbeatPeriod, TimeUnit.MILLISECONDS);
		}
	}

  //  主动发起心跳检查
	private void requestHeartbeat(HeartbeatMonitor<O> heartbeatMonitor) {
		O payload = getHeartbeatListener().retrievePayload(heartbeatMonitor.getHeartbeatTargetId());
		final HeartbeatTarget<O> heartbeatTarget = heartbeatMonitor.getHeartbeatTarget();
    // 调用 Target 的 requestHeartbeat 函数
		heartbeatTarget.requestHeartbeat(getOwnResourceID(), payload);
	}  
}

3.3.3 HeartbeatMonitorImpl

Heartbeat monitor管理心跳目标，它启动一个ScheduledExecutor。

• 如果在timeout时间内没有接收到心跳信号，则判定心跳超时，通知给HeartbeatListener。
• 如果在timeout时间内接收到心跳信号，则重置当前ScheduledExecutor。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

public class HeartbeatMonitorImpl<O> implements HeartbeatMonitor<O>, Runnable {

	/** Resource ID of the monitored heartbeat target. */
	private final ResourceID resourceID;   // 被监控的resource ID

	/** Associated heartbeat target. */
	private final HeartbeatTarget<O> heartbeatTarget; //心跳目标

	private final ScheduledExecutor scheduledExecutor;

	/** Listener which is notified about heartbeat timeouts. */
	private final HeartbeatListener<?, ?> heartbeatListener; // 心跳监听器

	/** Maximum heartbeat timeout interval. */
	private final long heartbeatTimeoutIntervalMs;

	private volatile ScheduledFuture<?> futureTimeout;
  //  AtomicReference  使用 
	private final AtomicReference<State> state = new AtomicReference<>(State.RUNNING);
   //  最近一次接收到心跳的时间
  private volatile long lastHeartbeat;
  
	// 报告心跳
	public void reportHeartbeat() {
    // 保留最近一次接收心跳时间
		lastHeartbeat = System.currentTimeMillis();
    // 接收心跳后，重置timeout线程
		resetHeartbeatTimeout(heartbeatTimeoutIntervalMs);
	}

  // 心跳超时，触发lister的notifyHeartbeatTimeout
	public void run() {
		// The heartbeat has timed out if we're in state running
		if (state.compareAndSet(State.RUNNING, State.TIMEOUT)) {
			heartbeatListener.notifyHeartbeatTimeout(resourceID);
		}
	}

  //  重置TIMEOUT
	void resetHeartbeatTimeout(long heartbeatTimeout) {
		if (state.get() == State.RUNNING) {
      //先取消线程，在重新开启
			cancelTimeout();
      // 启动超时线程
			futureTimeout = scheduledExecutor.schedule(this, heartbeatTimeout, TimeUnit.MILLISECONDS);

			// Double check for concurrent accesses (e.g. a firing of the scheduled future)
			if (state.get() != State.RUNNING) {
				cancelTimeout();
			}
		}
	}

3.3.3 HeartbeatServices

建立heartbeat receivers and heartbeat senders，主要是对外提供服务。这里我们可以看到：

• HeartbeatManagerImpl就是receivers。
• HeartbeatManagerSenderImpl就是senders。

1
2
3
4
5
6
7
8
9
10

public class HeartbeatServices {
	// Creates a heartbeat manager which does not actively send heartbeats.
	public <I, O> HeartbeatManager<I, O> createHeartbeatManager(...) {
		return new HeartbeatManagerImpl<>(...);
	}
	// Creates a heartbeat manager which actively sends heartbeats to monitoring targets.
	public <I, O> HeartbeatManager<I, O> createHeartbeatManagerSender(...) {
		return new HeartbeatManagerSenderImpl<>(...);
	}
}

0x04 初始化

4.1 心跳服务创建

心跳管理服务在Cluster入口创建。因为我们是调试，所以在MiniCluster.start调用。

1
2
3
4
5

public void start() throws Exception {
				......
				heartbeatServices = HeartbeatServices.fromConfiguration(configuration);
				......
}

HeartbeatServices.fromConfiguration会从Configuration中获取配置信息：

• 心跳间隔 heartbeat.interval
• 心跳超时时间 heartbeat.timeout

这个就是我们解决最开始问题的思路：从配置信息入手，扩大心跳间隔。

1
2
3
4
5
6
7
8
9
10
11

public HeartbeatServices(long heartbeatInterval, long heartbeatTimeout) {
		this.heartbeatInterval = heartbeatInterval;
		this.heartbeatTimeout = heartbeatTimeout;
}

public static HeartbeatServices fromConfiguration(Configuration configuration) {
		long heartbeatInterval = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL);
		long heartbeatTimeout = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT);

		return new HeartbeatServices(heartbeatInterval, heartbeatTimeout);
}

0x05 Flink中具体应用

5.1 总述

5.1.1 RM, JM, TM之间关系

系统中有几个ResourceManager？整个 Flink 集群中只有一个 ResourceManager。

系统中有几个JobManager？JobManager 负责管理作业的执行。默认情况下，每个 Flink 集群只有一个 JobManager 实例。JobManager 相当于整个集群的 Master 节点，负责整个集群的任务管理和资源管理。

系统中有几个TaskManager？这个由具体启动方式决定。比如Flink on Yarn，Session模式能够指定拉起多少个TaskManager。 Per job模式中TaskManager数量是在提交作业时根据并发度动态计算，即Number of TM = Parallelism/numberOfTaskSlots。比如：有一个作业，Parallelism为10，numberOfTaskSlots为1，则TaskManager为10。

5.1.2 三者间心跳机制

Flink中ResourceManager、JobMaster、TaskExecutor三者之间存在相互检测的心跳机制：

• ResourceManager会主动发送请求探测JobMaster、TaskExecutor是否存活。
• JobMaster也会主动发送请求探测TaskExecutor是否存活，以便进行任务重启或者失败处理。

我们之前讲过，HeartbeatManagerSenderImpl属于Sender，HeartbeatManagerImpl属于Receiver。

1. HeartbeatManagerImpl所处位置可以理解为client，存在于JobMaster与TaskExecutor中；
2. HeartbeatManagerSenderImpl类，继承 HeartbeatManagerImpl类，用于周期发送心跳请求，所处位置可以理解为server, 存在于JobMaster、ResourceManager中。

ResourceManager 级别最高，所以两个HM都是Sender，监控taskManager和jobManager

1
2
3
4
5
6

public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
		extends FencedRpcEndpoint<ResourceManagerId>
		implements ResourceManagerGateway, LeaderContender {
	taskManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender
	jobManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender
}

JobMaster级别中等，一个Sender, 一个Receiver，受到ResourceManager的监控，监控taskManager。

1
2
3
4

public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService {
	taskManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender
	resourceManagerHeartbeatManager = heartbeatServices.createHeartbeatManager
}

TaskExecutor级别最低，两个Receiver，分别被JM和RM疾控。

1
2
3
4

public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
	this.jobManagerHeartbeatManager = return heartbeatServices.createHeartbeatManager
	this.resourceManagerHeartbeatManager = return heartbeatServices.createHeartbeatManager
}

以JobManager和TaskManager为例。JM在启动时会开启周期调度，向已经注册到JM中的TM发起心跳检查，通过RPC调用TM的requestHeartbeat方法，重置对JM超时线程的调用，表示当前JM状态正常。在TM的requestHeartbeat方法被调用后，通过RPC调用JM的receiveHeartbeat，重置对TM超时线程的调用，表示TM状态正常。

5.2 初始化过程

5.2.1 TaskExecutor初始化

TM初始化生成了两个Receiver HM。

1
2
3
4
5
6
7
8
9
10
11

public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
	/** The heartbeat manager for job manager in the task manager. */
	private final HeartbeatManager<AllocatedSlotReport, AccumulatorReport> jobManagerHeartbeatManager;

	/** The heartbeat manager for resource manager in the task manager. */
	private final HeartbeatManager<Void, TaskExecutorHeartbeatPayload> resourceManagerHeartbeatManager;
  
  //初始化函数
	this.jobManagerHeartbeatManager = createJobManagerHeartbeatManager(heartbeatServices, resourceId);
	this.resourceManagerHeartbeatManager = createResourceManagerHeartbeatManager(heartbeatServices, resourceId);
}

生成HeartbeatManager时，就注册了ResourceManagerHeartbeatListener和JobManagerHeartbeatListener。

此时，两个HeartbeatManagerImpl中已经创建好对应monitor线程，只有在JM或者RM执行requestHeartbeat后，才会触发该线程的执行。

5.2.2 JobMaster的初始化

JM生成了一个Sender HM，一个Receiver HM。这里会注册 TaskManagerHeartbeatListener 和 ResourceManagerHeartbeatListener

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMasterGateway, JobMasterService {
  private HeartbeatManager<AccumulatorReport, AllocatedSlotReport> taskManagerHeartbeatManager;
  private HeartbeatManager<Void, Void> resourceManagerHeartbeatManager;

  private void startHeartbeatServices() {
      taskManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender(
        resourceId,
        new TaskManagerHeartbeatListener(),
        getMainThreadExecutor(),
        log);

      resourceManagerHeartbeatManager = heartbeatServices.createHeartbeatManager(
        resourceId,
        new ResourceManagerHeartbeatListener(),
        getMainThreadExecutor(),
        log);
  }
}

5.2.3 ResourceManager初始化

JobMaster在启动时候，会在startHeartbeatServices函数中生成两个Sender HeartbeatManager。

taskManagerHeartbeatManager ：HeartbeatManagerSenderImpl对象，会反复启动一个定时器，定时扫描需要探测的对象并且发送心跳请求。

jobManagerHeartbeatManager ：HeartbeatManagerSenderImpl，会反复启动一个定时器，定时扫描需要探测的对象并且发送心跳请求。

1
2
3
4
5
6
7
8
9
10
11

taskManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender(
			resourceId,
			new TaskManagerHeartbeatListener(),
			getMainThreadExecutor(),
			log);

jobManagerHeartbeatManager = heartbeatServices.createHeartbeatManagerSender(
			resourceId,
			new JobManagerHeartbeatListener(),
			getMainThreadExecutor(),
			log);  

5.3 注册过程

我们以TM与RM交互为例。TaskExecutor启动之后，需要注册到RM和JM中。

流程图如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

* 1. Run in Task Manager
*
*    TaskExecutor.onStart //Life cycle
*        |
*        +----> startTaskExecutorServices@TaskExecutor
*        |     //开始TM服务
*        |     
*        +----> resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
*        |     // 开始连接到RM
*        |     // start by connecting to the ResourceManager
*        |    
*        +----> notifyLeaderAddress@ResourceManagerLeaderListener  
*        |     // 当RM状态变化之后，将回调到这里    
*        |     // The listener for leader changes of the resource manager.
*        |        
*        +----> reconnectToResourceManager@TaskExecutor   
*        |     // 以下三步调用是渐进的，就是与RM联系。
*        |    
*        +----> tryConnectToResourceManager@TaskExecutor 
*        |     
*        +----> connectToResourceManager()@TaskExecutor
*        |     // 主要作用是生成了 TaskExecutorToResourceManagerConnection    
*        |  
*        +----> start@TaskExecutorToResourceManagerConnection
*        |     // 开始RPC调用，将会调用到其基类RegisteredRpcConnection的start   
*        |   
*        +----> start@RegisteredRpcConnection
*        |     // RegisteredRpcConnection实现了组件之间注册联系的基本RPC
*        |      
  
  
* ~~~~~~~~ 这里是 Akka RPC
  
* 2. Run in Resource Manager   
* 现在程序执行序列到达了RM, 主要是添加一个Target到RM 的 Sender HM；
*
*    registerTaskExecutor@ResourceManager
*        |
*        +----> taskExecutorGatewayFuture.handleAsync
*        |     // 异步调用到这里
*        |     
*        +----> registerTaskExecutorInternal@ResourceManager
*        |     // RM的内部实现，将把TM注册到RM自己这里
*        |      
*        +----> taskManagerHeartbeatManager.monitorTarget
*        |     // 生成HeartbeatMonitor，
*        |      
*        +---->  heartbeatTargets.put(resourceID,heartbeatMonitor);
*        |     // 把Monitor放到 HM in TM之中，就是说TM开始监控了RM
*        |        

* ~~~~~~~~ 这里是 Akka RPC
 
* 3. Run in Task Manager
* 现在程序回到了TM, 主要是添加一个Target到 TM 的 Receiver HM；
*
*    onRegistrationSuccess@TaskExecutorToResourceManagerConnection
*        |   
*        |   
*        +---->  onRegistrationSuccess@ResourceManagerRegistrationListener 
*        |       // 回调函数
*        |  
*        +---->  runAsync(establishResourceManagerConnection) 
*        |       // 异步执行
*        |  
*        +---->  establishResourceManagerConnection@TaskExecutor 
*        |       // 说明已经和RM建立了联系，所以可以开始监控RM了
*        |  
*        +---->  resourceManagerHeartbeatManager.monitorTarget 
*        |     // 生成HeartbeatMonitor，
*        |      
*        +---->  heartbeatTargets.put(resourceID,heartbeatMonitor);  
*        |       // 把 RM 也注册到 TM了
*        |       // monitor the resource manager as heartbeat target 

下面是具体文字描述。

5.3.1 TM注册到RM中

5.3.1.1 TM的操作

• TaskExecutor启动之后，调用onStart，开始其生命周期。
• onStart直接调用startTaskExecutorServices。
• 启动服务的第一步就是与ResourceManager取得联系，这里注册了一个ResourceManagerLeaderListener()，用来监听RM Leader的变化。

1
2
3
4

private final LeaderRetrievalService resourceManagerLeaderRetriever;
// resourceManagerLeaderRetriever其实是EmbeddedLeaderService的实现，A simple leader election service, which selects a leader among contenders and notifies listeners.

resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());

• 当得到RM Leader的地址之后，会调用到回调函数notifyLeaderAddress@ResourceManagerLeaderListener，然后调用notifyOfNewResourceManagerLeader。
• notifyOfNewResourceManagerLeader中获取到RM地址后，就通过reconnectToResourceManager与RM联系。
• reconnectToResourceManager中间接调用到TaskExecutorToResourceManagerConnection。其作用是建立TaskExecutor 和 ResourceManager之间的联系。因为知道 ResourceManagerGateway所以才能进行RPC操作。
• 然后在 TaskExecutorToResourceManagerConnection中，就通过RPC与RM联系。

5.3.1.2 RM的操作

• RPC调用后，程序就来到了RM中，RM做如下操作：
• 会注册一个新的TaskExecutor到自己的taskManagerHeartbeatManager中。
• registerTaskExecutor@ResourceManager会通过异步调用到registerTaskExecutorInternal。
• registerTaskExecutorInternal中首先看看是否这个TaskExecutor的ResourceID之前注册过，如果注册过就移除再添加一个新的TaskExecutor。
• 通过 taskManagerHeartbeatManager.monitorTarget 开始进行心跳机制的注册。

1
2
3
4
5
6
7
8
9

taskManagerHeartbeatManager.monitorTarget(taskExecutorResourceId, new HeartbeatTarget<Void>() {
				public void receiveHeartbeat(ResourceID resourceID, Void payload) {
					// the ResourceManager will always send heartbeat requests to the
					// TaskManager
				}
				public void requestHeartbeat(ResourceID resourceID, Void payload) {
					taskExecutorGateway.heartbeatFromResourceManager(resourceID);
				}
});

当注册完成后，RM中的Sender HM内部结构如下，能看出来多了一个Target：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

taskManagerHeartbeatManager = {HeartbeatManagerSenderImpl@8866} 
 heartbeatPeriod = 10000
 heartbeatTimeoutIntervalMs = 50000
 ownResourceID = {ResourceID@8871} "040709f36ebf38f309fed518a88946af"
 heartbeatListener = {ResourceManager$TaskManagerHeartbeatListener@8872} 
 mainThreadExecutor = {RpcEndpoint$MainThreadExecutor@8873} 
 heartbeatTargets = {ConcurrentHashMap@8875}  size = 1
  {ResourceID@8867} "630c15c9-4861-4b41-9c95-92504f458b71" -> {HeartbeatMonitorImpl@9448} 
   key = {ResourceID@8867} "630c15c9-4861-4b41-9c95-92504f458b71"
   value = {HeartbeatMonitorImpl@9448} 
    resourceID = {ResourceID@8867} "630c15c9-4861-4b41-9c95-92504f458b71"
    heartbeatTarget = {ResourceManager$2@8868} 
    scheduledExecutor = {RpcEndpoint$MainThreadExecutor@8873} 
    heartbeatListener = {ResourceManager$TaskManagerHeartbeatListener@8872} 
    heartbeatTimeoutIntervalMs = 50000
    futureTimeout = {ScheduledFutureAdapter@10140} 
    state = {AtomicReference@9786} "RUNNING"
    lastHeartbeat = 0

5.3.1.3 返回到TM

RM会通过RPC再次回到TaskExecutor，其新执行序列如下：

• 首先RPC调用到了 onRegistrationSuccess@TaskExecutorToResourceManagerConnection。
• 然后onRegistrationSuccess@ResourceManagerRegistrationListener中通过异步执行调用到了establishResourceManagerConnection。这说明TM已经和RM建立了联系，所以可以开始监控RM了。
• 然后和RM操作类似，通过resourceManagerHeartbeatManager.monitorTarget 来把RM注册到自己这里。

1
2

HeartbeatMonitor<O> heartbeatMonitor = heartbeatMonitorFactory.createHeartbeatMonitor 
heartbeatTargets.put(resourceID, heartbeatMonitor);  

当注册完成后，其Receiver HM结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

resourceManagerHeartbeatManager = {HeartbeatManagerImpl@10163} 
 heartbeatTimeoutIntervalMs = 50000
 ownResourceID = {ResourceID@8882} "96a9b80c-dd97-4b63-9049-afb6662ea3e2"
 heartbeatListener = {TaskExecutor$ResourceManagerHeartbeatListener@10425} 
 mainThreadExecutor = {RpcEndpoint$MainThreadExecutor@10426} 
 heartbeatTargets = {ConcurrentHashMap@10427}  size = 1
  {ResourceID@8886} "122fa66685133b11ea26ee1b1a6cef75" -> {HeartbeatMonitorImpl@10666} 
   key = {ResourceID@8886} "122fa66685133b11ea26ee1b1a6cef75"
   value = {HeartbeatMonitorImpl@10666} 
    resourceID = {ResourceID@8886} "122fa66685133b11ea26ee1b1a6cef75"
    heartbeatTarget = {TaskExecutor$1@10668} 
    scheduledExecutor = {RpcEndpoint$MainThreadExecutor@10426} 
    heartbeatListener = {TaskExecutor$ResourceManagerHeartbeatListener@10425} 
    heartbeatTimeoutIntervalMs = 50000
    futureTimeout = {ScheduledFutureAdapter@10992} 
    state = {AtomicReference@10667} "RUNNING"
    lastHeartbeat = 0

5.3.2 TM注册到 JM

其调用基本思路与之前相同，就是TM和JM之间互相注册一个代表对方的monitor：

1

JobLeaderListenerImpl ----> establishJobManagerConnection

消息到了JM中，做如下操作。

1
2

registerTaskManager ----> taskManagerHeartbeatManager.monitorTarget
  // monitor the task manager as heartbeat target

5.4 心跳过程

在任务提交之后，我们就进入了正常的心跳监控流程。我们依然用 TM 和 RM进行演示。

我们先给出一个流程图。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

* 1. Run in Resouce Manager
*
*    HeartbeatManagerSender in RM
*        |
*        +----> run@HeartbeatManagerSenderImpl
*        |     //遍历所有监控的Monitor（Target），逐一在Target上调用requestHeartbeat
*        |     
*        +----> requestHeartbeat@HeartbeatManagerSenderImpl 
*        |     // 将调用具体监控对象的自定义函数
*        |     // heartbeatTarget.requestHeartbeat(getOwnResourceID(), payload);
*        |      
*        +----> getHeartbeatListener().retrievePayload   
*        |     // 调用到TaskManagerHeartbeatListener@ResourceManager    
*        |     // 这里是return null;，因为RM不会是任何人的Receiver
*        |        
*        +----> requestHeartbeat@HeartbeatTarget   
*        |     // 调用到Target这里，代码在ResourceManager这里，就是生成Target时候赋值的
*        |    
*        +----> taskExecutorGateway.heartbeatFromResourceManager
*        |     // 会通过gateway RPC 调用到TM，这就是主动对TM发起了心跳请求
*        |      

* ~~~~~~~~ 这里是 Akka RPC
  
* 2. Run in Task Manager   
* 现在程序执行序列到达了TM, 主要是 1. 重置TM的Monitor线程; 2.返回一些负载信息；
*
*    heartbeatFromResourceManager@TaskExecutor
*        |
*        +----> resourceManagerHeartbeatManager.requestHeartbeat(resourceID, null);
*        |     //开始要调用到 Receiver HM in Task Manager
*        |     
*        +----> requestHeartbeat@HeartbeatManager in TM 
*        |     // 在Receiver HM in Task Manager 这里运行
*        |      
*        +----> reportHeartbeat@HeartbeatMonitor   
*        |     //reportHeartbeat : 记录发起请求的这个时间点，然后resetHeartbeatTimeout
*        |      
*        +----> resetHeartbeatTimeout@HeartbeatMonitor
*        |     // 如果Monitor状态依然是RUNNING，则取消之前设置的ScheduledFuture。
*        |     // 重新创建一个ScheduleFuture。因为如果不取消，则之前那个ScheduleFuture运行时
*        |     // 会调用HeartbeatMonitorImpl.run函数，run直接compareAndSet后，通知目标函数
*        |     // 目前已经超时，即调用heartbeatListener.notifyHeartbeatTimeout。
*        |     // 这里代表 JM 状态正常。   
*        |      
*        +---->  heartbeatListener.reportPayload
*        |     // 把Target节点的最新的heartbeatPayload通知给heartbeatListener。 
*        |     // heartbeatListerner是外部传入的，它根据所拥有的节点的心跳记录做监听管理。
*        |        
*        +---->  heartbeatTarget.receiveHeartbeat(getOwnResourceID(), heartbeatListener.retrievePayload(requestOrigin));
*        |      
*        |     
*        +---->  retrievePayload@ResourceManagerHeartbeatListener in TM   
*        |       // heartbeatTarget.receiveHeartbeat参数调用的
*        |     
*        +---->  return new TaskExecutorHeartbeatPayload
*        |   
*        |     
*        +---->  receiveHeartbeat in TM 
*        |       // 回到 heartbeatTarget.receiveHeartbeat，这就是TM生成Target的时候的自定义函数
*        |       // 就是响应一个心跳消息回给RM
*        |     
*        +----> resourceManagerGateway.heartbeatFromTaskManager  
*        |     // 会通过gateway RPC 调用到 ResourcManager
*        |      

* ~~~~~~~~ 这里是 Akka RPC
 
* 3. Run in Resouce Manager
* 现在程序回到了RM, 主要是 1.重置RM的Monitor线程；2. 上报收到TaskExecutor的负载信息  
*
*    heartbeatFromTaskManager in RM
*        |   
*        |   
*        +---->  taskManagerHeartbeatManager.receiveHeartbeat 
*        |       // 这是个Sender HM
*        |  
*        +---->  HeartbeatManagerImpl.receiveHeartbeat  
*        |  
*        |     
*        +---->  HeartbeatManagerImpl.reportHeartbeat(heartbeatOrigin);
*        | 
*        |    
*        +---->  heartbeatMonitor.reportHeartbeat(); 
*        |      // 这里就是重置RM 这里对应的Monitor。在reportHeartbeat重置 JM monitor线程的触发，即cancelTimeout取消注册时候的超时定时任务，并且注册下一个超时检测futureTimeout；这代表TM正常执行。
*        |     
*        +----> heartbeatListener.reportPayload    
*        |      //把Target节点的最新的heartbeatPayload通知给 TaskManagerHeartbeatListener。heartbeatListerner是外部传入的，它根据所拥有的节点的心跳记录做监听管理。 
*        |      
*        +----> slotManager.reportSlotStatus(instanceId, payload.getSlotReport());
*        |     // TaskManagerHeartbeatListener中调用，上报收到TaskExecutor的负载信息
*        |        

下面是具体文字描述。

5.4.1 ResourceManager主动发起

5.4.1.1 Sender遍历所有监控的Monitor（Target）

心跳机制是由Sender主动发起的。这里就是 ResourceManager 的HeartbeatManagerSenderImpl中定时schedual调用，这里会遍历所有监控的Monitor（Target），逐一在Target上调用requestHeartbeat。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// HeartbeatManagerSenderImpl中的代码

	@Override
	public void run() {
		if (!stopped) {
			for (HeartbeatMonitor<O> heartbeatMonitor : getHeartbeatTargets().values()) {
        // 这里向被监控对象节点发起一次心跳请求，载荷是heartbeatPayLoad,要求被监控对象回应心跳
				requestHeartbeat(heartbeatMonitor);
			}
			getMainThreadExecutor().schedule(this, heartbeatPeriod, TimeUnit.MILLISECONDS);
		}
	}
}

// 运行时候的变量
this = {HeartbeatManagerSenderImpl@9037} 
 heartbeatPeriod = 10000
 heartbeatTimeoutIntervalMs = 50000
 ownResourceID = {ResourceID@8788} "d349506cae32cadbe99b9f9c49a01c95"
 heartbeatListener = {ResourceManager$TaskManagerHeartbeatListener@8789} 
 mainThreadExecutor = {RpcEndpoint$MainThreadExecutor@8790} 

// 调用栈如下
requestHeartbeat:711, ResourceManager$2 (org.apache.flink.runtime.resourcemanager)
requestHeartbeat:702, ResourceManager$2 (org.apache.flink.runtime.resourcemanager)
requestHeartbeat:92, HeartbeatManagerSenderImpl (org.apache.flink.runtime.heartbeat)
run:81, HeartbeatManagerSenderImpl (org.apache.flink.runtime.heartbeat)
call:511, Executors$RunnableAdapter (java.util.concurrent)
run$$$capture:266, FutureTask (java.util.concurrent)
run:-1, FutureTask (java.util.concurrent)

5.4.1.2 Target进行具体操作

具体监控对象 Target 会调用自定义的requestHeartbeat。

1
2
3
4
5
6
7
8
9
10
11
12
13

HeartbeatManagerSenderImpl

	private void requestHeartbeat(HeartbeatMonitor<O> heartbeatMonitor) {
		O payload = getHeartbeatListener().retrievePayload(heartbeatMonitor.getHeartbeatTargetId());
		final HeartbeatTarget<O> heartbeatTarget = heartbeatMonitor.getHeartbeatTarget();

    // 这里就是具体监控对象
		heartbeatTarget.requestHeartbeat(getOwnResourceID(), payload);
	}

heartbeatTarget = {ResourceManager$2@10688} 
 taskExecutorGateway = {$Proxy42@9459} "org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler@6d0c8334"
 this$0 = {StandaloneResourceManager@9458} 

请注意，每一个Target都是由ResourceManager生成的。ResourceManager之前注册成为Monitor时候就注册了这个HeartbeatTarget。

这个HeartbeatTarget的定义如下，两个函数是：

• receiveHeartbeat ：这个是空，因为RM没有自己的Sender。
• requestHeartbeat ：这个针对TM，就是调用TM的heartbeatFromResourceManager，当然是通过RPC调用。

5.4.1.3 RPC调用

会调用到ResourceManager定义的函数requestHeartbeat，而requestHeartbeat会通过gateway调用到TM，这就是主动对TM发起了心跳请求。

1
2
3
4
5
6
7
8
9
10
11
12

taskManagerHeartbeatManager.monitorTarget(taskExecutorResourceId, new HeartbeatTarget<Void>() {
   @Override
   public void receiveHeartbeat(ResourceID resourceID, Void payload) {
      // the ResourceManager will always send heartbeat requests to the TaskManager
   }

   @Override
   public void requestHeartbeat(ResourceID resourceID, Void payload) {
      //就是调用到这里
      taskExecutorGateway.heartbeatFromResourceManager(resourceID); 
   }
});

5.4.2 RM通过RPC调用TM

通过taskExecutorGateway。心跳程序执行就通过RPC从RM跳跃到了TM。

taskExecutorGateway.heartbeatFromResourceManager 的意义就是：通过RPC调用回到TaskExecutor。这个是在TaskExecutorGateway就定义好的。

1
2

// TaskExecutor RPC gateway interface.
public interface TaskExecutorGateway extends RpcGateway

TaskExecutor实现了TaskExecutorGateway，所以具体在TaskExecutor内部实现了接口函数。

1
2
3
4
5

@Override
public void heartbeatFromResourceManager(ResourceID resourceID) {
    //调用到了这里 ...........
		resourceManagerHeartbeatManager.requestHeartbeat(resourceID, null);
}

TM中，resourceManagerHeartbeatManager 定义如下。

1
2

/** The heartbeat manager for resource manager in the task manager. */
private final HeartbeatManager<Void, TaskExecutorHeartbeatPayload> resourceManagerHeartbeatManager;

所以下面就是执行TM中的Receiver HM。在这个过程中有两个处理步骤：

1. 调用对应HeartbeatMonitor的reportHeartbeat方法，cancelTimeout取消注册时候的超时定时任务，并且注册下一个超时检测futureTimeout；
2. 调用monitorTarget的receiveHeartbeat方法，也就是会通过rpc调用JobMaster的heartbeatFromTaskManager方法返回一些负载信息；

具体是调用 requestHeartbeat@HeartbeatManager。在其中会

• 调用reportHeartbeat@HeartbeatMonitor，记录发起请求的这个时间点，然后resetHeartbeatTimeout。
• 在resetHeartbeatTimeout@HeartbeatMonitor之中，如果Monitor状态依然是RUNNING，则取消之前设置的ScheduledFuture。重新创建一个ScheduleFuture。因为如果不取消，则之前那个ScheduleFuture运行时会调用HeartbeatMonitorImpl.run函数，run直接compareAndSet后，通知目标函数目前已经超时，即调用heartbeatListener.notifyHeartbeatTimeout。
• 调用 heartbeatListener.reportPayload，把Target节点的最新的heartbeatPayload通知给heartbeatListener。
• 调用 heartbeatTarget.receiveHeartbeat(getOwnResourceID(), heartbeatListener.retrievePayload(requestOrigin)); 就是响应一个心跳消息回给RM。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

@Override
public void requestHeartbeat(final ResourceID requestOrigin, I heartbeatPayload) {
	if (!stopped) {
		log.debug("Received heartbeat request from {}.", requestOrigin);

		final HeartbeatTarget<O> heartbeatTarget = reportHeartbeat(requestOrigin);

		if (heartbeatTarget != null) {
			if (heartbeatPayload != null) {
				heartbeatListener.reportPayload(requestOrigin, heartbeatPayload);
			}

			heartbeatTarget.receiveHeartbeat(getOwnResourceID(), heartbeatListener.retrievePayload(requestOrigin));
		}
	}
}

最后会通过resourceManagerGateway.heartbeatFromTaskManager 调用到 ResourcManager。

5.4.3 TM 通过RPC回到 RM

JobMaster在接收到rpc请求后调用其heartbeatFromTaskManager方法，会调用taskManagerHeartbeatManager的receiveHeartbeat方法，在这个过程中同样有两个处理步骤：

1. 调用对应HeartbeatMonitor的reportHeartbeat方法，cancelTimeout取消注册时候的超时定时任务，并且注册下一个超时检测futureTimeout；
2. 调用TaskManagerHeartbeatListener的reportPayload方法，上报收到TaskExecutor的负载信息

至此一次完成心跳过程已经完成，会根据heartbeatInterval执行下一次心跳。

5.5 超时处理

5.5.1 TaskManager

首先，在HeartbeatMonitorImpl中，如果超时，会调用Listener。

1
2
3
4
5
6

public void run() {
   // The heartbeat has timed out if we're in state running
   if (state.compareAndSet(State.RUNNING, State.TIMEOUT)) {
      heartbeatListener.notifyHeartbeatTimeout(resourceID);
   }
}

这就来到了ResourceManagerHeartbeatListener，会尝试再次连接RM。

1
2
3
4
5
6
7
8
9
10
11
12
13

private class ResourceManagerHeartbeatListener implements HeartbeatListener<Void, TaskExecutorHeartbeatPayload> {

   @Override
   public void notifyHeartbeatTimeout(final ResourceID resourceId) {
      validateRunsInMainThread();
      // first check whether the timeout is still valid
      if (establishedResourceManagerConnection != null && establishedResourceManagerConnection.getResourceManagerResourceId().equals(resourceId)) {
         reconnectToResourceManager(new TaskManagerException(
            String.format("The heartbeat of ResourceManager with id %s timed out.", resourceId)));
      } else {
         .....
      }
   }

5.5.2 ResourceManager

RM就直接简单粗暴，关闭连接。

1
2
3
4
5
6
7
8
9
10

private class TaskManagerHeartbeatListener implements HeartbeatListener<TaskExecutorHeartbeatPayload, Void> {

   @Override
   public void notifyHeartbeatTimeout(final ResourceID resourceID) {
      validateRunsInMainThread();
      closeTaskManagerConnection(
         resourceID,
         new TimeoutException("The heartbeat of TaskManager with id " + resourceID + "  timed out."));
   }
}  

0x06 解决问题

心跳机制我们讲解完了，但是我们最初提到的异常应该如何解决呢？在程序最开始生成环境变量时候，通过设置环境变量的配置即可搞定：

1
2
3

Configuration conf = new Configuration();
conf.setString("heartbeat.timeout", "18000000");
final LocalEnvironment env = ExecutionEnvironment.createLocalEnvironment(conf);

0x07 参考

[flink-001]flink的心跳机制

Flink中心跳机制

flink1.8 心跳服务

你有必要了解一下Flink底层RPC使用的框架和原理

flink RPC(akka)

弄清Flink1.8的远程过程调用(RPC)

Apache Flink源码解析 （七）Flink RPC的底层实现

flink源码阅读第一篇—入口

flink-on-yarn 基础架构和启动流程

在Flink SQL中， 元数据的管理分为三层: catalog-> database-> table，
我们知道Flink SQL是依托calcite框架来进行SQL执行树生产，校验，优化等等， 所以本文讲介绍FlinkSQL是如何来结合Calcite来进行元数据管理的.

calcite开放的接口

1
2
3
4
5
6
7

public interface Schema {
    Table getTable(String name);

    Schema getSubSchema(String name);

    ....
}

如接口所示， Schema接口，可以通过table名来获得一张表， 可以通过schema名来获得一个子schema.

1
2
3
4

public interface Table {
    RelDataType getRowType(RelDataTypeFactory typeFactory);
    ....
}

看Table的接口， 主要就是返回table的RelDataType.

Flink的相关实现

接下来，我们来看下Flink是如何实现这些接口的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

public class CatalogManagerCalciteSchema extends FlinkSchema {
    @Override
    public Schema getSubSchema(String schemaName) {
        if (catalogManager.schemaExists(name)) {
            return new CatalogCalciteSchema(name, catalogManager, isStreamingMode);
        } else {
            return null;
        }
    }
}
public class CatalogCalciteSchema extends FlinkSchema {
    @Override
    public Schema getSubSchema(String schemaName) {
        if (catalogManager.schemaExists(catalogName, schemaName)) {
            return new DatabasecalciteSchema(schemaName, catalogNmae, catalogManager, isStreamingMode);
        }
    }
}
public class DatabaseCalciteSchema extends FlinkSchema {
    private final String databaseName;
    private final String catalogName;
    private final CatalogManager catalogManager;

    @Override
    public Table getTable(String tableName) {
        ObjectIdentifier identifier = ObjectIdentifier.of(catalogName, databaseName, tableName);
        return catalogManager.getTable(identifier)
            .map(result -> {
                CatalogBaseTable table = result.getTable();
                FlinkStatistic statistic = getStatistic(result.isTemporary(), table, identifier);
                return new CatalogSchemaTable(identifier,
                    table,
                    statistic,
                    catalogManager.getCatalog(catalogName)
                        .flatMap(Catalog::getTableFactory)
                        .orElse(null),
                    isStreamingMode,
                    result.isTemporary());
            })
            .orElse(null);
    }

    @Override
    public Schema getSubSchema(String name) {
        return null;
    }
}

很容易发现，CatalogSchema返回DatabaseSchema， DatabaseSchema返回Table,
这样就容易理解，Flink的三层结构是怎样的了. 同时， 具体的元数据实际上都是在catalogManager中。

DatabaseSchema中返回的Table类型为CatalogSchemaTable，我们来看下具体的结结构是怎样的，
上文中也提到了，Table接口主为getRowType函数， 用于返回某个table的type信息。
TableSchema是Flink内部用于保存各个字段的类型信息的类， 通过相关的转化函数，转换为calcite的type类型.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

public class CatalogSchemaTable extends AbstractTable implements TemporalTable {

    private final ObjectIdentifier tableIdentifier;
    private final CatalogBaseTable catalogBaseTable;
    private final FlinkStatistic statistic;
    private final boolean isStreamingMode;
    private final boolean isTemporary;
    ...
    private static RelDataType getRowType(RelDataTypeFactory typeFactory,
            CatalogBaseTable catalogBaseTable,
            boolean isStreamingMode) {
        final FlinkTypeFactory flinkTypeFactory = (FlinkTypeFactory) typeFactory;
        TableSchema tableSchema = catalogBaseTable.getSchema();
        final DataType[] fieldDataTypes = tableSchema.getFieldDataTypes();
        if (!isStreamingMode
            && catalogBaseTable instanceof ConnectorCatalogTable
            && ((ConnectorCatalogTable) catalogBaseTable).getTableSource().isPresent()) {
            // If the table source is bounded, materialize the time attributes to normal TIMESTAMP type.
            // Now for ConnectorCatalogTable, there is no way to
            // deduce if it is bounded in the table environment, so the data types in TableSchema
            // always patched with TimeAttribute.
            // See ConnectorCatalogTable#calculateSourceSchema
            // for details.

            // Remove the patched time attributes type to let the TableSourceTable handle it.
            // We should remove this logic if the isBatch flag in ConnectorCatalogTable is fixed.
            // TODO: Fix FLINK-14844.
            for (int i = 0; i < fieldDataTypes.length; i++) {
                LogicalType lt = fieldDataTypes[i].getLogicalType();
                if (lt instanceof TimestampType
                    && (((TimestampType) lt).getKind() == TimestampKind.PROCTIME
                    || ((TimestampType) lt).getKind() == TimestampKind.ROWTIME)) {
                    int precision = ((TimestampType) lt).getPrecision();
                    fieldDataTypes[i] = DataTypes.TIMESTAMP(precision);
                }
            }
        }
        return TableSourceUtil.getSourceRowType(flinkTypeFactory,
            tableSchema,
            scala.Option.empty(),
            isStreamingMode);
    }
}

CatalogBaseTable接口定义如下， Flink的Table的参数(schema参数,connector参数)都可以最终表示为一个map.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

public interface CatalogBaseTable {
    /**
     * Get the properties of the table.
     *
     * @return property map of the table/view
     */
    Map<String, String> getProperties();

    /**
     * Get the schema of the table.
     *
     * @return schema of the table/view.
     */
    TableSchema getSchema();

    /**
     * Get comment of the table or view.
     *
     * @return comment of the table/view.
     */
    String getComment();

    /**
     * Get a deep copy of the CatalogBaseTable instance.
     *
     * @return a copy of the CatalogBaseTable instance
     */
    CatalogBaseTable copy();

    /**
     * Get a brief description of the table or view.
     *
     * @return an optional short description of the table/view
     */
    Optional<String> getDescription();

    /**
     * Get a detailed description of the table or view.
     *
     * @return an optional long description of the table/view
     */
    Optional<String> getDetailedDescription();
}

FlinkSchema的使用

上面都是的相关接口都是Flink用于适配calcite框架元数据的相关实现。
那么这些类具体是在哪里调用的？ 已经什么时候会被调用到？
calcite中的schema，主要是在validate过程中， 获得对应table的字段信息， 对应的function的返回值信息，
确保SQL的字段名， 字段类型是正确的.
类的依赖关系为:
validator —> schemaReader —> schema

FlinkPlannerImpl.scala中

1
2
3
4
5
6
7
8
9
10

private def createSqlValidator(catalogReader: CatalogReader) = {
  val validator = new FlinkCalciteSqlValidator(
    operatorTable,
    catalogReader,
    typeFactory)
  validator.setIdentifierExpansion(true)
  // Disable implicit type coercion for now.
  validator.setEnableTypeCoercion(false)
  validator
}

PlanningConfigurationBuilder.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

private CatalogReader createCatalogReader(
        boolean lenientCaseSensitivity,
        String currentCatalog,
        String currentDatabase) {
    SqlParser.Config sqlParserConfig = getSqlParserConfig();
    final boolean caseSensitive;
    if (lenientCaseSensitivity) {
        caseSensitive = false;
    } else {
        caseSensitive = sqlParserConfig.caseSensitive();
    }

    SqlParser.Config parserConfig = SqlParser.configBuilder(sqlParserConfig)
        .setCaseSensitive(caseSensitive)
        .build();

    return new CatalogReader(
        rootSchema,
        asList(
            asList(currentCatalog, currentDatabase),
            singletonList(currentCatalog)
        ),
        typeFactory,
        CalciteConfig.connectionConfig(parserConfig));
}

综上所诉， 我们就知道了Flink是如何来利用calcite的schema来管理Flink的table信息的.

Flink SQL Join

Join的几种形式，与实现原理

双流Join

Left join

TTL

从 Flink 1.6 版本开始，社区引入了状态 TTL（Time-To-Live）特性。在通过Flink SQL 实现流处理时，开发者可以为作业 SQL 设置TTL，实现过期状态的自动清理，从而防止作业状态无限膨胀

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

SELECT
    t_date,
    COUNT ( DISTINCT user_id ) AS cnt_login, -- 今日登录用户数
    COUNT ( DISTINCT CASE WHEN t_date = t_debut THEN user_id END ) AS cnt_new -- 今日新用户数
FROM
  (
    -- 计算每个用户有史以来的最小登录时间
    SELECT
       t_date,
       user_id,
       MIN (t_date) OVER (
             PARTITION BY user_id
             ORDER BY proctime
             ROWS BETWEEN 1 PRECEDING AND CURRENT ROW
       ) AS t_debut
    FROM Login
) AS t
GROUP BY t_date

Query Configuration 查询配置

https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html

Flink Table API 和SQL接口提供参数来调整连续查询的准确性和资源消耗。参数通过 QueryConfig 对象指定。QueryConfig 可以从 TableEnvironment 获得。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// 获取 query configuration
val qConfig: StreamQueryConfig = tableEnv.queryConfig

// 设置查询参数
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24))

// 定义查询和 TableSink
val result: Table = ???
val sink: TableSink[Row] = ???

// TableSink 发送结果表时传递查询参数
result.writeToSink(sink, qConfig)

// 转换为 DataStream 时传递查询参数
val stream: DataStream[Row] = result.toAppendStream[Row](qConfig)

空闲状态保持时间（Idle State Retention Time）参数定义一个键的状态在一次更新之后保存多久后删除。

通过删除键的状态，连续查询会完全忘记它之前已经看过这个键。如果删除的键再次出现，则被视为具有相应键的第一个记录。对于前面的查询示例，这意味着 sessionId 的计数从0开始。

配置空闲状态保存时间有两个参数：

minimum idle state retention time，定义非活动键的状态在删除前至少保持多少时间。
maximum idle state retention time，定义非活动键的状态在删除前最多保持多少时间。

对于前面的查询示例：

1
2
3
4

val qConfig: StreamQueryConfig = ???

// 设置 idle state retention time: min = 12 hours, max = 24 hours
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24))

清理状态需要额外的记录，对于 minTime 和 maxTime 较大差异的情况成本更低，因此 minTime 和 maxTime 直接必须至少相差5分钟。

维表Join

启用AsyncIO

时间表Join: Temporal Table Join

1
2
3
4
5
6

SELECT
o.amout, o.currency, r.rate, o.amount * r.rate
FROM
Orders AS o
JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r
ON r.currency = o.currency

Flink SQL 代码生成

从”UDF不应有状态” 切入来剖析Flink SQL代码生成

核心类: org.apache.flink.table.runtime.CRowProcessRunner

打印出codegen语句:

1

log4j.logger.org.apache.flink.runtime.CRowProcessRunner=DEBUG, file

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

alter table
  b_cdg_cft_t_trade_user_fund_merge_ods_t_test_fund_Facc_time 
set
  (
    'consumerGroup' = 't_cdg_cft_b_cdg_cft_t_trade_user_fund_merge_ods_cg_rtdw_prod_003',
    'inDebugMode' = 'true',
    'consume-from-max' = 'true'
  );

-- 自定义控制台输出下游数据表，方便查看运行结果
alter table
  console
set
  (
    -- 同上，console的schema定义的是下游表的元数据
    'schema' = 'Facc_time:varchar,Flistid:varchar,Fsg_sh_type:bigint,Ftotal_fee:bigint,Funion_id:bigint,Fsg_sh_name:varchar,Ftrade_id:varchar'
  );
-- 结果数据写入console，可以在应用运维->TM日志中查看结果输出
insert into
  console
  select
    Facc_time,
    Flistid,
    Fsg_sh_type,
    Ftotal_fee,
    Funion_id,
    Fsg_sh_name,
    Ftrade_id
from
    b_cdg_cft_t_trade_user_fund_merge_ods_t_test_fund_Facc_time

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355

2021-10-27 15:06:43,520 DEBUG [Task-46f8730428df9ecd6d7318a02bdc405e(0/1)] org.apache.flink.table.runtime.conversion.CRowToJavaTupleMapRunner  - Compiling MapFunction: DataStreamSinkConversion$14 

 Code:

public class DataStreamSinkConversion$14 extends org.apache.flink.api.common.functions.RichMapFunction {


  final org.apache.flink.types.Row out =
      new org.apache.flink.types.Row(7);

  private org.apache.flink.types.Row in1;


  public DataStreamSinkConversion$14() throws Exception {


  }




  @Override
  public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {


  }

  @Override
  public Object map(Object _in1) throws Exception {
    in1 = (org.apache.flink.types.Row) _in1;

    boolean isNull$13 = (java.lang.String) in1.getField(6) == null;
    java.lang.String result$12;
    if (isNull$13) {
      result$12 = "";
    }
    else {
      result$12 = (java.lang.String) (java.lang.String) in1.getField(6);
    }


    boolean isNull$11 = (java.lang.String) in1.getField(5) == null;
    java.lang.String result$10;
    if (isNull$11) {
      result$10 = "";
    }
    else {
      result$10 = (java.lang.String) (java.lang.String) in1.getField(5);
    }


    boolean isNull$5 = (java.lang.Long) in1.getField(2) == null;
    long result$4;
    if (isNull$5) {
      result$4 = -1L;
    }
    else {
      result$4 = (java.lang.Long) in1.getField(2);
    }


    boolean isNull$1 = (java.lang.String) in1.getField(0) == null;
    java.lang.String result$0;
    if (isNull$1) {
      result$0 = "";
    }
    else {
      result$0 = (java.lang.String) (java.lang.String) in1.getField(0);
    }


    boolean isNull$9 = (java.lang.Long) in1.getField(4) == null;
    long result$8;
    if (isNull$9) {
      result$8 = -1L;
    }
    else {
      result$8 = (java.lang.Long) in1.getField(4);
    }


    boolean isNull$3 = (java.lang.String) in1.getField(1) == null;
    java.lang.String result$2;
    if (isNull$3) {
      result$2 = "";
    }
    else {
      result$2 = (java.lang.String) (java.lang.String) in1.getField(1);
    }


    boolean isNull$7 = (java.lang.Long) in1.getField(3) == null;
    long result$6;
    if (isNull$7) {
      result$6 = -1L;
    }
    else {
      result$6 = (java.lang.Long) in1.getField(3);
    }







    if (isNull$1) {
      out.setField(0, null);
    }
    else {
      out.setField(0, result$0);
    }



    if (isNull$3) {
      out.setField(1, null);
    }
    else {
      out.setField(1, result$2);
    }



    if (isNull$5) {
      out.setField(2, null);
    }
    else {
      out.setField(2, result$4);
    }



    if (isNull$7) {
      out.setField(3, null);
    }
    else {
      out.setField(3, result$6);
    }



    if (isNull$9) {
      out.setField(4, null);
    }
    else {
      out.setField(4, result$8);
    }



    if (isNull$11) {
      out.setField(5, null);
    }
    else {
      out.setField(5, result$10);
    }



    if (isNull$13) {
      out.setField(6, null);
    }
    else {
      out.setField(6, result$12);
    }

    return out;

  }

  @Override
  public void close() throws Exception {


  }
}

2021-10-27 15:06:43,520 DEBUG [Task-f24f04d8dcfb922ebc646161f8d38c37(0/1)] org.apache.flink.table.runtime.CRowMapRunner                  - Compiling MapFunction: DataStreamSinkConversion$29 

 Code:

public class DataStreamSinkConversion$29 extends org.apache.flink.api.common.functions.RichMapFunction {


  final org.apache.flink.types.Row out =
      new org.apache.flink.types.Row(7);

  private org.apache.flink.types.Row in1;


  public DataStreamSinkConversion$29() throws Exception {


  }




  @Override
  public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {


  }

  @Override
  public Object map(Object _in1) throws Exception {
    in1 = (org.apache.flink.types.Row) _in1;

    boolean isNull$28 = (java.lang.String) in1.getField(6) == null;
    java.lang.String result$27;
    if (isNull$28) {
      result$27 = "";
    }
    else {
      result$27 = (java.lang.String) (java.lang.String) in1.getField(6);
    }


    boolean isNull$26 = (java.lang.String) in1.getField(5) == null;
    java.lang.String result$25;
    if (isNull$26) {
      result$25 = "";
    }
    else {
      result$25 = (java.lang.String) (java.lang.String) in1.getField(5);
    }


    boolean isNull$20 = (java.lang.Long) in1.getField(2) == null;
    long result$19;
    if (isNull$20) {
      result$19 = -1L;
    }
    else {
      result$19 = (java.lang.Long) in1.getField(2);
    }


    boolean isNull$16 = (java.lang.String) in1.getField(0) == null;
    java.lang.String result$15;
    if (isNull$16) {
      result$15 = "";
    }
    else {
      result$15 = (java.lang.String) (java.lang.String) in1.getField(0);
    }


    boolean isNull$24 = (java.lang.Long) in1.getField(4) == null;
    long result$23;
    if (isNull$24) {
      result$23 = -1L;
    }
    else {
      result$23 = (java.lang.Long) in1.getField(4);
    }


    boolean isNull$18 = (java.lang.String) in1.getField(1) == null;
    java.lang.String result$17;
    if (isNull$18) {
      result$17 = "";
    }
    else {
      result$17 = (java.lang.String) (java.lang.String) in1.getField(1);
    }


    boolean isNull$22 = (java.lang.Long) in1.getField(3) == null;
    long result$21;
    if (isNull$22) {
      result$21 = -1L;
    }
    else {
      result$21 = (java.lang.Long) in1.getField(3);
    }







    if (isNull$16) {
      out.setField(0, null);
    }
    else {
      out.setField(0, result$15);
    }



    if (isNull$18) {
      out.setField(1, null);
    }
    else {
      out.setField(1, result$17);
    }



    if (isNull$20) {
      out.setField(2, null);
    }
    else {
      out.setField(2, result$19);
    }



    if (isNull$22) {
      out.setField(3, null);
    }
    else {
      out.setField(3, result$21);
    }



    if (isNull$24) {
      out.setField(4, null);
    }
    else {
      out.setField(4, result$23);
    }



    if (isNull$26) {
      out.setField(5, null);
    }
    else {
      out.setField(5, result$25);
    }



    if (isNull$28) {
      out.setField(6, null);
    }
    else {
      out.setField(6, result$27);
    }

    return out;

  }

  @Override
  public void close() throws Exception {


  }
}