/**
 * 
 */
package com.varone.node.utils;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import com.varone.node.MetricsProperties;

/**
 * @author allen
 *
 */
public class MetricsPropertiesParser {
	
	public final static String CSV_PERIOD = "*.sink.csv.period";
	public final static String CSV_UNIT = "*.sink.csv.unit";
	public final static String CSV_DIR = "*.sink.csv.directory";
	
	public static MetricsProperties loadProperties(String filePath) {
		MetricsProperties metricsProps = null;
		Properties prop = new Properties();
		InputStream input = null;
		
		try{
			input = new FileInputStream(filePath);

			prop.load(input);
			
			metricsProps = new MetricsProperties();
			
			String csvPeriod;
			if(null == (csvPeriod = prop.getProperty(CSV_PERIOD))){
				throw new RuntimeException("Can not find '"+CSV_PERIOD+"' in metrics properties file: " + filePath);
			} else {
				metricsProps.setCsvPeriod(csvPeriod);
			}
			
			String csvUnit;
			if(null == (csvUnit = prop.getProperty(CSV_UNIT))){
				throw new RuntimeException("Can not find '"+CSV_UNIT+"' in metrics properties file: " + filePath);
			} else {
				metricsProps.setCsvUnit(csvUnit);
			}
			
			String csvDir;
			if(null == (csvDir = prop.getProperty(CSV_DIR))){
				throw new RuntimeException("Can not find '"+CSV_DIR+"' in metrics properties file: " + filePath);
			} else {
				metricsProps.setCsvDir(csvDir);
			}
			
			
		} catch(IOException e){
			throw new RuntimeException(e);
		} finally {
			if (input != null) {
				try {
					input.close();
				} catch(IOException e){
					
				}
			}
		}
		
		return metricsProps;
		
	}
}

@@ -318,16 +318,75 @@
 ser %7B%0A%09%0A
+%09public final static String CSV_SINK = %22*.sink.csv.class%22;%0A
 %09public 
@@ -856,24 +856,310 @@
 ties();%0A%09%09%09%0A
+%09%09%09String csvSink;%0A%09%09%09if(null == (csvSink = prop.getProperty(CSV_SINK)) %7C%7C %0A%09%09%09%09%09!csvSink.equals(%22org.apache.spark.metrics.sink.CsvSink%22))%7B%0A%09%09%09%09throw new RuntimeException(%22Should allow output metrics to csv by setting *.sink.csv.class=org.apache.spark.metrics.sink.CsvSink%22);%0A%09%09%09%7D %0A%09%09%09%0A
 %09%09%09String cs

add the csv sink validation