flink checkpoint失败问题

短命女 2022-11-13 05:27 274阅读 0赞

1、flink ckpt失败的问题

[https://mp.weixin.qq.com/s/5o8rbz07Je17ZHnRz97-Rg][https_mp.weixin.qq.com_s_5o8rbz07Je17ZHnRz97-Rg]

2、先去jobmanager去查询相关日志，说明flinksql有点问题

java.lang.Exception: Could not complete snapshot 1 for operator GroupAggregate(groupBy: (user_id), select: (user_id, KeyByFunction(properties) AS properties, KeyByFunction(event) AS event, KeyByFunction(current_page) AS current_page, KeyByFunction(block_type) AS block_type, KeyByFunction(content_tag_id) AS content_tag_id, KeyByFunction(author_id) AS author_id, KeyByFunction(category_id_new) AS category_id_new, KeyByFunction(cspu_id) AS cspu_id, KeyByFunction(sex) AS sex, KeyByFunction(sexy) AS sexy, KeyByFunction(cspu_ids) AS cspu_ids, KeyByFunction(content_id) AS content_id, KeyByFunction(content_type) AS content_type, KeyByFunction(time) AS time, KeyByFunction(circle_id) AS circle_id)) -> Calc(select: (properties, user_id, event, current_page, block_type, content_tag_id, author_id, category_id_new, cspu_id, sex, sexy, cspu_ids, content_id, content_type, ((event = 'community_content_click':VARCHAR(65536)) CASE CAST(CAST(((properties MULTI_JSON_VALUE '$.algorithm_channel_Id' MULTI_JSON_VALUE '$.feedback_type' MULTI_JSON_VALUE '$.community_content_i... (11/24).
    	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:422)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1522)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1445)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:1034)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$5(StreamTask.java:880)
    	at org.apache.flink.util.checkpointlock.SynchronizedLockHelper.lockAndRun(SynchronizedLockHelper.java:58)
    	at org.apache.flink.util.checkpointlock.CheckpointLockDelegate.lockAndRun(CheckpointLockDelegate.java:57)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:855)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:815)
    	at org.apache.flink.streaming.runtime.io.BarrierBuffer.notifyCheckpoint(BarrierBuffer.java:428)
    	at org.apache.flink.streaming.runtime.io.BarrierBuffer.processBarrier(BarrierBuffer.java:331)
    	at org.apache.flink.streaming.runtime.io.BarrierBuffer.getNext(BarrierBuffer.java:204)
    	at org.apache.flink.streaming.runtime.io.BarrierBuffer.getNextNonBlocked(BarrierBuffer.java:162)
    	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:248)
    	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:113)
    	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:448)
    	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:758)
    	at java.lang.Thread.run(Thread.java:834)
    Caused by: java.lang.NullPointerException
    	at com.poizon.userportrait.handler.community.CommunityExposureEventHandler.doAttenuation(CommunityExposureEventHandler.java:209)
    	at com.poizon.userportrait.handler.community.CommunityExposureEventHandler.doEvent(CommunityExposureEventHandler.java:113)
    	at com.poizon.userportrait.sink.redis.community.UserPortraitCommunityRedisSink.doBatch(UserPortraitCommunityRedisSink.java:254)
    	at com.poizon.userportrait.sink.redis.community.UserPortraitCommunityRedisSink.sync(UserPortraitCommunityRedisSink.java:347)
    	at com.alibaba.blink.streaming.connector.custom.sink.CustomOutputFormat.sync(CustomOutputFormat.java:89)
    	at com.alibaba.blink.streaming.connectors.common.output.TupleOutputFormatAdapterSink.snapshotState(TupleOutputFormatAdapterSink.java:88)
    	at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:126)
    	at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:99)
    	at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:90)
    	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:400)
    	... 17 more

