Implement fault tolerance in Managed Service for Apache Flink

Checkpointing is the method that is used for implementing fault tolerance in Amazon Managed Service for Apache Flink. A checkpoint is an up-to-date backup of a running application that is used to recover immediately from an unexpected application disruption or failover.

For details on checkpointing in Apache Flink applications, see Checkpoints in the Apache Flink Documentation.

A snapshot is a backup of application state. You create snapshots. Managed Service for Apache Flink creates system snapshots for some operations. To restore a stopped application from a snapshot, call StartApplication. To restore a running application from a snapshot, call UpdateApplication. For more information, see Manage application backups using snapshots.

Managed Service for Apache Flink uses the following types of backup:

• Checkpoints: If checkpointing is enabled, Managed Service for Apache Flink creates checkpoints automatically for a running application. The application uses the newest checkpoint to recover from a failure.

• Enhanced snapshots: An enhanced snapshot is a backup of application state that has its own ARN. The enhanced snapshot remains after you delete the application. You can restore it into any application in enhanced snapshot mode in the same account and Region.

• Application snapshots (legacy): An application snapshot is a backup of application state that belongs to the application that created it. Only that application can restore from it.

An application uses one snapshot type at a time. Enhanced snapshots require Apache Flink version 1.20 or later and enhanced snapshot mode. For more information, see How enhanced snapshots work.

If checkpointing is enabled for your application, then the service provides fault tolerance by creating and loading backups of application data in the event of unexpected application restarts. These unexpected application restarts could be caused by unexpected job restarts, instance failures, etc. This gives the application the same semantics as failure-free execution during these restarts.

If snapshots are enabled for the application, and configured using the application's ApplicationRestoreConfiguration, then the service provides exactly-once processing semantics during application updates, or during service-related scaling or maintenance.

Configure checkpointing in Managed Service for Apache Flink

You can configure your application's checkpointing behavior. You can define whether it persists the checkpointing state, how often it saves its state to checkpoints, and the minimum interval between the end of one checkpoint operation and the beginning of another.

You configure the following settings using the CreateApplication or UpdateApplication API operations:

• CheckpointingEnabled — Indicates whether checkpointing is enabled in the application.

• CheckpointInterval — Contains the time in milliseconds between checkpoint (persistence) operations.

• ConfigurationType — Set this value to DEFAULT to use the default checkpointing behavior. Set this value to CUSTOM to configure other values.

  Note

  The default checkpoint behavior is as follows:

  • CheckpointingEnabled: true

  • CheckpointInterval: 60000

  • MinPauseBetweenCheckpoints: 5000

  If ConfigurationType is set to DEFAULT, the preceding values will be used, even if they are set to other values using either using the AWS Command Line Interface, or by setting the values in the application code.

  Note

  For Apache Flink 1.15 and later, Managed Service for Apache Flink uses stop-with-savepoint to create a system snapshot. For the operations that create a system snapshot, see System snapshots.

• MinPauseBetweenCheckpoints — The minimum time in milliseconds between the end of one checkpoint operation and the start of another. Setting this value prevents the application from checkpointing continuously when a checkpoint operation takes longer than the CheckpointInterval.

Review checkpointing API examples

This section includes example requests for API actions for configuring checkpointing for an application. For information about how to use a JSON file for input for an API action, see Managed Service for Apache Flink API example code.

Configure checkpointing for a new application

The following example request for the CreateApplication action configures checkpointing when you are creating an application:

{
   "ApplicationName": "MyApplication",
   "RuntimeEnvironment":"FLINK-1_19",
   "ServiceExecutionRole":"arn:aws:iam::123456789123:role/myrole",
   "ApplicationConfiguration": { 
      "ApplicationCodeConfiguration":{
      "CodeContent":{
        "S3ContentLocation":{
          "BucketARN":"arn:aws:s3:::amzn-s3-demo-bucket",
          "FileKey":"myflink.jar",
          "ObjectVersion":"AbCdEfGhIjKlMnOpQrStUvWxYz12345"
        }
      },
      "FlinkApplicationConfiguration": { 
         "CheckpointConfiguration": { 
            "CheckpointingEnabled": "true",
            "CheckpointInterval": 20000,
            "ConfigurationType": "CUSTOM",
            "MinPauseBetweenCheckpoints": 10000
         }
      }
}

Disable checkpointing for a new application

The following example request for the CreateApplication action disables checkpointing when you are creating an application:

{
   "ApplicationName": "MyApplication",
   "RuntimeEnvironment":"FLINK-1_19",
   "ServiceExecutionRole":"arn:aws:iam::123456789123:role/myrole",
   "ApplicationConfiguration": { 
      "ApplicationCodeConfiguration":{
      "CodeContent":{
        "S3ContentLocation":{
          "BucketARN":"arn:aws:s3:::amzn-s3-demo-bucket",
          "FileKey":"myflink.jar",
          "ObjectVersion":"AbCdEfGhIjKlMnOpQrStUvWxYz12345"
        }
      },
      "FlinkApplicationConfiguration": { 
         "CheckpointConfiguration": { 
            "CheckpointingEnabled": "false"
         }
      }
}

Configure checkpointing for an existing application

The following example request for the UpdateApplication action configures checkpointing for an existing application:

{
   "ApplicationName": "MyApplication",
   "ApplicationConfigurationUpdate": { 
      "FlinkApplicationConfigurationUpdate": { 
         "CheckpointConfigurationUpdate": { 
            "CheckpointingEnabledUpdate": true,
            "CheckpointIntervalUpdate": 20000,
            "ConfigurationTypeUpdate": "CUSTOM",
            "MinPauseBetweenCheckpointsUpdate": 10000
         }
      }
   }
}

Disable checkpointing for an existing application

The following example request for the UpdateApplication action disables checkpointing for an existing application:

{
   "ApplicationName": "MyApplication",
   "ApplicationConfigurationUpdate": { 
      "FlinkApplicationConfigurationUpdate": { 
         "CheckpointConfigurationUpdate": { 
            "CheckpointingEnabledUpdate": false,
            "CheckpointIntervalUpdate": 20000,
            "ConfigurationTypeUpdate": "CUSTOM",
            "MinPauseBetweenCheckpointsUpdate": 10000
         }
      }
   }
}