code: public void getTopicPartitionLogSize(TopicPartitionInfo stat){
      	BrokerEndPoint leader = findLeader(stat.getTopic(), stat.getPartition()).leader();
      	SimpleConsumer consumer = getConsumerClient(leader.host(), leader.port());	
      	
      	try {			
      		long logsize = getLastOffset(consumer,stat.getTopic(), stat.getPartition(), kafka.api.OffsetRequest.LatestTime());
      		stat.setLogSize(logsize);
  		} finally {
  			consumer.close();
  		}
      }
docs: 获取指定主题及分区logsize
  @param stat