TestCheckpointJobWithBootstrap.java

public class TestCheckpointJobWithBootstrap {

    public static void main(String[] args) throws Exception {
      bootstrapConfig();
      ... rest same as previous snippet ...
    }
    static void bootstrapConfig() throws IOException {
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        ExistingSavepoint existingSavepoint = Savepoint.load(executionEnvironment, "oldSavepointPath", new MemoryStateBackend());

        TestConfig testConfig = new TestConfig();
        testConfig.setKey("global");
        testConfig.setThresholdValue(10);

        DataSet<TestConfig> configDataSet = executionEnvironment.fromElements(testConfig);
        BootstrapTransformation<TestConfig> transformation = OperatorTransformation
                .bootstrapWith(configDataSet)
                .keyBy(TestConfig::getKey)
                .transform(new ConfigBootstrapper());

        String newSavepointPath = "newSavepointPath";
        existingSavepoint.withOperator("classify_data", transformation).write(newSavepointPath);
    }
}