Échec de point de contrôle pour l’application Apache Beam

Si votre application Beam est configurée avec une valeur shutdownSourcesAfterIdleMsdéfinie sur 0 ms, les points de contrôle peuvent ne pas se déclencher car les tâches sont à l'état « TERMINÉ ». Cette section décrit les symptômes et la résolution de ce problème.

Symptôme

Accédez aux CloudWatch journaux de votre application Managed Service for Apache Flink et vérifiez si le message de journal suivant a été enregistré. Le message de journal suivant indique que le point de contrôle n’a pas pu être déclenché, car certaines tâches sont terminées.

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

Vous pouvez également le voir sur le tableau de bord de Flink, où certaines tâches sont passées à l’état « TERMINÉ » et où le point de contrôle n’est plus possible.

Cause

shutdownSourcesAfterIdleMs est une variable de configuration Beam qui arrête les sources qui sont restées inactives pendant la durée configurée de millisecondes. Une fois qu’une source a été fermée, le point de contrôle n’est plus possible. Cela pourrait entraîner un échec du point de contrôle.

L'une des raisons pour lesquelles les tâches passent à l'état « TERMINÉ » shutdownSourcesAfter IdleMs est lorsqu'elle est définie sur 0 ms, ce qui signifie que les tâches inactives seront immédiatement arrêtées.

Solution

Pour empêcher les tâches de passer immédiatement à l'état « TERMINÉ », définissez ce paramètre sur shutdownSourcesAfter IdleMs Long.MAX_VALUE. Vous pouvez effectuer cette opération de deux façons :

• Option 1 : Si la configuration de votre faisceau est définie sur la page de configuration de votre application Managed Service for Apache Flink, vous pouvez ajouter une nouvelle paire clé-valeur pour définir shutdpwnSourcesAfteridle Ms comme suit :

  

• Option 2 : Si la configuration de votre faisceau est définie dans votre fichier JAR, vous pouvez définir shutdownSourcesAfter IdleMs comme suit :

  
                          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