1. Have a factory method for creating offset manager instances. Got rid of create offsets and have that logic in startup
2. Use concurrentHashMap for storing offsets
3. Added guage to trace the size of the hash table
4. Better naming of channel and producer methods