转载:记一次flink不做checkpoint的问题 请参考原文,这里仅仅是学习一下。
问题现象: Flink UI界面查看checkpoint的metrics发现一直没有做checkpoint,仔细排查发现有部分subtask的状态是finished。
下图是测试环境复现问题
问题原因: 仔细排查代码后发现source是消费kafka的数据,配置的并行度大于kafka的partition数,导致有部分subtask空闲,然后状态变为finished。后来查看了checkpoint过程的源码得以佐证。
在CheckpointCoordinator类的triggerCheckpoint方法中有如下代码段
// check if all tasks that we need to trigger are running.
// if not, abort the checkpoint
Execution[] executions = new Execution[tasksToTrigger.length];
for (int i = 0; i < tasksToTrigger.length; i++) {
Execution ee = tasksToTrigger[i].getCurrentExecutionAttempt();
if (ee == null) {
LOG.info("Checkpoint triggering task {} of job {} is not being executed at the moment. Aborting checkpoint.",
tasksToTrigger[i].getTaskNameWithSubtaskIndex(),
job);
throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
} else if (ee.getState() == ExecutionState.RUNNING) {
executions[i] = ee;
} else {
LOG.info("Checkpoint triggering task {} of job {} is not in state {} but {} instead. Aborting checkpoint.",
tasksToTrigger[i].getTaskNameWithSubtaskIndex(),
job,
ExecutionState.RUNNING,
ee.getState());
throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
}
ee.getState() == ExecutionState.RUNNING判断execution的状态是否为running,否则不做checkpoint
问题结论: 在消费kafka的数据时,source的并发度不能超过kafka的partition数,可以小于partition,但是部分subtask就会消费多个partition的数据,导致吞吐达不到最大,理想状态是source并发度等于partition数。
注: 此问题出现时使用的是flink1.5
版本,高版本已经没有此问题了。
版权说明 : 本文为转载文章, 版权归原作者所有 版权申明
原文链接 : https://blog.csdn.net/qq_21383435/article/details/121443208
内容来源于网络,如有侵权,请联系作者删除!