2021-03-25 22:43:35,146 ERROR [flink-akka.actor.default-dispatcher-4] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 3.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:43:35,174 ERROR [flink-akka.actor.default-dispatcher-4] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 16.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:43:35,257 ERROR [flink-akka.actor.default-dispatcher-4] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:17,483 ERROR [flink-akka.actor.default-dispatcher-81] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 3.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:17,493 ERROR [flink-akka.actor.default-dispatcher-81] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 16.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:17,756 ERROR [flink-akka.actor.default-dispatcher-4] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:17,770 ERROR [flink-akka.actor.default-dispatcher-81] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 2.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:18,197 ERROR [flink-akka.actor.default-dispatcher-4] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:18,339 ERROR [flink-akka.actor.default-dispatcher-81] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:20,494 ERROR [flink-akka.actor.default-dispatcher-4] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 3.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:20,542 ERROR [flink-akka.actor.default-dispatcher-4] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 2.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:20,672 ERROR [flink-akka.actor.default-dispatcher-81] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:20,786 ERROR [flink-akka.actor.default-dispatcher-81] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 16.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:21,078 ERROR [flink-akka.actor.default-dispatcher-81] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:21,253 ERROR [flink-akka.actor.default-dispatcher-108] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:22,772 INFO  [flink-akka.actor.default-dispatcher-101] org.apache.flink.runtime.rest.handler.job.JobVertexDetailsHandler  - job vertex detail get log file from ResourceManager Api.
    2021-03-25 22:45:22,800 INFO  [flink-akka.actor.default-dispatcher-82] org.apache.flink.runtime.rest.handler.job.JobVertexDetailsHandler  - job vertex detail get log file from ResourceManager Api.
    2021-03-25 22:45:24,923 ERROR [flink-akka.actor.default-dispatcher-82] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:24,926 ERROR [flink-akka.actor.default-dispatcher-82] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 16.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:25,144 ERROR [flink-akka.actor.default-dispatcher-82] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 3.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:25,168 ERROR [flink-akka.actor.default-dispatcher-82] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:25,193 ERROR [flink-akka.actor.default-dispatcher-82] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 2.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
    2021-03-25 22:45:25,344 ERROR [flink-akka.actor.default-dispatcher-82] org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler  - Exception occurred in REST handler.
    org.apache.flink.runtime.rest.handler.RestHandlerException: Could not find checkpointing statistics for checkpoint 1657.
    	at org.apache.flink.runtime.rest.handler.job.checkpoints.AbstractCheckpointHandler.handleRequest(AbstractCheckpointHandler.java:85)
    	at org.apache.flink.runtime.rest.handler.job.AbstractExecutionGraphHandler.lambda$handleRequest$0(AbstractExecutionGraphHandler.java:81)
    	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
    	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
    	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
    	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
    	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
    	at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    	at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    	at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    	at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

4、总结，一定要从jobmanager里查看相关日志信息。会有相关报错信息的

5、因为自己继承了自定义sink类CustomSinkBase。**而该类中的sync方法，会在checkpoint的时候调用，如果里面有npe错误，则导致checkpoint失败的情况，所以在调用sync的方法还是要注意点，注意错误捕获**。该方法主要是在一些保存使用的。

public abstract class CustomSinkBase implements Serializable {
        protected Map<String, String> userParamsMap;
        protected Set<String> primaryKeys;
        protected List<String> headerFields;
        protected RowTypeInfo rowTypeInfo;
    
        public CustomSinkBase() {
        }
    
        public CustomSinkBase setUserParamsMap(Map<String, String> userParamsMap) {
            this.userParamsMap = userParamsMap;
            return this;
        }
    
        public CustomSinkBase setPrimaryKeys(Set<String> primaryKeys) {
            this.primaryKeys = primaryKeys;
            return this;
        }
    
        public CustomSinkBase setHeaderFields(List<String> headerFields) {
            this.headerFields = headerFields;
            return this;
        }
    
        public CustomSinkBase setRowTypeInfo(RowTypeInfo rowTypeInfo) {
            this.rowTypeInfo = rowTypeInfo;
            return this;
        }
    
        public abstract void open(int var1, int var2) throws IOException;
    
        public abstract void close() throws IOException;
    
        public abstract void writeAddRecord(Row var1) throws IOException;
    
        public abstract void writeDeleteRecord(Row var1) throws IOException;
    
        public abstract void sync() throws IOException;
    
        public abstract String getName();
    }

[https_mp.weixin.qq.com_s_5o8rbz07Je17ZHnRz97-Rg]: https://mp.weixin.qq.com/s/5o8rbz07Je17ZHnRz97-Rg