Implement counters for kafka input operator

Add brokerid, brokerhost,(new java bean per partition) connection num to KafkaMeterStats
Add KafkaMeterStatsAggregator for aggregation of kafka stats
Change msg/s, bytes/s from double to long
Reuse producer in KafkaOutputBenchmark application