add consumer offfset persistence and locks