Merge squashed changes from kafka

Add initial offset logging
Rename consumer.bufferSize to consumer.cacheSize
Set corePoolSize for single metadata refresh thread.
MLHR-1471 #resolve #comment force close kafka connection
MLHR-1439 #resolve #comment use single thread to refresh the metadata
Close the BenchmarkProducer in periodical producing mode
Make consumer bufferSize configurable
Set to repartition*Interval < 0 to make one_to_many none repartitionable
Use ConcurrentHashMap to avoid ConcurrentModificationException and read-write lock
MLHR-1441 #resolve #comment Use ConcurrentHashMap to avoid ConcurrentModificationException and read-write lock
Make KafkaOutputBenchmark application be able to control the throughput and Fix ConcurrentModificationException
Change offsetManager to be not transient