flink 1.15 在写flink sql 程序，不能从检查点恢复数据，希望大佬帮忙看看

flink 1.15 在写flink sql 程序on yarn的时候，设置了检查点，手动取消任务，然后从检查点启动程序，好像不能从检查点读取到状态。希望大佬帮忙看看，困扰好几天了。

说明: flink版本:flink 1.15 运行环境:flink on yarn的应用模式 代码模式:在java中写的flink sql

逻辑描述: 1.建一张读取kafka数据的来源表 2.建一张用于sink的mysql表 3.汇总kafka来源表的数据，然后插入sink表（2个字段：编号名称，编号的数量）

问题描述: 1.启动程序 2.往kafka写入2条相同编号(00001)的数据 3.程序会统计这个编号的数据量写入sink表，结果为:00001,2 4.手动停止程序 5.从检查点启动程序 6.往kafka再写入一条编号(00001)的数据 7.程序将结果写入sink表，结果为：00001,1 问题来了，这里统计的编号00001的数量为1，正常来说从检查点恢复，统计的结果应该是数量为3才对。

代码: 1.读取kafka数据代码: String createSourceKakfaDDL ="create table source_kafka_table_report_info( \n" + " msg_id STRING,\n" + " report_no STRING,\n" + " min_bms_soc decimal(11,2),\n" + " max_bms_soc decimal(11,2),\n" + " min_ev_mileage decimal(11,2),\n" + " max_ev_mileage decimal(11,2),\n" + " min_fuel_mileage decimal(11,2),\n" + " max_fuel_mileage decimal(11,2),\n" + " min_fuel_remaining decimal(11,2),\n" + " max_fuel_remaining decimal(11,2),\n" + " BCM_TyrePressure_LF decimal(11,2),\n" + " BCM_TyrePressure_RF decimal(11,2),\n" + " BCM_TyrePressure_LR decimal(11,2),\n" + " BCM_TyrePressure_RR decimal(11,2),\n" + " file_path STRING,\n" + " collect_time TIMESTAMP(3),\n" + " vin STRING,\n" + " min_mileage decimal(11,2),\n" + " max_mileage decimal(11,2),\n" + " proctime AS PROCTIME(),\n" + //" ts AS TO_TIMESTAMP(collect_time),"+ //" ts AS TO_TIMESTAMP(FROM_UNIXTIME(collect_time/1000))\n" + " WATERMARK FOR collect_time AS collect_time - INTERVAL '1' SECOND" + " ) WITH (\n" + " 'connector' = 'kafka',\n" + " 'topic' = 'topic-atp-vat-report-info',\n" + " 'properties.bootstrap.servers' = '"+bootstrap_servers+"',\n" + //" 'scan.startup.mode' = 'earliest-offset',\n" + " 'scan.startup.mode' = 'latest-offset'," + " 'properties.group.id' = 'test_group03',\n" +
" 'format' = 'json' \n" + " )"; tableEnv.executeSql(createSourceKakfaDDL);

2.设置检查点代码: env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setCheckpointStorage("hdfs://nameservice:8020/user/hdfs/flink/test01_status"); env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));

3.创建输出表代码: String createOutDDL ="create table eol_test_report_xt_test(" + " report_no string," + " report_count decimal(11,2)," + " primary key (report_no) not enforced" + ") WITH (" + "'connector' = 'jdbc'," + " 'url' ='"+mysql_url+"'," + " 'driver' = 'com.mysql.jdbc.Driver', " + " 'table-name' = 'eol_test_report_xt_test', " + " 'username' = '"+mysql_username+"', " + " 'password' = '"+mysql_password+"', " + " 'sink.buffer-flush.max-rows' = '10000', " + " 'sink.buffer-flush.interval' = '5s', " + " 'sink.max-retries' = '3', " + " 'sink.parallelism' = '2' " + ")";

    tableEnv.executeSql(createOutDDL);

4.插入目标表代码: String insertSinkDDL ="insert into eol_test_report_xt_test"+ " ("+ " report_no," + " report_count" + " )" + " select " + " t01.report_no, "+ " count(*) as report_count" + " from source_kafka_table_report_info t01" + " group by t01.report_no "+ ""; tableEnv.executeSql(insertSinkDDL);

5.从检查点恢复程序命令: bin/flink run-application -t yarn-application -d -Dtaskmanager.memory.process.size=2048M -Dtaskmanager.numberOfTaskSlots=3 -p 1 -c com.dfsk.atp.vat.faultdetection.Test01 ./projects/atp-vat-flink-data-1.0-SNAPSHOT.jar --mode dev -s hdfs://nameservice:8020/user/hdfs/flink/test01_status/ccbab43446872f8b69e165b54b191cfb/chk-10/_metadata