package com.ennova.flink.checkpoint;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class CheckPointApp {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//经测试实际结果是每次间隔22秒
env.enableCheckpointing(5000);
DataStreamSource source = env.socketTextStream(“192.168.1.130”, 9527);
source.map(new MapFunction<String, String>() {
@Override
public String map(String value) throws Exception {
if (value.contains(“rambo”)){
throw new Exception(“rambo is coming”);
}else {
return value.toUpperCase();
}
}
}).print();
env.execute(“jobname”);
}
}

-------测试结果如下图----
测试了好多次都是这样的结果，而且我把您的代码拉下来跑了一下，也是20多秒。