StatefulProcess.java

public class StatefulProcess extends KeyedProcessFunction<String, KeyValue, KeyValue> {
    ValueState<Integer> processedInt;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        processedInt = getRuntimeContext().getState(new ValueStateDescriptor<>("processedInt", Integer.class));
    }

    @Override
    public void processElement(KeyValue keyValue, Context context, Collector<KeyValue> collector) throws Exception {
        try{
            Integer a =  Integer.parseInt(keyValue.getValue());
            processedInt.update(a);
            collector.collect(keyValue);
        }catch(Exception e){
            e.printStackTrace();
        }
    }
}