fix issues about kafka offset management