idea 中模拟flink checkpoint

在IDEA中模拟Flink的Checkpoint，可以使用Flink提供的TestingUtils类来进行模拟。具体步骤如下：

1. 创建一个Flink的测试环境，可以使用TestingUtils类的createTestEnvironment方法来创建。

2. 在测试环境中，创建一个带有Checkpoint的DataStream。

3. 启动测试环境，并等待Checkpoint完成。

4. 对于每个Checkpoint，可以使用TestingUtils类的advanceTime方法来模拟时间的推进。

5. 在测试完成后，可以使用TestingUtils类的assertStreamEquals方法来验证DataStream的输出结果是否正确。

下面是一个简单的示例代码：

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.util.TestStreamEnvironment;
import org.apache.flink.test.util.AbstractTestBase;
import org.junit.Test;

import java.util.concurrent.TimeUnit;

public class CheckpointTest extends AbstractTestBase {

    @Test
    public void testCheckpoint() throws Exception {
        StreamExecutionEnvironment env = TestStreamEnvironment.createTestStreamEnvironment();

        env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);

        env.setParallelism(1);

        env.addSource(new SourceFunction<String>() {
            private volatile boolean running = true;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (running) {
                    ctx.collectWithTimestamp("hello", System.currentTimeMillis());
                    ctx.emitWatermark(new Watermark(System.currentTimeMillis()));
                    TimeUnit.MILLISECONDS.sleep(500);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        })
                .keyBy(value -> value)
                .timeWindow(Time.seconds(1))
                .reduce((value1, value2) -> value1 + " " + value2)
                .print();

        env.execute();

        // 模拟第一个checkpoint
        TestStreamEnvironment.setCurrentTime(1000);
        env.getCheckpointCoordinator().triggerCheckpoint(System.currentTimeMillis(), false);

        // 模拟第二个checkpoint
        TestStreamEnvironment.setCurrentTime(2000);
        env.getCheckpointCoordinator().triggerCheckpoint(System.currentTimeMillis(), false);
    }
}

上面的代码演示了如何在测试环境中模拟Flink的Checkpoint，其中使用了TestingUtils类的一些方法来模拟时间的推进和验证输出结果。