稳定复现 checkpoint 正常生成，在web ui和hdfs目录里边都可以确认。 我们jobmanager没有做ha，不知道是否是这个原因导致的？ 日志里边能看到是从指定的-s恢复的，没有指定-s的时候，重启的时候也并没有使用最新的checkpoint文件。 目前这个问题困扰了我很久，也没有一个好的思路，下一步先把ha搞起来再试试。 >> org.apache.flink.configuration.GlobalConfiguration [] - Loading >> configuration property: execution.savepoint.path, >> hdfs:///user/flink/checkpoints/default/f9b85edbc6ca779b6e60414f3e3964f2/chk-100
在 2021-05-28 18:15:38，"刘建刚" <[email protected]> 写道： >这种情况是不符合预期的。请问通过以下步骤可以稳定复现吗？ >1、从savepoint恢复； >2、作业开始定期做savepoint； >3、作业failover。 >如果是的话，可能需要排查下checkpoint 文件是否存在，zookeeper上是否更新。 >如果还是有问题，需要通过日志来排查了。 > >董建 <[email protected]> 于2021年5月28日周五 下午5:37写道： > >> 我遇到的问题现象是这样的 >> >> >> >> >> 1、flink版本flink-1.12.2，启动命令如下，指定-s是因为job有做过cancel，这里重启。 >> >> >> >> >> flink run -d -s >> hdfs:///user/flink/checkpoints/default/f9b85edbc6ca779b6e60414f3e3964f2/chk-100 >> -t yarn-per-job -m yarn-cluser -D yarn.application.name=xxxx >> /tmp/flink-1.0-SNAPSHOT.jar -c com.test.myStream --profile prod >> >> >> >> >> 2、flink-conf.xml >> >> >> >> >> state.checkpoints.dir: hdfs:///user/flink/checkpoints/default >> >> >> >> >> 3、代码checkpoint设置 >> >> >> >> >> StreamExecutionEnvironment env = >> StreamExecutionEnvironment.getExecutionEnvironment(); >> >> >> >> >> env.setRestartStrategy(RestartStrategies.fixedDelayRestart(100, >> 10)); >> >> >> >> >> CheckpointConfig checkpointConfig = env.getCheckpointConfig(); >> >> >> >> >> >> >> checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); >> >> >> >> >> env.enableCheckpointing(1 * 60 * 1000); >> >> >> >> >> >> checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); >> >> >> >> >> checkpointConfig.setTolerableCheckpointFailureNumber(100); >> >> >> >> >> checkpointConfig.setCheckpointTimeout(60 * 1000); >> >> >> >> >> checkpointConfig.setMaxConcurrentCheckpoints(1); >> >> >> >> >> 4、问题现象 >> >> >> >> >> a)运维同事切换yarn >> resourc-manager，我的flink任务也会重启（重启后application-id和job-id并没有改变），但是jobmanager和taskmanager更换了机器 >> >> >> >> >> >> b)我的flink任务每隔1分钟做一个checkpoint，假如任务在上次重启后又运行了一段时间，checkpoint已经到了chk-200 >> >> >> >> >> c)集群的同事再次切换resource-manager,这个时候预期是flink任务自动从chk-200 >> restore，从日志中看还是从chk-100 restore的。 >> >> >> >> >> d)任务大致逻辑是通过flink-mysql-cdc消费binlog，DebeziumSourceFunction<String> >> sourceMilApplysLogStream = MySQLSource.<String>builder() >> >> >> >> >> 重启导致了再次从chk-100的binlog pos开始消费，即了chk-100~chk-200之间的binlog重复消费 >> >> >> >> >> e)日志中有打印如下，难道是因为上次我启动flink任务的时候用了-s，所以不会去自动找最新的checkpoint重启吗？ >> >> >> >> >> 2021-05-24 16:49:50,398 INFO >> org.apache.flink.configuration.GlobalConfiguration [] - Loading >> configuration property: execution.savepoint.path, >> hdfs:///user/flink/checkpoints/default/f9b85edbc6ca779b6e60414f3e3964f2/chk-100 >> >> >> >> 预期是任务重启后自动从最新的checkpoint开始继续消费，即从chk-200开始消费 >> >> >> >> >> 现在的问题是任务重启后总是从flink -s指定的checkpoint恢复的。
