public enum Status {
	NEW, STARTED, COMPLETED, FAILED
} public interface DCMCodec extends Closeable {

	public OutputStream createOutputStream(Configuration conf, String path,
			boolean append) throws IOException;

	public InputStream createInputStream(Configuration conf, String path)
			throws IOException;

	public boolean deleteSoureFile(String path) throws IOException;

	public boolean isSplitable();

	public boolean isExistsPath(String path) throws IOException;

	public List<FileTuple> getInputPaths(String path,
			Collection<String> excludeList) throws Exception;

	public List<FileTuple> getInputPaths(Collection<String> paths,
			Collection<String> excludeList) throws Exception;
} public class DCMCodecFactory {

	public static DCMCodec getCodec(Configuration conf, ConnectionConfig config)
			throws IOException {
		try {
			String scheme = null;
			FileSystem fs = null;
			switch (config.getType()) {

			case WEBHDFS:
				scheme = DCMConstants.WEBHDFS_DEFAULT_PROTOCOL;
				break;
			case HDFS:
				scheme = DCMConstants.HDFS_DEFAULT_PROTOCOL;
				break;
			case HFTP:
				scheme = DCMConstants.HFTP_DEFAULT_PROTOCOL;
				break;
			case HAR:
				scheme = DCMConstants.HAR_DEFAULT_PROTOCOL;
				break;
			case FTP:
				scheme = DCMConstants.FTP_DEFAULT_PROTOCOL;
				break;
			case GS:
				scheme = DCMConstants.GS_DEFAULT_PROTOCOL;
				break;
			case CUSTOM:

			default:
				break;
			}
			if (scheme == null)
				throw new Exception("Unknown Filesystem, " + config.getType());

			if (fs == null)
				fs = getHadoopFilesystem(scheme, conf, config);
			return new GenericHadoopCodec(fs);
		} catch (Exception e) {
			throw new IOException(e);
		}
	}

	public static FileSystem getHadoopFilesystem(String scheme,
			Configuration conf, ConnectionConfig config) throws Exception {

		String httpfsUrl = null;
		if (config.getPort() > 0)
			httpfsUrl = scheme + config.getHost() + ":" + config.getPort();
		else
			httpfsUrl = scheme + config.getHost();

		if (config.getSecurityType() == SecurityType.KERBEROS)
			return FileSystem.newInstance(new URI(httpfsUrl), conf);
		else
			return FileSystem.newInstance(new URI(httpfsUrl), conf,
					config.getUserName());
	}

} public class DCMConfig {

	private String batchName = null;

	private int numWorkers = 0;

	private boolean ignoreException = false;

	private String statusPath = null;

	private StateManagerImpl stateManagerType = null;

	private SourceConfig sourceConfig = null;

	private SinkConfig sinkConfig = null;

	private DBConfig dbConfig = null;

	public long getBatchID() {
		return DCMConstants.HASH_MEDIAN
				+ DCMConstants.hasher.hashBytes(
						batchName.toLowerCase().getBytes()).asInt();
	}

	public String getBatchName() {
		return batchName;
	}

	public void setBatchName(String batchName) {
		this.batchName = batchName;
	}

	public boolean isIgnoreException() {
		return ignoreException;
	}

	public void setIgnoreException(boolean ignoreException) {
		this.ignoreException = ignoreException;
	}

	public String getStatusPath() {
		return statusPath;
	}

	public void setStatusPath(String statusPath) {
		this.statusPath = statusPath;
	}

	public SourceConfig getSourceConfig() {
		return sourceConfig;
	}

	public void setSourceConfig(SourceConfig sourceConfig) {
		this.sourceConfig = sourceConfig;
	}

	public SinkConfig getSinkConfig() {
		return sinkConfig;
	}

	public void setSinkConfig(SinkConfig sinkConfig) {
		this.sinkConfig = sinkConfig;
	}

	public DBConfig getDbConfig() {
		return dbConfig;
	}

	public void setDbConfig(DBConfig dbConfig) {
		this.dbConfig = dbConfig;
	}

	public int getNumWorkers() {
		return numWorkers;
	}

	public void setNumWorkers(int numWorkers) {
		this.numWorkers = numWorkers;
	}

	public StateManagerImpl getStateManagerType() {
		return stateManagerType;
	}

	public void setStateManagerType(StateManagerImpl stateManagerType) {
		this.stateManagerType = stateManagerType;
	}

	@Override
	public String toString() {
		Gson gson = new Gson();
		return gson.toJson(this);
	}
} public static class FileTuple implements Writable {
	String fileName;
	long size;
	long ts;

	public FileTuple() {

	}

	public FileTuple(String path, long len, long ts) {
		this.fileName = path;
		this.size = len;
		this.ts = ts;
	}

	@Override
	public void write(DataOutput out) throws IOException {
		Text.writeString(out, fileName);
		out.writeLong(size);
		out.writeLong(ts);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		fileName = Text.readString(in);
		size = in.readLong();
		ts = in.readLong();
	}

} public interface StateManager extends Closeable {

	public void beginBatch() throws IOException;

	public void completeBatch(Status status) throws IOException;

	public void updateTransferStatus(TransferStatus status) throws IOException;

	public Map<String, TransferStatus> getTransferStatus(String taskId)
			throws IOException;

	public void savePreiviousTransferStatus(
			Map<String, TransferStatus> prevState) throws IOException;

	public Map<String, TransferStatus> getPreviousTransferStatus()
			throws IOException;

	public Path getReportPath();

	public String getRunId();
} public class StateManagerFactory {

	public static StateManager stateManager = null;

	public static StateManager getStateManager(Configuration conf,
			DCMConfig config) throws IOException {

		if (stateManager == null) {
			switch (config.getStateManagerType()) {
			case HDFS:
				stateManager = new HDFSStateManager(conf, config);
				break;
			case DB:
				stateManager = new MySqlStateManager(conf, config);
				break;
			default:
				stateManager = new HDFSStateManager(conf, config);
				break;
			}
		}
		return stateManager;
	}
} public class TransferStatus {

	private String inputPath = null;
	private String outputPath = null;
	private long inputSize = 0;
	private long outputSize = 0;
	private long ts = 0;
	private boolean inputCompressed = false;
	private boolean inputTransformed = false;
	private boolean outputCompressed = false;
	private String md5Digest = null;
	private String taskID = null;
	private Status status = Status.NEW;

	public String getInputPath() {
		return inputPath;
	}

	public void setInputPath(String inputPath) {
		this.inputPath = inputPath;
	}

	public String getOutputPath() {
		return outputPath;
	}

	public void setOutputPath(String outputPath) {
		this.outputPath = outputPath;
	}

	public long getTs() {
		return ts;
	}

	public void setTs(long ts) {
		this.ts = ts;
	}

	public boolean isInputCompressed() {
		return inputCompressed;
	}

	public void setInputCompressed(boolean inputCompressed) {
		this.inputCompressed = inputCompressed;
	}

	public boolean isInputTransformed() {
		return inputTransformed;
	}

	public void setInputTransformed(boolean inputTransformed) {
		this.inputTransformed = inputTransformed;
	}

	public boolean isOutputCompressed() {
		return outputCompressed;
	}

	public void setOutputCompressed(boolean outputCompressed) {
		this.outputCompressed = outputCompressed;
	}

	public String getMd5Digest() {
		return md5Digest;
	}

	public void setMd5Digest(String md5Digest) {
		this.md5Digest = md5Digest;
	}

	public Status getStatus() {
		return status;
	}

	public void setStatus(Status status) {
		this.status = status;
	}

	public String getTaskID() {
		return taskID;
	}

	public void setTaskID(String taskID) {
		this.taskID = taskID;
	}

	public long getInputSize() {
		return inputSize;
	}

	public void setInputSize(long inputSize) {
		this.inputSize = inputSize;
	}

	public long getOutputSize() {
		return outputSize;
	}

	public void setOutputSize(long outputSize) {
		this.outputSize = outputSize;
	}

	@Override
	public String toString() {
		Gson gson = new Gson();
		return gson.toJson(this);
	}
} public class MirrorUtils {

	public static DCMConfig getConfigFromConf(Configuration conf) {
		String configString = conf.get(MirrorFileInputFormat.DCM_CONFIG);
		Gson gson = new Gson();
		DCMConfig config = gson.fromJson(configString, DCMConfig.class);
		return config;
	}

	public static void setConfigInConf(Configuration conf, DCMConfig config) {

		Gson gson = new Gson();
		String configString = gson.toJson(config);
		conf.set(MirrorFileInputFormat.DCM_CONFIG, configString);

	}

	public static String getCodecNameFromPath(Configuration conf, String path) {
		CompressionCodecFactory compressionCodecs = new CompressionCodecFactory(
				conf);
		CompressionCodec codec = compressionCodecs.getCodec(new Path(path));
		if (codec == null)
			return null;
		else
			return codec.getDefaultExtension();
	}

	public static OutputStream getCodecOutputStream(Configuration conf,
			DCMConfig config, String outPath, OutputStream out)
			throws IOException {
		if (config.getSinkConfig().isUseCompression()) {

			CompressionCodecFactory compressionCodecs = new CompressionCodecFactory(
					conf);
			CompressionCodec codec = compressionCodecs.getCodecByName(config
					.getSinkConfig().getCompressionCodec());
			Compressor compressor = codec.createCompressor();
			out = codec.createOutputStream(out, compressor);
		}
		return out;
	}

	public static InputStream getCodecInputStream(Configuration conf,
			DCMConfig config, String inPath, InputStream in) throws IOException {

		CompressionCodecFactory compressionCodecs = new CompressionCodecFactory(
				conf);
		CompressionCodec codec = compressionCodecs.getCodec(new Path(inPath));
		if (codec == null)
			return in;
		Decompressor compressor = codec.createDecompressor();
		in = codec.createInputStream(in, compressor);

		return in;
	}

	public static MD5Digester copy(InputStream input, OutputStream result,
			TaskAttemptContext context) throws IOException {

		byte[] buffer = new byte[65536]; // 8K=8192 12K=12288 64K=65536
		long count = 0L;
		int n = 0;

		long sts = System.currentTimeMillis();
		MD5Digester digester = new MD5Digester();

		while (-1 != (n = input.read(buffer))) {

			result.write(buffer, 0, n);

			digester.updateMd5digester(buffer, 0, n);
			count += n;

			if (count % 67108864 == 0) {
				System.out.println("Wrote 64M Data to Destination Total: "
						+ count + ", Time Taken(ms): "
						+ (System.currentTimeMillis() - sts));

				sts = System.currentTimeMillis();
				context.progress();
			}
		}
		System.out.println("Transfer Complete Total: " + count
				+ ", Time Taken(ms): " + (System.currentTimeMillis() - sts));
		return digester;
	}

	public static Set<String> getFileAsLists(String fileName) {
		Set<String> files = new HashSet<String>();
		if (fileName != null && fileName.trim().length() > 1) {
			try {
				BufferedReader reader = new BufferedReader(new FileReader(
						fileName));
				String line = null;
				while ((line = reader.readLine()) != null) {
					if (line.trim().length() <= 1 || line.startsWith("#"))
						continue;

					files.add(line);
				}
				reader.close();
			} catch (Exception e) {
				System.out.println("Ignoring FileSet from: " + fileName);
				System.out.println("Excpetion: " + e.getMessage());
			}
		}
		return files;
	}

	public static Set<String> getStringAsLists(String flist) {
		Set<String> files = new HashSet<String>();
		if (flist == null || flist.trim().length() <= 0)
			return files;
		String fileList[] = flist.split(",");

		for (String file : fileList)
			files.add(file);
		return files;
	}

	public static String getListAsString(Collection<String> flist) {
		StringBuilder builder = new StringBuilder();
		if (flist == null || flist.size() <= 0)
			return builder.toString();
		int count = 0;
		for (String file : flist) {
			if (count > 0)
				builder.append(",");
			builder.append(file);
			count++;
		}

		return builder.toString();
	}

	public static String stripExtension(String path) {
		if (path.lastIndexOf('.') < 1)
			return path;
		return path.substring(0, path.lastIndexOf('.'));
	}

	public static String getSimplePath(Path path) {
		return path.toUri().getPath();
	}
} public class OptimTuple implements IInputJob {

	public final String jobKey;
	public final long size;

	public OptimTuple(String commaSepratedInput) {
		String[] inputArray = commaSepratedInput.split(",");
		this.jobKey = inputArray[0];
		this.size = Integer.valueOf(inputArray[2]);
	}

	public OptimTuple(String jobKey, long size) {
		this.jobKey = jobKey;
		this.size = size;
	}

	public int compareTo(IInputJob o) {
		Long l = getJobSize();
		if (o == null)
			return l.compareTo(0L);
		return l.compareTo(o.getJobSize());
	}

	public long getJobSize() {
		return size;
	}

	public String getJobKey() {
		return jobKey;
	}

	public String toString() {
		return "ReduceDataSet{" + "jobKey='" + jobKey + ", size=" + size + '}';
	}
} public interface IInputJob extends Comparable<IInputJob> {
	long getJobSize();

	String getJobKey();
} public interface IJobLoadOptimizer {
	public List<Set<IInputJob>> getOptimizedLoadSets(
			Set<? extends IInputJob> jobs, int noOfWorkers);
} public class JobLoadOptimizerFactory {
	public static enum Optimizer {
		PRIORITY_QUEUE_BASED, AMBITIOUS, PTASOPTMIZER
	}

	public static IJobLoadOptimizer getJobLoadOptimizerFactory(
			Optimizer optimizer) {
		if (optimizer == Optimizer.PRIORITY_QUEUE_BASED) {
			return new PriorityQueueBasedOptimizer();
		} else if (optimizer == Optimizer.AMBITIOUS) {
			return new AmbitiousOptimizer();
		} else if (optimizer == Optimizer.PTASOPTMIZER) {
			return new PTASOptimizer();
		}
		return null;
	}

} public static enum Optimizer {
	PRIORITY_QUEUE_BASED, AMBITIOUS, PTASOPTMIZER
}
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import com.flipkart.fdp.migration.db.models.Status;
import com.flipkart.fdp.migration.distcp.codec.DCMCodec;
import com.flipkart.fdp.migration.distcp.codec.DCMCodecFactory;
import com.flipkart.fdp.migration.distcp.config.DCMConfig;
import com.flipkart.fdp.migration.distcp.core.MirrorDCMImpl.FileTuple;
import com.flipkart.fdp.migration.distcp.state.StateManager;
import com.flipkart.fdp.migration.distcp.state.StateManagerFactory;
import com.flipkart.fdp.migration.distcp.state.TransferStatus;
import com.flipkart.fdp.migration.distcp.utils.MirrorUtils;
import com.flipkart.fdp.optimizer.OptimTuple;
import com.flipkart.fdp.optimizer.api.IInputJob;
import com.flipkart.fdp.optimizer.api.IJobLoadOptimizer;
import com.flipkart.fdp.optimizer.api.JobLoadOptimizerFactory;
import com.flipkart.fdp.optimizer.api.JobLoadOptimizerFactory.Optimizer;
/*
 *
 *  Copyright 2015 Flipkart Internet Pvt. Ltd.
 *
 *     Licensed under the Apache License, Version 2.0 (the "License");
 *     you may not use this file except in compliance with the License.
 *     You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *     Unless required by applicable law or agreed to in writing, software
 *     distributed under the License is distributed on an "AS IS" BASIS,
 *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *     See the License for the specific language governing permissions and
 *     limitations under the License.
 *
 */

