SAXException, RaidConfigurationException,
ClassNotFoundException, ParserConfigurationException, JSONException {
// find the matched raid policy
String policyName = args[startIndex++];
ConfigManager configManager = new ConfigManager(conf);
PolicyInfo policy = configManager.getPolicy(policyName);
if (policy == null) {
System.err.println ("Invalid policy: " + policyName);
return -1;
}
Codec codec = Codec.getCodec(policy.getCodecId());
if (codec == null) {
System.err.println("Policy " + policyName
+ " with invalid codec " + policy.getCodecId());
}
// find the matched paths to raid
FileSystem fs = FileSystem.get(conf);
List<FileStatus> pathsToRaid = new ArrayList<FileStatus>();
List<Path> policySrcPaths = policy.getSrcPathExpanded();
for (int i = startIndex; i< args.length; i++) {
boolean invalidPathToRaid = true;
Path pathToRaid = new Path(args[i]).makeQualified(fs);
String pathToRaidStr = pathToRaid.toString();
if (!pathToRaidStr.endsWith(Path.SEPARATOR)) {
pathToRaidStr = pathToRaidStr.concat(Path.SEPARATOR);
}
for (Path srcPath : policySrcPaths) {
String srcStr = srcPath.toString();
if (!srcStr.endsWith(Path.SEPARATOR)) {
srcStr = srcStr.concat(Path.SEPARATOR);
}
if (pathToRaidStr.startsWith(srcStr)) {
if (codec.isDirRaid) {
FileUtil.listStatusForLeafDir(
fs, fs.getFileStatus(pathToRaid), pathsToRaid);
} else {
FileUtil.listStatusHelper(fs, pathToRaid,
Integer.MAX_VALUE, pathsToRaid);
}
invalidPathToRaid = false;
break;
}
}
if (invalidPathToRaid) {
System.err.println("Path " + pathToRaidStr +
" does not support by the given policy " + policyName);
}
}
// Check if files are valid
List<FileStatus> validPaths = new ArrayList<FileStatus>();
List<PolicyInfo> policyInfos = new ArrayList<PolicyInfo>(1);
policyInfos.add(policy);
RaidState.Checker checker = new RaidState.Checker(
policyInfos, conf);
long now = System.currentTimeMillis();
for (FileStatus fileStatus : pathsToRaid) {
FileStatus[] dirStats = null;
if (codec.isDirRaid) {
dirStats = fs.listStatus(fileStatus.getPath());
}
RaidState stat = checker.check(
policy, fileStatus, now, false,
dirStats == null ? null : Arrays.asList(dirStats));
if (stat == RaidState.NOT_RAIDED_BUT_SHOULD) {
validPaths.add(fileStatus);
} else {
System.err.println("Path " + fileStatus.getPath() +
" is not qualified for raiding: " + stat);
}
}
if (validPaths.isEmpty()) {
System.err.println("No file can be raided");
return 0;
}
DistRaid dr = new DistRaid(conf);
//add paths for distributed raiding
List<EncodingCandidate> validEC =
RaidNode.splitPaths(conf, Codec.getCodec(policy.getCodecId()), validPaths);
dr.addRaidPaths(policy, validEC);
if (dr.startDistRaid()) {
System.out.println("Job started: " + dr.getJobTrackingURL());
System.out.print("Job in progress ");