Add unit test and fix the OffsetManager bug in Kafka ONE_TO_MANY, partitionable input operator

partitionable input operator scenario