Automatic retry configuration

This article provides implementation ideas, configuration examples and verification methods for Flink’s automatic retry scenario, which can be used as a reference for actual project configuration.

Case

Real-time synchronization configuration automatic retry

solution

Modify job startup configuration

{
  "execution.checkpointing.interval": "10000",
  "restart-strategy.fixed-delay.delay": "60s",
  "restart-strategy.type": "fixed-delay",
  "restart-strategy.fixed-delay.attempts": 10,
  "state.backend.type": "filesystem",
  "state.checkpoints.dir": "file:///Users/ispong/flink",
  "execution.checkpointing.externalized-checkpoint-retention": "RETAIN_ON_CANCELLATION"
}

CREATE TABLE from_table
(
    username STRING,
    age      INT
) WITH ( 'connector' = 'mysql-cdc',
      'hostname' = 'localhost',
      'port' = '30306',
      'username' = 'ispong',
      'password' = 'ispong123',
      'database-name' = 'isxcode_db',
      'table-name' = 'cdc_source',
      'scan.incremental.snapshot.enabled' = 'false',
      'server-time-zone' = 'UTC',
      'scan.startup.mode' = 'latest-offset'
      );

CREATE TABLE target_table
(
    username STRING,
    age      INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH ( 'connector' = 'jdbc',
      'url' = 'jdbc:mysql://localhost:30306/isxcode_db?useSSL=false&allowPublicKeyRetrieval=true',
      'driver' = 'com.mysql.cj.jdbc.Driver',
      'table-name' = 'cdc_target',
      'username' = 'ispong',
      'password' = 'ispong123');

INSERT INTO target_table
select *
from from_table;

Specify checkpoint path recovery

Specify a specific point file for recovery

{
  "execution.checkpointing.interval": "10000",
  "restart-strategy.fixed-delay.delay": "60s",
  "restart-strategy.type": "fixed-delay",
  "restart-strategy.fixed-delay.attempts": 10,
  "state.backend.type": "filesystem",
  "state.checkpoints.dir": "file:///Users/ispong/flink",
  "execution.savepoint.path": "file:///Users/ispong/flink/948eeadc5c91d9d3da00046304874ce8/chk-4",
  "execution.checkpointing.externalized-checkpoint-retention": "RETAIN_ON_CANCELLATION"
}