问题一：flink job 跑一段时间 watermark 不推进的问题
flink job 跑一段时间 watermark 不推进，任务没挂，source 是 kafka ，kafka 各个partition 均有数据， flink job statue backend 为 memory 。有debug 的姿势推荐吗？ 看过 CPU GC 等指标，看不出来有异常。
参考回答：
Flink
metrics里有一项是task相关的指标currentWatermark，从中可以知道subtask_index,task_name,watermark三项信息，应该能帮助排查watermark的推进情况。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/370126
问题二：flink1.11 Application 模式下启动失败问题
我把作业提交模式从 yarn-cluster 换成 application 模式，启动失败，报两个错误： 1、java.lang.ClassNotFoundException: org.apache.hadoop.yarn.api.records.ResourceInformation 2、cannot assign instance of org.apache.commons.collections.map.LinkedMap to field org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.pendingOffsetsToCommit of type org.apache.commons.collections.map.LinkedMap in instance of com.tydic.tysc.core.flink.cal.v3.core.connector.kafka.source.KafkaTableSource$CustomerFlinkKafkaConsumer 在 yarn-cluster 下正常运行，请各位帮忙看下。
参考回答：
Application和Perjob模式本质上的区别是用户main运行的位置不一样，所以会导致JM启动的classpath
也不太一样的
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/370125
问题三：flink sqlUpdate,如何获取里面的字段，字段类型，with 等3个属性
我用flink 注册一张表： CREATE TABLE dim_mysql ( id int, -- type varchar -- ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3390/test', 'table-name' = 'flink_test', 'driver' = 'com.mysql.cj.jdbc.Driver', 'username' = '', 'password' = '', 'lookup.cache.max-rows' = '5000', 'lookup.cache.ttl' = '1s', 'lookup.max-retries' = '3' ) 有没有通过 tableEnv 去获取，字段[id,type] 类型[INTEGER,VARCHAR] 以及属性，map 这种。 我看阿里官方有blink 支持自定义sink: publicabstractclassCustomSinkBaseimplementsSerializable{ protectedMap userParamsMap;// 您在sql with语句中定义的键值对，但所有的键均为小写 protectedSet primaryKeys;// 您定义的主键字段名 protectedList headerFields;// 标记为header的字段列表 protectedRowTypeInfo rowTypeInfo;// 字段类型和名称 核心需求是：获取定义的表的所有属性，自己实现自己的功能，包括 join sink 等各种逻辑
参考回答：
tableEnv 中 可以通过
tableEvn.from(xx).getSchema() 拿到该表的schema信息，但是没法拿到对应的properties。
如果要拿到properties，可以通过catalog的接口得到 [1]。
如果要自定义实现source/sink，可以参考 [2]
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/catalogs.html
[2]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/sourceSinks.html
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/370124
问题四：flink 1.11 executeSql查询kafka表print没有输出
本人当前使用flink版本1.11.0，但是在执行executeSql后，print时没有在console打印出结果（查看kafka是一直有数据产生的）, sql如下:
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
Catalog catalog = new HiveCatalog("x", "default", "D:\conf", "1.1.0");
tEnv.registerCatalog("x", catalog);
TableResult execute = tEnv.executeSql("select * from x.ods.ods_binlog_test_trip_create_t_order_1");
execute.print();
建表语句如下：
CREATE TABLE x.ods.ods_binlog_test_trip_create_t_order_1 (
_type STRING,
order_no STRING,
order_time STRING,
dt as TO_TIMESTAMP(order_time),
proctime as PROCTIME(),
WATERMARK FOR dt AS dt - INTERVAL '5' SECOND
) WITH (
'connector.type' = 'kafka',
'connector.properties.bootstrap.servers' = '***',
'connector.properties.zookeeper.connect' = '****',
'connector.version' = 'universal',
'format.type' = 'json',
'connector.properties.group.id' = 'testGroup',
'connector.startup-mode' = 'group-offsets',
'connector.topic' = 'test'
)
参考回答：
1.11的 TableResult.collect() 和 TableResult.print() 方法在流模式下， 都是exactly once语义，需要配置checkpoint才能得到结果。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/370123
问题五：flink1.9写权限认证的es6
请问flink如何将数据写入到权限认证的es集群哪，没找到配置用户名密码的地方，哪位大佬帮忙解答一下
参考回答：
SQL添加认证的逻辑已经在FLINK-18361[1] 中完成了，1.12版本会支持这个功能
[1] https://issues.apache.org/jira/browse/FLINK-18361
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/372088
