support for (1) kafka offsets reset, (2) read from kafka latest