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

partitionable input operator scenario
Should not call updateOffsets on every stats update.