Deploy & operate — upgrade a stateful job via savepoint without losing state

package de.flk;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;

/** Provided — do not edit. Emits 0..K-1 then blocks forever. Keeps its emitted
 *  offset in OPERATOR state so a savepoint resumes it where it left off (so
 *  after restore with a larger K it emits only the NEW records). */
public class BoundedThenBlock
        implements SourceFunction<Long>, CheckpointedFunction {

    private volatile boolean running = true;
    private final int k;
    private long next = 0;
    private transient ListState<Long> offsetState;

    public BoundedThenBlock(int k) {
        this.k = k;
    }

    @Override
    public void run(SourceContext<Long> ctx) throws Exception {
        while (running) {
            synchronized (ctx.getCheckpointLock()) {
                if (next < k) {
                    ctx.collect(next);
                    next++;
                }
            }
            if (next >= k) {
                Thread.sleep(50);
            }
        }
    }

    @Override
    public void cancel() {
        running = false;
    }

    @Override
    public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
        offsetState.clear();
        offsetState.add(next);
    }

    @Override
    public void initializeState(FunctionInitializationContext ctx) throws Exception {
        offsetState = ctx.getOperatorStateStore()
            .getListState(new ListStateDescriptor<>("offset", Long.class));
        if (ctx.isRestored()) {
            for (Long v : offsetState.get()) {
                next = v;
            }
        }
    }
}

package de.flk;

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;

/** Starter — this keeps the count in a plain instance field, which is NOT part
 *  of Flink's managed state, so it resets to zero after a savepoint restore
 *  (the test ends at 5 instead of 10). Fix it with keyed ValueState — see
 *  README. */
public class Counter extends KeyedProcessFunction<Integer, Long, Long> {

    private transient long n;  // TODO: this is not managed state — use ValueState
    private final String outPath;

    public Counter(String outPath) {
        this.outPath = outPath;
    }

    @Override
    public void open(OpenContext ctx) {
        n = 0;
    }

    @Override
    public void processElement(Long value, Context ctx, Collector<Long> out)
            throws Exception {
        n++;
        Files.write(Paths.get(outPath), (n + "\n").getBytes(),
            StandardOpenOption.CREATE, StandardOpenOption.WRITE,
            StandardOpenOption.TRUNCATE_EXISTING);
        out.collect(n);
    }
}

package de.flk;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/** Provided — do not edit. Wires the pipeline and assigns a stable uid() so the
 *  savepoint can map the Counter's state back on restore.
 *  Program args: [0] output file path, [1] K (records the source emits). */
public class Job {
    public static void main(String[] args) throws Exception {
        String outPath = args[0];
        int k = Integer.parseInt(args[1]);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(400);
        env.setParallelism(1);

        DataStream<Long> src = env.addSource(new BoundedThenBlock(k)).uid("src");
        SingleOutputStreamOperator<Long> counted =
            src.keyBy(x -> 0).process(new Counter(outPath)).uid("counter-v1");
        counted.print();

        env.execute("stateful-counter");
    }
}