Add unit test and fix the OffsetManager bug in ONE_TO_MANY,
partitionable input operator scenario
Tune the offsetmanager and add initialoffset to OffsetManagerTest
Add logs
Merge pull request #1138 from siyuanh/kafkafix

Fix MLHR-1510