Falha no ponto de verificação do aplicativo Apache Beam

Se seu aplicativo Beam estiver configurado com shutdownSourcesAfterIdleMs0 ms, os pontos de verificação podem falhar ao serem acionados porque as tarefas estão no estado “CONCLUÍDO”. Esta seção descreve os sintomas e a resolução dessa condição.

Sintomas

Acesse os CloudWatch registros do aplicativo Managed Service for Apache Flink e verifique se a seguinte mensagem de log foi registrada. A mensagem de log a seguir indica que o ponto de verificação falhou ao ser acionado, pois algumas tarefas foram concluídas.

                {
                "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"
                }
            

Isso também pode ser encontrado no painel do Flink, onde algumas tarefas entraram no estado “CONCLUÍDO” e o ponto de verificação não é mais possível.

Causa

shutdownSourcesAfterIdleMs é uma variável de configuração do Beam que desliga fontes que ficaram ociosas pelo tempo configurado de milissegundos. Depois que uma fonte é desligada, o ponto de verificação não é mais possível. Isso pode levar à falha do ponto de verificação.

Uma das causas para as tarefas entrarem no estado “CONCLUÍDO” é quando shutdownSourcesAfter IdleMs está configurado para 0 ms, o que significa que as tarefas que estão ociosas serão encerradas imediatamente.

Solução

Para evitar que as tarefas entrem imediatamente no estado “FINISHED”, shutdownSourcesAfter IdleMs defina como Long.MAX_VALUE. Isso pode ser feito de duas maneiras:

• Opção 1: Se a configuração do beam estiver definida na página de configuração do aplicativo Managed Service for Apache Flink, você poderá adicionar um novo par de valores-chave para definir shutdpwnSourcesAfteridle Ms da seguinte forma:

  

• Opção 2: Se a configuração do feixe estiver definida no arquivo JAR, você poderá definir da shutdownSourcesAfter IdleMs seguinte forma:

  
                          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