state无法从checkpoint中恢复

state无法从checkpoint中恢复 配置代码env.enableCheckpointing(1000);env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

//作业失败后不重启

env.setRestartStrategy(RestartStrategies.noRestart());

env.getCheckpointConfig().setCheckpointTimeout(500);

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

env.setStateBackend(new RocksDBStateBackend("file:///opt/flink/flink-1.7.2/checkpoints")); 使用状态的代码private transient ListState<String> counts;

@Override

public void open(Configuration parameters) throws Exception {

StateTtlConfig ttlConfig = StateTtlConfig

.newBuilder(Time.minutes(30))

.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)

.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)

.build();

ListStateDescriptor<String> lastUserLogin = new ListStateDescriptor<>("lastUserLogin", String.class);

lastUserLogin.enableTimeToLive(ttlConfig);

counts = getRuntimeContext().getListState(lastUserLogin);

}

我重启了task managers 后。发现 counts 里面的数据都丢失了

*来自志愿者整理的flink邮件归档