Apache Beam 应用程序出现检查点故障 - Managed Service for Apache Flink
症状原因解决方案

Amazon Managed Service for Apache Flink 之前称为 Amazon Kinesis Data Analytics for Apache Flink。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

Apache Beam 应用程序出现检查点故障

如果您的 Beam 应用程序配置为shutdownSourcesAfterIdleMs设置为 0ms，则检查点可能无法触发，因为任务处于 “FINISHED” 状态。本节介绍这种情况的症状和解决方法。

症状

前往您的 Apache Flink 应用程序 CloudWatch 日志托管服务，检查是否记录了以下日志消息。以下日志消息表明，由于某些任务已经完成，检查点未能触发。


                {
                "locationInformation": "org.apache.flink.runtime.checkpoint.CheckpointCoordinator.onTriggerFailure(CheckpointCoordinator.java:888)",
                "logger": "org.apache.flink.runtime.checkpoint.CheckpointCoordinator",
                "message": "Failed to trigger checkpoint for job your job ID since some tasks of job your job ID has been finished, abort the checkpoint Failure reason: Not all required tasks are currently running.",
                "threadName": "Checkpoint Timer",
                "applicationARN": your application ARN,
                "applicationVersionId": "5",
                "messageSchemaVersion": "1",
                "messageType": "INFO"
                }

这也可以在 Flink 控制面板上找到，其中一些任务已进入 “FINISHED” 状态，并且无法再进行检查点操作了。

处于 “FINISHED” 状态的任务

原因

shutdownSourcesAfterIdleMs 是一个 Beam 配置变量，用于关闭在配置的毫秒时间内处于空闲状态的信号源。一旦源被关闭，就无法再进行检查点检查了。这可能导致检查点失败。

任务进入 “FINISHED” 状态的原因之一 shutdownSourcesAfterIdleMs 是何时设置为 0 毫秒，这意味着空闲的任务将立即关闭。

解决方案

要防止任务立即进入 “FINISHED” 状态，请 shutdownSourcesAfterIdleMs 将其设置为 “长”。 MAX_ VALUE。这可以通过两种方式完成：

  • 选项 1：如果在 Apache Flink 托管服务应用程序配置页面中设置了 beam 配置，则可以添加新的密钥值对来设置 shutdpwnSourcesAfteridle Ms，如下所示：

    设置 shutdownSourcesAfterIdleMs 为 “长”。 MAX_ VALUE

  • 选项 2：如果您的光束配置已在JAR文件中设置，则可以 shutdownSourcesAfterIdleMs 按以下方式进行设置：

    
                        FlinkPipelineOptions options = PipelineOptionsFactory.create().as(FlinkPipelineOptions.class); // Initialize Beam Options object

                        options.setShutdownSourcesAfterIdleMs(Long.MAX_VALUE); // set shutdownSourcesAfterIdleMs to Long.MAX_VALUE
                        options.setRunner(FlinkRunner.class);

                        Pipeline p = Pipeline.create(options); // attach specified options to Beam pipeline