code: @Override
  	public void snapshotState(FunctionSnapshotContext context) throws Exception {
  		sessionIdsPerSnapshot.add(new Tuple2<>(context.getCheckpointId(), sessionIds));
  		sessionIds = new ArrayList<>(64);
  		super.snapshotState(context);
  	}
docs: ------------------------------------------------------------------------