Big kafka consumer change

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