Reproduce

gistfile1.txt

public class ExampleProgram {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.socketTextStream("localhost", 9999)
				.shuffle()
				.filter(new StatefulFilter()).uid("f")
				.shuffle()
				// .map(x -> x)
				.print();

		CheckpointConfig cfg = env.getCheckpointConfig();
		cfg.setCheckpointInterval(1000);
		env.setStateBackend(new MemoryStateBackend());

		env.execute();
	}

	private static class StatefulFilter implements FilterFunction<String>, ListCheckpointed<String> {
		@Override
		public boolean filter(String arg0) throws Exception {
			return true;
		}

		@Override
		public void restoreState(List<String> state) throws Exception {}

		@Override
		public List<String> snapshotState(long checkpointId, long timestamp) throws Exception {
			return Collections.singletonList("a");
		}

	}
}