Squashed commit of the following:

MLHR-1471 #resolve #comment force close kafka connection

MLHR-1439 #resolve #comment use single thread to refresh the metadata

Other minor change:
1. Close the BenchmarkProducer in periodical producing mode
2. Make consumer bufferSize configurable
3. Set to repartition*Interval < 0 to make one_to_many none repartitionable
Merge pull request #1105 from siyuanh/ukafkafix

Big kafka consumer change