package com.flipkart.fdp.migration.distcp.core;




public class MirrorFileInputFormat extends InputFormat<Text, Text> {

	public static final String DCM_CONFIG = "dcm_config";
	public static final String INCLUDE_FILES = "include_files";
	public static final String EXCLUDE_FILES = "exclude_files";

	private DCMCodec dcmCodec = null;
	private Configuration conf = null;
	private DCMConfig dcmConfig = null;
	private StateManager stateManager = null;

	@Override
	public List<InputSplit> getSplits(JobContext context) throws IOException,
			InterruptedException {

		Set<String> excludeList = null;
		Set<String> includeList = null;
		Map<String, TransferStatus> previousState = null;

		System.out.println("Calculating Job Splits...");
		conf = context.getConfiguration();
		dcmConfig = MirrorUtils.getConfigFromConf(conf);

		dcmCodec = DCMCodecFactory.getCodec(conf, dcmConfig.getSourceConfig()
				.getConnectionConfig());

		excludeList = getExclusionsFileList(conf);
		includeList = getInclusionFileList(conf);
		HashMap<String, FileTuple> inputFileMap = new HashMap<String, FileTuple>();

		List<InputSplit> splits = new ArrayList<InputSplit>();
		Set<OptimTuple> locations = new HashSet<OptimTuple>();

		long totalBatchSize = 0;
		try {
			System.out.println("Scanning source location...");
			List<FileTuple> fstats = null;
			if (includeList != null && includeList.size() > 0)
				fstats = dcmCodec.getInputPaths(includeList, excludeList);
			else
				fstats = dcmCodec.getInputPaths(dcmConfig.getSourceConfig()
						.getPath(), excludeList);

			stateManager = StateManagerFactory.getStateManager(conf, dcmConfig);

			System.out
					.println("Fetching previous transfer states from StateManager...");
			previousState = stateManager.getPreviousTransferStatus();

			System.out
					.println("Filtering Input File Set based on User defined filters.");
			for (FileTuple fstat : fstats) {

				if (!ignoreFile(fstat, excludeList, previousState)) {

					locations.add(new OptimTuple(fstat.fileName, fstat.size));
					inputFileMap.put(fstat.fileName, fstat);
				}
			}
			System.out.println("Optimizing Splits...");
			int numWorkers = locations.size();
			if (dcmConfig.getNumWorkers() > 0
					&& dcmConfig.getNumWorkers() < numWorkers) {

				numWorkers = dcmConfig.getNumWorkers();
				List<Set<IInputJob>> splitTasks = optimizeWorkload(locations,