diff --git a/CHANGES.txt b/CHANGES.txt
index f238271..54e1064 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -21,6 +21,9 @@
     MAPREDUCE-1970.  Reed-Solomon code implementation for HDFS RAID.
     (Scott Chen via dhruba)
 
+    MAPREDUCE-2169. Integrated Reed-Solomon code with RaidNode. (Ramkumar
+    Vadali via schen)
+
   IMPROVEMENTS
 
     MAPREDUCE-2141. Add an "extra data" field to Task for use by Mesos. (matei)
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/hdfs/DistributedRaidFileSystem.java b/src/contrib/raid/src/java/org/apache/hadoop/hdfs/DistributedRaidFileSystem.java
index fd1c33e..fffffb7 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/hdfs/DistributedRaidFileSystem.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/hdfs/DistributedRaidFileSystem.java
@@ -20,26 +20,33 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.DataInput;
+import java.io.PrintStream;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Random;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.ChecksumException;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FilterFileSystem;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSInputStream;
+import org.apache.hadoop.hdfs.BlockMissingException;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.util.Progressable;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.raid.Decoder;
 import org.apache.hadoop.raid.RaidNode;
+import org.apache.hadoop.raid.ReedSolomonDecoder;
 import org.apache.hadoop.raid.XORDecoder;
-import org.apache.hadoop.hdfs.BlockMissingException;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.raid.protocol.PolicyInfo.ErasureCodeType;
 
 /**
  * This is an implementation of the Hadoop  RAID Filesystem. This FileSystem 
@@ -51,7 +58,7 @@
 public class DistributedRaidFileSystem extends FilterFileSystem {
 
   // these are alternate locations that can be used for read-only access
-  Path[]     alternates;
+  DecodeInfo[] alternates;
   Configuration conf;
   int stripeLength;
 
@@ -64,6 +71,30 @@
     stripeLength = 0;
   }
 
+  // Information required for decoding a source file
+  static private class DecodeInfo {
+    final Path destPath;
+    final ErasureCodeType type;
+    final Configuration conf;
+    final int stripeLength;
+    private DecodeInfo(Configuration conf, ErasureCodeType type, Path destPath) {
+      this.conf = conf;
+      this.type = type;
+      this.destPath = destPath;
+      this.stripeLength = RaidNode.getStripeLength(conf);
+    }
+
+    Decoder createDecoder() {
+      if (this.type == ErasureCodeType.XOR) {
+        return new XORDecoder(conf, stripeLength);
+      } else if (this.type == ErasureCodeType.RS) {
+        return new ReedSolomonDecoder(conf, stripeLength,
+                              RaidNode.rsParityLength(conf));
+      }
+      return null;
+    }
+  }
+
   /* Initialize a Raid FileSystem
    */
   public void initialize(URI name, Configuration conf) throws IOException {
@@ -78,23 +109,6 @@
     this.fs = (FileSystem)ReflectionUtils.newInstance(clazz, null); 
     super.initialize(name, conf);
     
-    String alt = conf.get("hdfs.raid.locations");
-    
-    // If no alternates are specified, then behave absolutely same as 
-    // the original file system.
-    if (alt == null || alt.length() == 0) {
-      LOG.info("hdfs.raid.locations not defined. Using defaults...");
-      alt = RaidNode.DEFAULT_RAID_LOCATION;
-    }
-
-    // fs.alternate.filesystem.prefix can be of the form:
-    // "hdfs://host:port/myPrefixPath, file:///localPrefix,hftp://host1:port1/"
-    String[] strs  = alt.split(",");
-    if (strs == null || strs.length == 0) {
-      LOG.info("hdfs.raid.locations badly defined. Ignoring...");
-      return;
-    }
-
     // find stripe length configured
     stripeLength = RaidNode.getStripeLength(conf);
     if (stripeLength == 0) {
@@ -103,12 +117,12 @@
       return;
     }
 
-    // create a reference to all underlying alternate path prefix
-    alternates = new Path[strs.length];
-    for (int i = 0; i < strs.length; i++) {
-      alternates[i] = new Path(strs[i].trim());
-      alternates[i] = alternates[i].makeQualified(fs);
-    }
+    // Put XOR and RS in alternates
+    alternates= new DecodeInfo[2];
+    Path xorPath = RaidNode.xorDestinationPath(conf, fs);
+    alternates[0] = new DecodeInfo(conf, ErasureCodeType.XOR, xorPath);
+    Path rsPath = RaidNode.rsDestinationPath(conf, fs);
+    alternates[1] = new DecodeInfo(conf, ErasureCodeType.RS, rsPath);
   }
 
   /*
@@ -153,13 +167,13 @@
       private int nextLocation;
       private DistributedRaidFileSystem lfs;
       private Path path;
-      private final Path[] alternates;
+      private final DecodeInfo[] alternates;
       private final int buffersize;
       private final Configuration conf;
       private final int stripeLength;
 
-      ExtFsInputStream(Configuration conf, DistributedRaidFileSystem lfs, Path[] alternates,
-                       Path path, int stripeLength, int buffersize)
+      ExtFsInputStream(Configuration conf, DistributedRaidFileSystem lfs,
+          DecodeInfo[] alternates, Path path, int stripeLength, int buffersize)
           throws IOException {
         this.underLyingStream = lfs.fs.open(path, buffersize);
         this.path = path;
@@ -345,10 +359,13 @@
             clientConf.set("fs.hdfs.impl", clazz.getName());
             // Disable caching so that a previously cached RaidDfs is not used.
             clientConf.setBoolean("fs.hdfs.impl.disable.cache", true);
-            Decoder decoder =
-              new XORDecoder(clientConf, RaidNode.getStripeLength(clientConf));
-            Path npath = RaidNode.unRaid(clientConf, path, alternates[idx],
-                              decoder, stripeLength, corruptOffset);
+            Path npath = RaidNode.unRaid(clientConf, path,
+                         alternates[idx].destPath,
+                         alternates[idx].createDecoder(),
+                         stripeLength, corruptOffset);
+            if (npath == null)
+              continue;
+
             FileSystem fs1 = getUnderlyingFileSystem(conf);
             fs1.initialize(npath.toUri(), conf);
             LOG.info("Opening alternate path " + npath + " at offset " + curpos);
@@ -392,7 +409,7 @@
      * @throws IOException
      */
     public ExtFSDataInputStream(Configuration conf, DistributedRaidFileSystem lfs,
-      Path[] alternates, Path  p, int stripeLength, int buffersize) throws IOException {
+      DecodeInfo[] alternates, Path  p, int stripeLength, int buffersize) throws IOException {
         super(new ExtFsInputStream(conf, lfs, alternates, p, stripeLength, buffersize));
     }
   }
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java
index eafbba3..01497ad 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/BlockFixer.java
@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.raid;
 
-import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -60,6 +59,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.fs.ChecksumException;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileStatus;
@@ -71,36 +71,73 @@
 
 import org.apache.hadoop.raid.RaidNode;
 import org.apache.hadoop.raid.RaidUtils;
+import org.apache.hadoop.raid.protocol.PolicyInfo.ErasureCodeType;
+
 
 /**
- * This class fixes source file blocks using the parity file,
- * and parity file blocks using the source file.
- * It periodically fetches the list of corrupt files from the namenode,
- * and figures out the location of the bad block by reading through
- * the corrupt file.
+ * contains the core functionality of the block fixer
+ *
+ * raid.blockfix.interval          - interval between checks for corrupt files
+ *
+ * raid.blockfix.history.interval  - interval before fixing same file again
+ *
+ * raid.blockfix.read.timeout      - read time out
+ *
+ * raid.blockfix.write.timeout     - write time out
  */
-public class BlockFixer implements Runnable {
+public class BlockFixer extends Configured implements Runnable {
   public static final Log LOG = LogFactory.getLog(
                                   "org.apache.hadoop.raid.BlockFixer");
-  private java.util.HashMap<String, java.util.Date> history;
-  private int blockFixInterval = 60*1000; // 1min
-  private long numFilesFixed = 0;
-  private Configuration conf;
-  private String xorPrefix;
-  private XOREncoder xorEncoder;
-  private XORDecoder xorDecoder;
 
-  boolean running = true;
+  public static final String BLOCKFIX_INTERVAL = "raid.blockfix.interval";
+  public static final String BLOCKFIX_HISTORY_INTERVAL =
+    "raid.blockfix.history.interval";
+  public static final String BLOCKFIX_READ_TIMEOUT =
+    "raid.blockfix.read.timeout";
+  public static final String BLOCKFIX_WRITE_TIMEOUT =
+    "raid.blockfix.write.timeout";
+
+  public static final long DEFAULT_BLOCKFIX_INTERVAL = 60 * 1000; // 1 min
+  public static final long DEFAULT_BLOCKFIX_HISTORY_INTERVAL =
+    60 * 60 * 1000; // 60 mins
+
+  private java.util.HashMap<String, java.util.Date> history;
+  private long numFilesFixed = 0;
+  private String xorPrefix;
+  private String rsPrefix;
+  private Encoder xorEncoder;
+  private Decoder xorDecoder;
+  private Encoder rsEncoder;
+  private Decoder rsDecoder;
+
+  // interval between checks for corrupt files
+  protected long blockFixInterval = DEFAULT_BLOCKFIX_INTERVAL;
+
+  // interval before fixing same file again
+  protected long historyInterval = DEFAULT_BLOCKFIX_HISTORY_INTERVAL;
+
+  public volatile boolean running = true;
+
 
   public BlockFixer(Configuration conf) throws IOException {
-    this.conf = conf;
+    super(conf);
     history = new java.util.HashMap<String, java.util.Date>();
-    blockFixInterval = conf.getInt("raid.blockfix.interval",
-                                   blockFixInterval);
-    xorPrefix = RaidNode.getDestinationPath(conf).toUri().getPath();
-    int stripeLength = RaidNode.getStripeLength(conf);
-    xorEncoder = new XOREncoder(conf, stripeLength);
-    xorDecoder = new XORDecoder(conf, stripeLength);
+    blockFixInterval = getConf().getInt(BLOCKFIX_INTERVAL,
+                                   (int) blockFixInterval);
+    xorPrefix = RaidNode.xorDestinationPath(getConf()).toUri().getPath();
+    if (!xorPrefix.endsWith(Path.SEPARATOR)) {
+      xorPrefix += Path.SEPARATOR;
+    }
+    int stripeLength = RaidNode.getStripeLength(getConf());
+    xorEncoder = new XOREncoder(getConf(), stripeLength);
+    xorDecoder = new XORDecoder(getConf(), stripeLength);
+    rsPrefix = RaidNode.rsDestinationPath(getConf()).toUri().getPath();
+    if (!rsPrefix.endsWith(Path.SEPARATOR)) {
+      rsPrefix += Path.SEPARATOR;
+    }
+    int parityLength = RaidNode.rsParityLength(getConf());
+    rsEncoder = new ReedSolomonEncoder(getConf(), stripeLength, parityLength);
+    rsDecoder = new ReedSolomonDecoder(getConf(), stripeLength, parityLength);
   }
 
   public void run() {
@@ -154,6 +191,7 @@
 
 
   void fixFile(Path srcPath) throws IOException {
+
     if (RaidNode.isParityHarPartFile(srcPath)) {
       processCorruptParityHarPartFile(srcPath);
       return;
@@ -165,24 +203,30 @@
       return;
     }
 
-    // The corrupted file is a source file
-
-    // Do we have a parity file for this file?
-    RaidNode.ParityFilePair ppair = null;
-    Decoder decoder = null;
-    Path destPath = null;
-    try {
-      destPath = RaidNode.getDestinationPath(conf);
-      ppair = RaidNode.getParityFile(destPath, srcPath, conf);
-      if (ppair != null) {
-        decoder = xorDecoder;
-      }
-    } catch (FileNotFoundException e) {
+    // The corrupted file is a ReedSolomon parity file
+    if (isRsParityFile(srcPath)) {
+      processCorruptParityFile(srcPath, rsEncoder);
+      return;
     }
+
+    // The corrupted file is a source file
+    RaidNode.ParityFilePair ppair =
+      RaidNode.xorParityForSource(srcPath, getConf());
+    Decoder decoder = null;
+    if (ppair != null) {
+      decoder = xorDecoder;
+    } else  {
+      ppair = RaidNode.rsParityForSource(srcPath, getConf());
+      if (ppair != null) {
+        decoder = rsDecoder;
+      }
+    }
+
     // If we have a parity file, process the file and fix it.
     if (ppair != null) {
-      processCorruptFile(srcPath, destPath, decoder);
+      processCorruptFile(srcPath, ppair, decoder);
     }
+
   }
 
   /**
@@ -193,8 +237,8 @@
    */
   void purgeHistory() {
     // Default history interval is 1 hour.
-    long historyInterval = conf.getLong(
-                             "raid.blockfix.history.interval", 3600*1000);
+    long historyInterval = getConf().getLong(
+                             BLOCKFIX_HISTORY_INTERVAL, 3600*1000);
     java.util.Date cutOff = new java.util.Date(
                                    System.currentTimeMillis()-historyInterval);
     List<String> toRemove = new java.util.ArrayList<String>();
@@ -217,14 +261,14 @@
   List<Path> getCorruptFiles() throws IOException {
     DistributedFileSystem dfs = getDFS(new Path("/"));
 
-    String[] nnCorruptFiles = RaidDFSUtil.getCorruptFiles(conf);
+    String[] nnCorruptFiles = RaidDFSUtil.getCorruptFiles(getConf());
     List<Path> corruptFiles = new LinkedList<Path>();
     for (String file: nnCorruptFiles) {
       if (!history.containsKey(file)) {
         corruptFiles.add(new Path(file));
       }
     }
-    RaidUtils.filterTrash(conf, corruptFiles);
+    RaidUtils.filterTrash(getConf(), corruptFiles);
     return corruptFiles;
   }
 
@@ -235,11 +279,11 @@
     // TODO: We should first fix the files that lose more blocks
     Comparator<Path> comp = new Comparator<Path>() {
       public int compare(Path p1, Path p2) {
-        if (isXorParityFile(p2)) {
+        if (isXorParityFile(p2) || isRsParityFile(p2)) {
           // If p2 is a parity file, p1 is smaller.
           return -1;
         }
-        if (isXorParityFile(p1)) {
+        if (isXorParityFile(p1) || isRsParityFile(p1)) {
           // If p1 is a parity file, p2 is smaller.
           return 1;
         }
@@ -250,21 +294,13 @@
     Collections.sort(files, comp);
   }
 
-
-  /**
-   * Returns a DistributedFileSystem hosting the path supplied.
-   */
-  private DistributedFileSystem getDFS(Path p) throws IOException {
-    return (DistributedFileSystem) p.getFileSystem(conf);
-  }
-
   /**
    * Reads through a corrupt source file fixing corrupt blocks on the way.
    * @param srcPath Path identifying the corrupt file.
    * @throws IOException
    */
-  void processCorruptFile(Path srcPath, Path destPath, Decoder decoder)
-      throws IOException {
+  void processCorruptFile(Path srcPath, RaidNode.ParityFilePair parityPair,
+      Decoder decoder) throws IOException {
     LOG.info("Processing corrupt file " + srcPath);
 
     DistributedFileSystem srcFs = getDFS(srcPath);
@@ -290,9 +326,6 @@
       localBlockFile.deleteOnExit();
 
       try {
-        RaidNode.ParityFilePair parityPair = RaidNode.getParityFile(
-            destPath, srcPath, conf);
-
         decoder.recoverBlockToFile(srcFs, srcPath, parityPair.getFileSystem(),
           parityPair.getPath(), blockSize, corruptOffset, localBlockFile,
           blockContentsSize);
@@ -313,6 +346,9 @@
     numFilesFixed++;
   }
 
+  /**
+   * checks whether file is xor parity file
+   */
   boolean isXorParityFile(Path p) {
     String pathStr = p.toUri().getPath();
     if (pathStr.contains(RaidNode.HAR_SUFFIX)) {
@@ -322,7 +358,25 @@
   }
 
   /**
-   * Reads through a parity file, fixing corrupt blocks on the way.
+   * checks whether file is rs parity file
+   */
+  boolean isRsParityFile(Path p) {
+    String pathStr = p.toUri().getPath();
+    if (pathStr.contains(RaidNode.HAR_SUFFIX)) {
+      return false;
+    }
+    return pathStr.startsWith(rsPrefix);
+  }
+
+  /**
+   * Returns a DistributedFileSystem hosting the path supplied.
+   */
+  protected DistributedFileSystem getDFS(Path p) throws IOException {
+    return (DistributedFileSystem) p.getFileSystem(getConf());
+  }
+
+  /**
+   * Fixes corrupt blocks in a parity file.
    * This function uses the corresponding source file to regenerate parity
    * file blocks.
    */
@@ -469,6 +523,8 @@
         Encoder encoder;
         if (isXorParityFile(parityFile)) {
           encoder = xorEncoder;
+        } else if (isRsParityFile(parityFile)) {
+          encoder = rsEncoder;
         } else {
           String msg = "Could not figure out parity file correctly";
           LOG.warn(msg);
@@ -605,7 +661,7 @@
     DataInputStream blockMetadata = null;
     try {
       blockContents = new FileInputStream(localBlockFile);
-      blockMetadata = computeMetadata(conf, blockContents);
+      blockMetadata = computeMetadata(getConf(), blockContents);
       blockContents.close();
       // Reopen
       blockContents = new FileInputStream(localBlockFile);
@@ -638,13 +694,13 @@
     InetSocketAddress target = NetUtils.createSocketAddr(datanode.name);
     Socket sock = SocketChannel.open().socket();
 
-    int readTimeout =  conf.getInt(
-        "raid.blockfix.read.timeout", HdfsConstants.READ_TIMEOUT);
+    int readTimeout = getConf().getInt(BLOCKFIX_READ_TIMEOUT,
+      HdfsConstants.READ_TIMEOUT);
     NetUtils.connect(sock, target, readTimeout);
     sock.setSoTimeout(readTimeout);
 
-    int writeTimeout = conf.getInt(
-        "raid.blockfix.write.timeout", HdfsConstants.WRITE_TIMEOUT);
+    int writeTimeout = getConf().getInt(BLOCKFIX_WRITE_TIMEOUT,
+      HdfsConstants.WRITE_TIMEOUT);
 
     OutputStream baseStream = NetUtils.getOutputStream(sock, writeTimeout);
     DataOutputStream out = new DataOutputStream(
@@ -685,11 +741,18 @@
     }
   }
 
+  /**
+   * returns the source file corresponding to a parity file
+   */
   Path sourcePathFromParityPath(Path parityPath) {
     String parityPathStr = parityPath.toUri().getPath();
     if (parityPathStr.startsWith(xorPrefix)) {
       // Remove the prefix to get the source file.
-      String src = parityPathStr.replaceFirst(xorPrefix, "");
+      String src = parityPathStr.replaceFirst(xorPrefix, "/");
+      return new Path(src);
+    } else if (parityPathStr.startsWith(rsPrefix)) {
+      // Remove the prefix to get the source file.
+      String src = parityPathStr.replaceFirst(rsPrefix, "/");
       return new Path(src);
     }
     return null;
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/ConfigManager.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/ConfigManager.java
index deb9865..6ae6044 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/ConfigManager.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/ConfigManager.java
@@ -22,7 +22,9 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -146,8 +148,9 @@
    * 
    <configuration>
     <srcPath prefix="hdfs://hadoop.myhost.com:9000/user/warehouse/u_full/*">
-      <destPath> hdfs://dfsarch.data.facebook.com:9000/archive/</destPath>
       <policy name = RaidScanWeekly>
+        <destPath> hdfs://dfsname.myhost.com:9000/archive/</destPath>
+        <parentPolicy> RaidScanMonthly</parentPolicy>
         <property>
           <name>targetReplication</name>
           <value>2</value>
@@ -220,6 +223,8 @@
           "top-level element not <configuration>");
     NodeList elements = root.getChildNodes();
 
+    Map<String, PolicyInfo> existingPolicies =
+      new HashMap<String, PolicyInfo>();
     // loop through all the configured source paths.
     for (int i = 0; i < elements.getLength(); i++) {
       Node node = elements.item(i);
@@ -231,15 +236,15 @@
       if ("srcPath".equalsIgnoreCase(elementTagName)) {
         String srcPathPrefix = element.getAttribute("prefix");
 
-        if (srcPathPrefix == null || srcPathPrefix.length() == 0) {
-          throw new RaidConfigurationException("Bad configuration file: " + 
-             "srcPathPrefix not set.");
+        PolicyList policyList = null;
+        if (srcPathPrefix != null && srcPathPrefix.length() != 0) {
+          // Empty srcPath will have no effect but policies will be processed
+          // This allow us to define some "abstract" policies
+          policyList = new PolicyList();
+          all.add(policyList);
+          policyList.setSrcPath(conf, srcPathPrefix);
         }
-        PolicyList policyList = new PolicyList();
-        all.add(policyList);
 
-        policyList.setSrcPath(conf, srcPathPrefix);
-        
         // loop through all the policies for this source path
         NodeList policies = element.getChildNodes();
         for (int j = 0; j < policies.getLength(); j++) {
@@ -253,12 +258,13 @@
               "Expecting <policy> for srcPath " + srcPathPrefix);
           }
           String policyName = policy.getAttribute("name");
-          PolicyInfo pinfo = new PolicyInfo(policyName, conf);
-          pinfo.setSrcPath(srcPathPrefix);
-          policyList.add(pinfo);
-
+          PolicyInfo curr = new PolicyInfo(policyName, conf);
+          if (srcPathPrefix != null && srcPathPrefix.length() > 0) {
+            curr.setSrcPath(srcPathPrefix);
+          }
           // loop through all the properties of this policy
           NodeList properties = policy.getChildNodes();
+          PolicyInfo parent = null;
           for (int k = 0; k < properties.getLength(); k++) {
             Node node2 = properties.item(k);
             if (!(node2 instanceof Element)) {
@@ -266,13 +272,16 @@
             }
             Element property = (Element)node2;
             String propertyName = property.getTagName();
-            if ("destPath".equalsIgnoreCase(propertyName)) {
+            if ("erasureCode".equalsIgnoreCase(propertyName)) {
               String text = ((Text)property.getFirstChild()).getData().trim();
-              LOG.info(policyName + ".destPath = " + text);
-              pinfo.setDestinationPath(text);
+              LOG.info(policyName + ".erasureCode = " + text);
+              curr.setErasureCode(text);
             } else if ("description".equalsIgnoreCase(propertyName)) {
               String text = ((Text)property.getFirstChild()).getData().trim();
-              pinfo.setDescription(text);
+              curr.setDescription(text);
+            } else if ("parentPolicy".equalsIgnoreCase(propertyName)) {
+              String text = ((Text)property.getFirstChild()).getData().trim();
+              parent = existingPolicies.get(text);
             } else if ("property".equalsIgnoreCase(propertyName)) {
               NodeList nl = property.getChildNodes();
               String pname=null,pvalue=null;
@@ -291,7 +300,7 @@
               }
               if (pname != null && pvalue != null) {
                 LOG.info(policyName + "." + pname + " = " + pvalue);
-                pinfo.setProperty(pname,pvalue);
+                curr.setProperty(pname,pvalue);
               }
             } else {
               LOG.info("Found bad property " + propertyName +
@@ -300,6 +309,20 @@
                        ". Ignoring."); 
             }
           }  // done with all properties of this policy
+
+          PolicyInfo pinfo;
+          if (parent != null) {
+            pinfo = new PolicyInfo(policyName, conf);
+            pinfo.copyFrom(parent);
+            pinfo.copyFrom(curr);
+          } else {
+            pinfo = curr;
+          }
+          if (policyList != null) {
+            policyList.add(pinfo);
+          }
+          existingPolicies.put(policyName, pinfo);
+
         }    // done with all policies for this srcpath
       } 
     }        // done with all srcPaths
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/DistRaid.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/DistRaid.java
index e87c2ff..bc06606 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/DistRaid.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/DistRaid.java
@@ -335,9 +335,6 @@
          LOG.info("Job Complete(Failed): " + jobID);
        }
        raidPolicyPathPairList.clear();
-       Counters ctrs = runningJob.getCounters();
-       long filesRaided = ctrs.findCounter(Counter.FILES_SUCCEEDED).getValue();
-       long filesFailed = ctrs.findCounter(Counter.FILES_FAILED).getValue();
        return true;
      } else {
        String report =  (" job " + jobID +
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/Encoder.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/Encoder.java
index 350ee78..f74b3a0 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/Encoder.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/Encoder.java
@@ -107,7 +107,8 @@
    * @param srcFile The source file.
    * @param parityFile The parity file to be generated.
    */
-  public void encodeFile(FileSystem fs, Path srcFile, Path parityFile,
+  public void encodeFile(
+    FileSystem fs, Path srcFile, FileSystem parityFs, Path parityFile,
     short parityRepl, Progressable reporter) throws IOException {
     FileStatus srcStat = fs.getFileStatus(srcFile);
     long srcSize = srcStat.getLen();
@@ -116,10 +117,13 @@
     configureBuffers(blockSize);
 
     // Create a tmp file to which we will write first.
-    Path parityTmp = new Path(conf.get("fs.raid.tmpdir", "/tmp/raid") +
-                              parityFile.toUri().getPath() +
-                              "." + rand.nextLong() + ".tmp");
-    FSDataOutputStream out = fs.create(
+    Path tmpDir = getParityTempPath();
+    if (!parityFs.mkdirs(tmpDir)) {
+      throw new IOException("Could not create tmp dir " + tmpDir);
+    }
+    Path parityTmp = new Path(tmpDir,
+                            parityFile.getName() + rand.nextLong());
+    FSDataOutputStream out = parityFs.create(
                                parityTmp,
                                true,
                                conf.getInt("io.file.buffer.size", 64 * 1024),
@@ -133,11 +137,11 @@
       LOG.info("Wrote temp parity file " + parityTmp);
 
       // delete destination if exists
-      if (fs.exists(parityFile)){
-        fs.delete(parityFile, false);
+      if (parityFs.exists(parityFile)){
+        parityFs.delete(parityFile, false);
       }
-      fs.mkdirs(parityFile.getParent());
-      if (!fs.rename(parityTmp, parityFile)) {
+      parityFs.mkdirs(parityFile.getParent());
+      if (!parityFs.rename(parityTmp, parityFile)) {
         String msg = "Unable to rename file " + parityTmp + " to " + parityFile;
         throw new IOException (msg);
       }
@@ -146,7 +150,7 @@
       if (out != null) {
         out.close();
       }
-      fs.delete(parityTmp, false);
+      parityFs.delete(parityTmp, false);
     }
   }
 
@@ -225,7 +229,7 @@
     LOG.info("Starting recovery by using source stripe " +
               srcFile + ":" + stripeStart);
     // Read the data from the blocks and write to the parity file.
-    encodeStripe(blocks, stripeStart, blockSize, outs, 
+    encodeStripe(blocks, stripeStart, blockSize, outs,
                  new RaidUtils.DummyProgressable());
   }
 
@@ -339,4 +343,9 @@
     long blockSize,
     OutputStream[] outs,
     Progressable reporter) throws IOException;
+
+  /**
+   * Return the temp path for the parity file
+   */
+  protected abstract Path getParityTempPath();
 }
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidFilter.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidFilter.java
index 2a660b4..2af4ed9 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidFilter.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidFilter.java
@@ -218,4 +218,42 @@
       return len;
     }
   }
+
+  static class PreferenceFilter extends Configured
+    implements DirectoryTraversal.FileFilter {
+    Path firstChoicePrefix;
+    DirectoryTraversal.FileFilter secondChoiceFilter;
+
+    PreferenceFilter(Configuration conf,
+      Path firstChoicePrefix, Path secondChoicePrefix,
+      int targetRepl, long startTime, long modTimePeriod) {
+      super(conf);
+      this.firstChoicePrefix = firstChoicePrefix;
+      this.secondChoiceFilter = new TimeBasedFilter(conf,
+        secondChoicePrefix, targetRepl, startTime, modTimePeriod);
+    }
+
+    PreferenceFilter(Configuration conf,
+      Path firstChoicePrefix, Path secondChoicePrefix,
+      PolicyInfo info, List<PolicyInfo> allPolicies, long startTime,
+        Statistics stats) {
+      super(conf);
+      this.firstChoicePrefix = firstChoicePrefix;
+      this.secondChoiceFilter = new TimeBasedFilter(
+        conf, secondChoicePrefix, info, allPolicies, startTime, stats);
+    }
+
+    public boolean check(FileStatus f) throws IOException {
+      Object firstChoicePPair =
+        RaidNode.getParityFile(firstChoicePrefix, f.getPath(), getConf());
+      if (firstChoicePPair == null) {
+        // The decision is upto the the second choice filter.
+        return secondChoiceFilter.check(f);
+      } else {
+        // There is already a parity file under the first choice path.
+        // We dont want to choose this file.
+        return false;
+      }
+    }
+  }
 }
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
index 76cd389..db7b86e 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
@@ -20,12 +20,12 @@
 
 import java.io.IOException;
 import java.io.FileNotFoundException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.LinkedList;
-import java.util.Iterator;
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Random;
@@ -63,6 +63,7 @@
 import org.apache.hadoop.raid.protocol.PolicyInfo;
 import org.apache.hadoop.raid.protocol.PolicyList;
 import org.apache.hadoop.raid.protocol.RaidProtocol;
+import org.apache.hadoop.raid.protocol.PolicyInfo.ErasureCodeType;
 
 /**
  * A base class that implements {@link RaidProtocol}.
@@ -77,19 +78,35 @@
     Configuration.addDefaultResource("mapred-default.xml");
     Configuration.addDefaultResource("mapred-site.xml");
   }
-
   public static final Log LOG = LogFactory.getLog( "org.apache.hadoop.raid.RaidNode");
   public static final long SLEEP_TIME = 10000L; // 10 seconds
   public static final int DEFAULT_PORT = 60000;
-  public static final int DEFAULT_STRIPE_LENGTH = 5; // default value of stripe length
+  // Default stripe length = 5, parity length for RS code = 3
+  public static final int DEFAULT_STRIPE_LENGTH = 5;
+  public static final int RS_PARITY_LENGTH_DEFAULT = 3;
+
+  public static final String RS_PARITY_LENGTH_KEY = "hdfs.raidrs.paritylength";
   public static final String STRIPE_LENGTH_KEY = "hdfs.raid.stripeLength";
+
   public static final String DEFAULT_RAID_LOCATION = "/raid";
   public static final String RAID_LOCATION_KEY = "hdfs.raid.locations";
+  public static final String DEFAULT_RAID_TMP_LOCATION = "/tmp/raid";
+  public static final String RAID_TMP_LOCATION_KEY = "fs.raid.tmpdir";
+  public static final String DEFAULT_RAID_HAR_TMP_LOCATION = "/tmp/raid_har";
+  public static final String RAID_HAR_TMP_LOCATION_KEY = "fs.raid.hartmpdir";
+
+  public static final String DEFAULT_RAIDRS_LOCATION = "/raidrs";
+  public static final String RAIDRS_LOCATION_KEY = "hdfs.raidrs.locations";
+  public static final String DEFAULT_RAIDRS_TMP_LOCATION = "/tmp/raidrs";
+  public static final String RAIDRS_TMP_LOCATION_KEY = "fs.raidrs.tmpdir";
+  public static final String DEFAULT_RAIDRS_HAR_TMP_LOCATION = "/tmp/raidrs_har";
+  public static final String RAIDRS_HAR_TMP_LOCATION_KEY = "fs.raidrs.hartmpdir";
+
   public static final String HAR_SUFFIX = "_raid.har";
   public static final Pattern PARITY_HAR_PARTFILE_PATTERN =
     Pattern.compile(".*" + HAR_SUFFIX + "/part-.*");
-  
-  public static final String RAIDNODE_CLASSNAME_KEY = "raid.classname";  
+
+  public static final String RAIDNODE_CLASSNAME_KEY = "raid.classname";
 
   /** RPC server */
   private Server server;
@@ -131,11 +148,11 @@
     long metaSize;           // total disk space for meta files
 
     public void clear() {
-    	numProcessedBlocks = 0;
-    	processedSize = 0;
-    	remainingSize = 0;
-    	numMetaBlocks = 0;
-    	metaSize = 0;
+      numProcessedBlocks = 0;
+      processedSize = 0;
+      remainingSize = 0;
+      numMetaBlocks = 0;
+      metaSize = 0;
     }
     public String toString() {
       long save = processedSize - (remainingSize + metaSize);
@@ -310,13 +327,24 @@
     // find stripe length from config
     int stripeLength = getStripeLength(conf);
 
-    Path destPref = getDestinationPath(conf);
+    // first try decode using XOR code
+    Path destPref = xorDestinationPath(conf);
     Decoder decoder = new XORDecoder(conf, RaidNode.getStripeLength(conf));
     Path unraided = unRaid(conf, srcPath, destPref, decoder,
         stripeLength, corruptOffset);
     if (unraided != null) {
       return unraided.toString();
     }
+
+    // try decode using ReedSolomon code
+    destPref = rsDestinationPath(conf);
+    decoder = new ReedSolomonDecoder(conf, RaidNode.getStripeLength(conf),
+        RaidNode.rsParityLength(conf));
+    unraided = unRaid(conf, srcPath, destPref, decoder,
+        stripeLength, corruptOffset);
+    if (unraided != null) {
+      return unraided.toString();
+    }
     return null;
   }
 
@@ -329,7 +357,6 @@
    * Periodically checks to see which policies should be fired.
    */
   class TriggerMonitor implements Runnable {
-
     class ScanState {
       long fullScanStartTime;
       DirectoryTraversal pendingTraversal;
@@ -393,7 +420,7 @@
     */
     private List<FileStatus> selectFiles(
       PolicyInfo info, ArrayList<PolicyInfo> allPolicies) throws IOException {
-      Path destPrefix = getDestinationPath(conf);
+      Path destPrefix = getDestinationPath(info.getErasureCode(), conf);
       String policyName = info.getName();
       Path srcPath = info.getSrcPath();
       long modTimePeriod = Long.parseLong(info.getProperty("modTimePeriod"));
@@ -480,7 +507,6 @@
             }
           }
         }
-
         for (PolicyInfo info: allPolicies) {
           if (!scanStateMap.containsKey(info.getName())) {
             scanStateMap.put(info.getName(), new ScanState());
@@ -526,8 +552,19 @@
       long startTime, PolicyInfo info, List<PolicyInfo> allPolicies,
       RaidFilter.Statistics stats)
       throws IOException {
-      return new RaidFilter.TimeBasedFilter(conf, getDestinationPath(conf),
-        info, allPolicies, startTime, stats);
+      switch (info.getErasureCode()) {
+      case XOR:
+        // Return a preference-based filter that prefers RS parity files
+        // over XOR parity files.
+        return new RaidFilter.PreferenceFilter(
+          conf, rsDestinationPath(conf), xorDestinationPath(conf),
+          info, allPolicies, startTime, stats);
+      case RS:
+        return new RaidFilter.TimeBasedFilter(conf, rsDestinationPath(conf),
+          info, allPolicies, startTime, stats);
+      default:
+        return null;
+      }
     }
   }
 
@@ -623,13 +660,31 @@
 
     return null; // NULL if no parity file
   }
-  
-  private ParityFilePair getParityFile(Path destPathPrefix, Path srcPath) throws IOException {
-    
-    return getParityFile(destPathPrefix, srcPath, conf);
-    
+
+  static ParityFilePair xorParityForSource(Path srcPath, Configuration conf)
+      throws IOException {
+    try {
+      Path destPath = xorDestinationPath(conf);
+      return getParityFile(destPath, srcPath, conf);
+    } catch (FileNotFoundException e) {
+    }
+    return null;
   }
-  
+
+  static ParityFilePair rsParityForSource(Path srcPath, Configuration conf)
+      throws IOException {
+    try {
+      Path destPath = rsDestinationPath(conf);
+      return getParityFile(destPath, srcPath, conf);
+    } catch (FileNotFoundException e) {
+    }
+    return null;
+  }
+
+  private ParityFilePair getParityFile(Path destPathPrefix, Path srcPath)
+      throws IOException {
+    return getParityFile(destPathPrefix, srcPath, conf);
+  }
 
   /**
    * RAID a list of files.
@@ -639,7 +694,7 @@
     int targetRepl = Integer.parseInt(info.getProperty("targetReplication"));
     int metaRepl = Integer.parseInt(info.getProperty("metaReplication"));
     int stripeLength = getStripeLength(conf);
-    Path destPref = getDestinationPath(conf);
+    Path destPref = getDestinationPath(info.getErasureCode(), conf);
     String simulate = info.getProperty("simulate");
     boolean doSimulate = simulate == null ? false : Boolean
         .parseBoolean(simulate);
@@ -648,8 +703,9 @@
     int count = 0;
 
     for (FileStatus s : paths) {
-      doRaid(conf, s, destPref, statistics, new RaidUtils.DummyProgressable(),
-             doSimulate, targetRepl, metaRepl, stripeLength);
+      doRaid(conf, s, destPref, info.getErasureCode(), statistics,
+        new RaidUtils.DummyProgressable(), doSimulate, targetRepl, metaRepl,
+        stripeLength);
       if (count % 1000 == 0) {
         LOG.info("RAID statistics " + statistics.toString());
       }
@@ -664,25 +720,28 @@
    */
 
   static public void doRaid(Configuration conf, PolicyInfo info,
-      FileStatus src, Statistics statistics, Progressable reporter) throws IOException {
+      FileStatus src, Statistics statistics, Progressable reporter)
+      throws IOException {
     int targetRepl = Integer.parseInt(info.getProperty("targetReplication"));
     int metaRepl = Integer.parseInt(info.getProperty("metaReplication"));
     int stripeLength = getStripeLength(conf);
-    Path destPref = getDestinationPath(conf);
+
+    Path destPref = getDestinationPath(info.getErasureCode(), conf);
     String simulate = info.getProperty("simulate");
     boolean doSimulate = simulate == null ? false : Boolean
         .parseBoolean(simulate);
 
-    doRaid(conf, src, destPref, statistics, reporter, doSimulate,
-           targetRepl, metaRepl, stripeLength);
+    doRaid(conf, src, destPref, info.getErasureCode(), statistics,
+                  reporter, doSimulate, targetRepl, metaRepl, stripeLength);
   }
 
   /**
    * RAID an individual file
    */
   static private void doRaid(Configuration conf, FileStatus stat, Path destPath,
-                      Statistics statistics, Progressable reporter, boolean doSimulate,
-                      int targetRepl, int metaRepl, int stripeLength) 
+                      PolicyInfo.ErasureCodeType code, Statistics statistics,
+                      Progressable reporter, boolean doSimulate,
+                      int targetRepl, int metaRepl, int stripeLength)
     throws IOException {
     Path p = stat.getPath();
     FileSystem srcFs = p.getFileSystem(conf);
@@ -704,7 +763,8 @@
     statistics.processedSize += diskSpace;
 
     // generate parity file
-    generateParityFile(conf, stat, reporter, srcFs, destPath, locations, metaRepl, stripeLength);
+    generateParityFile(conf, stat, reporter, srcFs, destPath, code,
+        locations, metaRepl, stripeLength);
 
     // reduce the replication factor of the source file
     if (!doSimulate) {
@@ -740,7 +800,9 @@
   static private void generateParityFile(Configuration conf, FileStatus stat,
                                   Progressable reporter,
                                   FileSystem inFs,
-                                  Path destPathPrefix, BlockLocation[] locations,
+                                  Path destPathPrefix,
+                                  ErasureCodeType code,
+                                  BlockLocation[] locations,
                                   int metaRepl, int stripeLength) throws IOException {
 
     Path inpath = stat.getPath();
@@ -751,16 +813,16 @@
     try {
       FileStatus stmp = outFs.getFileStatus(outpath);
       if (stmp.getModificationTime() == stat.getModificationTime()) {
-        LOG.info("Parity file for " + inpath + "(" + locations.length + ") is " + outpath +
-                 " already upto-date. Nothing more to do.");
+        LOG.info("Parity file for " + inpath + "(" + locations.length +
+              ") is " + outpath + " already upto-date. Nothing more to do.");
         return;
       }
     } catch (IOException e) {
       // ignore errors because the raid file might not exist yet.
     }
 
-    XOREncoder encoder = new XOREncoder(conf, stripeLength);
-    encoder.encodeFile(inFs, inpath, outpath, (short)metaRepl, reporter);
+    Encoder encoder = encoderForCode(conf, code);
+    encoder.encodeFile(inFs, inpath, outFs, outpath, (short)metaRepl, reporter);
 
     // set the modification time of the RAID file. This is done so that the modTime of the
     // RAID file reflects that contents of the source file that it has RAIDed. This should
@@ -792,8 +854,8 @@
       return null;
     }
 
-    final Path recoveryDestination = 
-      new Path(conf.get("fs.raid.tmpdir", "/tmp/raid"));
+    final Path recoveryDestination = new Path(
+      RaidNode.unraidTmpDirectory(conf));
     FileSystem destFs = recoveryDestination.getFileSystem(conf);
     final Path recoveredPrefix = 
       destFs.makeQualified(new Path(recoveryDestination, makeRelative(srcPath)));
@@ -827,6 +889,43 @@
     }
 
     /**
+     * Traverse the parity destination directory, removing directories that
+     * no longer existing in the source.
+     * @throws IOException
+     */
+    private void purgeDirectories(FileSystem fs, Path root) throws IOException {
+      String prefix = root.toUri().getPath();
+      List<FileStatus> startPaths = new LinkedList<FileStatus>();
+      try {
+        startPaths.add(fs.getFileStatus(root));
+      } catch (FileNotFoundException e) {
+        return;
+      }
+      DirectoryTraversal dt = new DirectoryTraversal(fs, startPaths);
+      FileStatus dir = dt.getNextDirectory();
+      for (; dir != null; dir = dt.getNextDirectory()) {
+        Path dirPath = dir.getPath();
+        if (dirPath.toUri().getPath().endsWith(HAR_SUFFIX)) {
+          continue;
+        }
+        String dirStr = dirPath.toUri().getPath();
+        if (!dirStr.startsWith(prefix)) {
+          continue;
+        }
+        String src = dirStr.replaceFirst(prefix, "");
+        if (src.length() == 0) continue;
+        Path srcPath = new Path(src);
+        if (!fs.exists(srcPath)) {
+          LOG.info("Purging directory " + dirPath);
+          boolean done = fs.delete(dirPath, true);
+          if (!done) {
+            LOG.error("Could not purge " + dirPath);
+          }
+        }
+      }
+    }
+
+    /**
      * Delete orphaned files. The reason this is done by a separate thread 
      * is to not burden the TriggerMonitor with scanning the 
      * destination directories.
@@ -844,13 +943,23 @@
         LOG.info("Started purge scan");
         prevExec = now();
 
+        // expand destination prefix path
+        Path destPref = xorDestinationPath(conf);
+        FileSystem destFs = destPref.getFileSystem(conf);
+        purgeDirectories(destFs, destPref);
+
+        destPref = rsDestinationPath(conf);
+        destFs = destPref.getFileSystem(conf);
+        purgeDirectories(destFs, destPref);
+
+        // fetch all categories
         for (PolicyList category : configMgr.getAllPolicies()) {
           for (PolicyInfo info: category.getAll()) {
 
             try {
               // expand destination prefix path
-              Path destPref = getDestinationPath(conf);
-              FileSystem destFs = destPref.getFileSystem(conf);
+              destPref = getDestinationPath(info.getErasureCode(), conf);
+              destFs = destPref.getFileSystem(conf);
 
               //get srcPaths
               Path[] srcPaths = info.getSrcPathExpanded();
@@ -870,7 +979,8 @@
                   if (stat != null) {
                     LOG.info("Purging obsolete parity files for policy " + 
                               info.getName() + " " + destPath);
-                    recursePurge(srcFs, destFs, destPref.toUri().getPath(), stat);
+                    recursePurge(info.getErasureCode(), srcFs, destFs,
+                              destPref.toUri().getPath(), stat);
                   }
 
                 }
@@ -889,7 +999,8 @@
      * The destPrefix is the absolute pathname of the destinationPath
      * specified in the policy (without the host:port)
      */ 
-    private void recursePurge(FileSystem srcFs, FileSystem destFs,
+    void recursePurge(ErasureCodeType code,
+                              FileSystem srcFs, FileSystem destFs,
                               String destPrefix, FileStatus dest) 
       throws IOException {
 
@@ -901,7 +1012,7 @@
       if (dest.isDirectory() && destStr.endsWith(HAR_SUFFIX)) {
         try {
           int harUsedPercent =
-            usefulHar(srcFs, destFs, destPath, destPrefix, conf);
+            usefulHar(code, srcFs, destFs, destPath, destPrefix, conf);
           LOG.info("Useful percentage of " + destStr + " " + harUsedPercent);
           // Delete the har if its usefulness reaches a threshold.
           if (harUsedPercent <= conf.getInt("raid.har.usage.threshold", 0)) {
@@ -940,7 +1051,7 @@
         }
         if (files != null) {
           for (FileStatus one:files) {
-            recursePurge(srcFs, destFs, destPrefix, one);
+            recursePurge(code, srcFs, destFs, destPrefix, one);
           }
         }
         // If the directory is empty now, it will be purged the next time this
@@ -950,12 +1061,40 @@
       
       String src = destStr.replaceFirst(destPrefix, "");
       
-      // if the source path does not exist or the parity file has been HARed, 
-      // then delete the parity file
       Path srcPath = new Path(src);
-      Path dstPath = (new Path(destPrefix.trim())).makeQualified(destFs);
-      if (!srcFs.exists(srcPath) || 
-          !destPath.equals(getParityFile(dstPath,srcPath).getPath())) {
+      boolean shouldDelete = false;
+
+      if (!srcFs.exists(srcPath)) {
+        shouldDelete = true;
+      } else {
+        try {
+          // If there is a RS parity file, the XOR parity can be deleted.
+          if (code == ErasureCodeType.XOR) {
+            ParityFilePair ppair = getParityFile(
+               getDestinationPath(ErasureCodeType.RS, conf), srcPath, conf);
+            if (ppair != null) {
+              shouldDelete = true;
+            }
+          }
+          if (!shouldDelete) {
+            Path dstPath = (new Path(destPrefix.trim())).makeQualified(destFs);
+            ParityFilePair ppair = getParityFile(dstPath,srcPath);
+            // If the parity file is not the appropriate one for the source or
+            // the parityFs is not the same as this file's filesystem
+            // (it is a HAR), this file can be deleted.
+            if ( ppair == null ||
+                 !destFs.equals(ppair.getFileSystem()) ||
+                 !destPath.equals(ppair.getPath())) {
+              shouldDelete = true;
+            }
+          }
+        } catch (IOException e) {
+          LOG.warn("Error during purging " + src + " " +
+                   StringUtils.stringifyException(e));
+        }
+      }
+
+      if (shouldDelete) {
         boolean done = destFs.delete(destPath, false);
         if (done) {
           LOG.info("Purged file " + destPath );
@@ -971,6 +1110,7 @@
   // total number of files in the har.
   //
   protected static int usefulHar(
+    ErasureCodeType code,
     FileSystem srcFs, FileSystem destFs,
     Path harPath, String destPrefix, Configuration conf) throws IOException {
 
@@ -996,6 +1136,15 @@
         continue;
       }
       String src = parityStr.substring(prefixToReplace.length());
+      if (code == ErasureCodeType.XOR) {
+        ParityFilePair ppair = getParityFile(
+          getDestinationPath(ErasureCodeType.RS, conf), new Path(src), conf);
+        if (ppair != null) {
+          // There is a valid RS parity file, so the XOR one is useless.
+          numUseless++;
+          continue;
+        }
+      }
       try {
         FileStatus srcStatus = srcFs.getFileStatus(new Path(src));
         if (srcStatus == null) {
@@ -1026,18 +1175,16 @@
       LOG.info("Started archive scan");
       prevExec = now();
 
+      // fetch all categories
       for (PolicyList category : configMgr.getAllPolicies()) {
         for (PolicyInfo info: category.getAll()) {
+          String tmpHarPath = tmpHarPathForCode(conf, info.getErasureCode());
           String str = info.getProperty("time_before_har");
-          String tmpHarPath = info.getProperty("har_tmp_dir");
-          if (tmpHarPath == null) {
-            tmpHarPath = "/tmp/raid_har";
-          }
           if (str != null) {
             try {
               long cutoff = now() - ( Long.parseLong(str) * 24L * 3600000L );
 
-              Path destPref = getDestinationPath(conf);
+              Path destPref = getDestinationPath(info.getErasureCode(), conf);
               FileSystem destFs = destPref.getFileSystem(conf); 
 
               //get srcPaths
@@ -1074,7 +1221,7 @@
     return;
   }
   
-  private void recurseHar(PolicyInfo info, FileSystem destFs, FileStatus dest,
+  void recurseHar(PolicyInfo info, FileSystem destFs, FileStatus dest,
     String destPrefix, FileSystem srcFs, long cutoff, String tmpHarPath)
     throws IOException {
 
@@ -1198,10 +1345,71 @@
   }
 
   /**
-   * Return the path prefix that stores the parity files
+   * Return the temp path for XOR parity files
    */
-  static Path getDestinationPath(Configuration conf)
-      throws IOException {
+  public static String unraidTmpDirectory(Configuration conf) {
+    return conf.get(RAID_TMP_LOCATION_KEY, DEFAULT_RAID_TMP_LOCATION);
+  }
+
+  /**
+   * Return the temp path for ReedSolomonEncoder parity files
+   */
+  public static String rsTempPrefix(Configuration conf) {
+    return conf.get(RAIDRS_TMP_LOCATION_KEY, DEFAULT_RAIDRS_TMP_LOCATION);
+  }
+
+  /**
+   * Return the temp path for XOR parity files
+   */
+  public static String xorHarTempPrefix(Configuration conf) {
+    return conf.get(RAID_HAR_TMP_LOCATION_KEY, DEFAULT_RAID_HAR_TMP_LOCATION);
+  }
+
+  /**
+   * Return the temp path for ReedSolomonEncoder parity files
+   */
+  public static String rsHarTempPrefix(Configuration conf) {
+    return conf.get(RAIDRS_HAR_TMP_LOCATION_KEY,
+        DEFAULT_RAIDRS_HAR_TMP_LOCATION);
+  }
+
+  /**
+   * Return the destination path for ReedSolomon parity files
+   */
+  public static Path rsDestinationPath(Configuration conf, FileSystem fs) {
+    String loc = conf.get(RAIDRS_LOCATION_KEY, DEFAULT_RAIDRS_LOCATION);
+    Path p = new Path(loc.trim());
+    p = p.makeQualified(fs);
+    return p;
+  }
+
+  /**
+   * Return the destination path for ReedSolomon parity files
+   */
+  public static Path rsDestinationPath(Configuration conf)
+    throws IOException {
+    String loc = conf.get(RAIDRS_LOCATION_KEY, DEFAULT_RAIDRS_LOCATION);
+    Path p = new Path(loc.trim());
+    FileSystem fs = FileSystem.get(p.toUri(), conf);
+    p = p.makeQualified(fs);
+    return p;
+  }
+
+  /**
+   * Return the destination path for XOR parity files
+   */
+  public static Path xorDestinationPath(Configuration conf, FileSystem fs) {
+    String loc = conf.get(RAID_LOCATION_KEY, DEFAULT_RAID_LOCATION);
+    Path p = new Path(loc.trim());
+    p = p.makeQualified(fs);
+    return p;
+  }
+
+  /**
+   * Return the destination path for XOR parity files
+   */
+  public static Path xorDestinationPath(Configuration conf)
+    throws IOException {
     String loc = conf.get(RAID_LOCATION_KEY, DEFAULT_RAID_LOCATION);
     Path p = new Path(loc.trim());
     FileSystem fs = FileSystem.get(p.toUri(), conf);
@@ -1210,12 +1418,57 @@
   }
 
   /**
+   * Return the path prefix that stores the parity files
+   */
+  static Path getDestinationPath(ErasureCodeType code, Configuration conf)
+    throws IOException {
+    switch (code) {
+      case XOR:
+        return xorDestinationPath(conf);
+      case RS:
+        return rsDestinationPath(conf);
+      default:
+        return null;
+    }
+  }
+
+  static Encoder encoderForCode(Configuration conf, ErasureCodeType code) {
+    int stripeLength = getStripeLength(conf);
+    switch (code) {
+      case XOR:
+        return new XOREncoder(conf, stripeLength);
+      case RS:
+        return new ReedSolomonEncoder(conf, stripeLength, rsParityLength(conf));
+      default:
+        return null;
+    }
+  }
+
+  static String tmpHarPathForCode(Configuration conf, ErasureCodeType code) {
+    switch (code) {
+      case XOR:
+        return xorHarTempPrefix(conf);
+      case RS:
+        return rsHarTempPrefix(conf);
+      default:
+        return null;
+    }
+  }
+
+  /**
    * Obtain stripe length from configuration
    */
   public static int getStripeLength(Configuration conf) {
     return conf.getInt(STRIPE_LENGTH_KEY, DEFAULT_STRIPE_LENGTH);
   }
 
+  /**
+   * Obtain stripe length from configuration
+   */
+  public static int rsParityLength(Configuration conf) {
+    return conf.getInt(RS_PARITY_LENGTH_KEY, RS_PARITY_LENGTH_DEFAULT);
+  }
+
   static boolean isParityHarPartFile(Path p) {
     Matcher m = PARITY_HAR_PARTFILE_PATTERN.matcher(p.toUri().getPath());
     return m.matches();
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java
index 5619566..7ce0c82 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidUtils.java
@@ -39,7 +39,7 @@
    *
    * We could have used Reporter.NULL here but that would introduce
    * a dependency on mapreduce.
-   */ 
+   */
   public static class DummyProgressable implements Progressable {
     /**
      * Do nothing.
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonCode.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonCode.java
index 88d0b51..0a6050a 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonCode.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonCode.java
@@ -22,9 +22,9 @@
 
   private final int stripeSize;
   private final int paritySize;
-  private final int[] generatingRoots;
   private final int[] generatingPolynomial;
   private final int PRIMITIVE_ROOT = 2;
+  private final int[] primitivePower;
   private final GaloisField GF = new GaloisField();
   private int[] errSignature;
   private final int[] paritySymbolLocations;
@@ -40,14 +40,17 @@
     for (int i = 0; i < paritySize; i++) {
       paritySymbolLocations[i] = i;
     }
-    this.generatingRoots = new int[paritySize];
-    
+
+    this.primitivePower = new int[stripeSize + paritySize];
+    // compute powers of the primitive root
+    for (int i = 0; i < stripeSize + paritySize; i++) {
+      primitivePower[i] = GF.power(PRIMITIVE_ROOT, i);
+    }
     // compute generating polynomial
     int[] gen = {1};
     int[] poly = new int[2];
     for (int i = 0; i < paritySize; i++) {
-      generatingRoots[i] = GF.power(PRIMITIVE_ROOT, i);
-      poly[0] = generatingRoots[i];
+      poly[0] = primitivePower[i];
       poly[1] = 1;
       gen = GF.multiply(gen, poly);
     }
@@ -80,8 +83,8 @@
       data[erasedLocation[i]] = 0;
     }
     for (int i = 0; i < erasedLocation.length; i++) {
-      errSignature[i] = GF.power(PRIMITIVE_ROOT, erasedLocation[i]);
-      erasedValue[i] = GF.substitute(data, generatingRoots[i]);
+      errSignature[i] = primitivePower[erasedLocation[i]];
+      erasedValue[i] = GF.substitute(data, primitivePower[i]);
     }
     GF.solveVandermondeSystem(errSignature, erasedValue, erasedLocation.length);
   }
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonDecoder.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonDecoder.java
new file mode 100644
index 0000000..35eab7d
--- /dev/null
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonDecoder.java
@@ -0,0 +1,226 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hadoop.raid;
+
+import java.io.OutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.ChecksumException;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.BlockMissingException;
+
+public class ReedSolomonDecoder extends Decoder {
+  public static final Log LOG = LogFactory.getLog(
+                                  "org.apache.hadoop.raid.ReedSolomonDecoder");
+  private ErasureCode reedSolomonCode;
+
+  public ReedSolomonDecoder(
+    Configuration conf, int stripeSize, int paritySize) {
+    super(conf, stripeSize, paritySize);
+    this.reedSolomonCode = new ReedSolomonCode(stripeSize, paritySize);
+  }
+
+  @Override
+  protected void fixErasedBlock(
+      FileSystem fs, Path srcFile,
+      FileSystem parityFs, Path parityFile,
+      long blockSize, long errorOffset, long bytesToSkip, long limit,
+      OutputStream out) throws IOException {
+    FSDataInputStream[] inputs = new FSDataInputStream[stripeSize + paritySize];
+    int[] erasedLocations = buildInputs(fs, srcFile, parityFs, parityFile,
+                                        errorOffset, inputs);
+    int blockIdxInStripe = ((int)(errorOffset/blockSize)) % stripeSize;
+    int erasedLocationToFix = paritySize + blockIdxInStripe;
+    writeFixedBlock(inputs, erasedLocations, erasedLocationToFix,
+                    bytesToSkip, limit, out);
+  }
+
+  protected int[] buildInputs(FileSystem fs, Path srcFile,
+                              FileSystem parityFs, Path parityFile,
+                              long errorOffset, FSDataInputStream[] inputs)
+      throws IOException {
+    LOG.info("Building inputs to recover block starting at " + errorOffset);
+    FileStatus srcStat = fs.getFileStatus(srcFile);
+    long blockSize = srcStat.getBlockSize();
+    long blockIdx = (int)(errorOffset / blockSize);
+    long stripeIdx = blockIdx / stripeSize;
+    LOG.info("FileSize = " + srcStat.getLen() + ", blockSize = " + blockSize +
+             ", blockIdx = " + blockIdx + ", stripeIdx = " + stripeIdx);
+    ArrayList<Integer> erasedLocations = new ArrayList<Integer>();
+    // First open streams to the parity blocks.
+    for (int i = 0; i < paritySize; i++) {
+      long offset = blockSize * (stripeIdx * paritySize + i);
+      FSDataInputStream in = parityFs.open(
+        parityFile, conf.getInt("io.file.buffer.size", 64 * 1024));
+      in.seek(offset);
+      LOG.info("Adding " + parityFile + ":" + offset + " as input " + i);
+      inputs[i] = in;
+    }
+    // Now open streams to the data blocks.
+    for (int i = paritySize; i < paritySize + stripeSize; i++) {
+      long offset = blockSize * (stripeIdx * stripeSize + i - paritySize);
+      if (offset == errorOffset) {
+        LOG.info(srcFile + ":" + offset +
+            " is known to have error, adding zeros as input " + i);
+        inputs[i] = new FSDataInputStream(new RaidUtils.ZeroInputStream(
+            offset + blockSize));
+        erasedLocations.add(i);
+      } else if (offset > srcStat.getLen()) {
+        LOG.info(srcFile + ":" + offset +
+                 " is past file size, adding zeros as input " + i);
+        inputs[i] = new FSDataInputStream(new RaidUtils.ZeroInputStream(
+            offset + blockSize));
+      } else {
+        FSDataInputStream in = fs.open(
+          srcFile, conf.getInt("io.file.buffer.size", 64 * 1024));
+        in.seek(offset);
+        LOG.info("Adding " + srcFile + ":" + offset + " as input " + i);
+        inputs[i] = in;
+      }
+    }
+    if (erasedLocations.size() > paritySize) {
+      String msg = "Too many erased locations: " + erasedLocations.size();
+      LOG.error(msg);
+      throw new IOException(msg);
+    }
+    int[] locs = new int[erasedLocations.size()];
+    for (int i = 0; i < locs.length; i++) {
+      locs[i] = erasedLocations.get(i);
+    }
+    return locs;
+  }
+
+  /**
+   * Decode the inputs provided and write to the output.
+   * @param inputs array of inputs.
+   * @param erasedLocations indexes in the inputs which are known to be erased.
+   * @param erasedLocationToFix index in the inputs which needs to be fixed.
+   * @param skipBytes number of bytes to skip before writing to output.
+   * @param limit maximum number of bytes to be written/skipped.
+   * @param out the output.
+   * @throws IOException
+   */
+  void writeFixedBlock(
+          FSDataInputStream[] inputs,
+          int[] erasedLocations,
+          int erasedLocationToFix,
+          long skipBytes,
+          long limit,
+          OutputStream out) throws IOException {
+
+    LOG.info("Need to write " + (limit - skipBytes) +
+             " bytes for erased location index " + erasedLocationToFix);
+    int[] tmp = new int[inputs.length];
+    int[] decoded = new int[erasedLocations.length];
+    long toDiscard = skipBytes;
+    // Loop while the number of skipped + written bytes is less than the max.
+    for (long written = 0; skipBytes + written < limit; ) {
+      erasedLocations = readFromInputs(inputs, erasedLocations, limit);
+      if (decoded.length != erasedLocations.length) {
+        decoded = new int[erasedLocations.length];
+      }
+
+      int toWrite = (int)Math.min((long)bufSize, limit - (skipBytes + written));
+      if (toDiscard >= toWrite) {
+        toDiscard -= toWrite;
+        continue;
+      }
+
+      // Decoded bufSize amount of data.
+      for (int i = 0; i < bufSize; i++) {
+        performDecode(readBufs, writeBufs, i, tmp, erasedLocations, decoded);
+      }
+
+      for (int i = 0; i < erasedLocations.length; i++) {
+        if (erasedLocations[i] == erasedLocationToFix) {
+          toWrite -= toDiscard;
+          out.write(writeBufs[i], (int)toDiscard, toWrite);
+          toDiscard = 0;
+          written += toWrite;
+          LOG.debug("Wrote " + toWrite + " bytes for erased location index " +
+                    erasedLocationToFix);
+          break;
+        }
+      }
+    }
+  }
+
+  int[] readFromInputs(
+          FSDataInputStream[] inputs,
+          int[] erasedLocations,
+          long limit) throws IOException {
+    // For every input, read some data = bufSize.
+    for (int i = 0; i < inputs.length; i++) {
+      long curPos = inputs[i].getPos();
+      try {
+        RaidUtils.readTillEnd(inputs[i], readBufs[i], true);
+        continue;
+      } catch (BlockMissingException e) {
+        LOG.error("Encountered BlockMissingException in stream " + i);
+      } catch (ChecksumException e) {
+        LOG.error("Encountered ChecksumException in stream " + i);
+      }
+
+      // Found a new erased location.
+      if (erasedLocations.length == paritySize) {
+        String msg = "Too many read errors";
+        LOG.error(msg);
+        throw new IOException(msg);
+      }
+
+      // Add this stream to the set of erased locations.
+      int[] newErasedLocations = new int[erasedLocations.length + 1];
+      for (int j = 0; j < erasedLocations.length; j++) {
+        newErasedLocations[j] = erasedLocations[j];
+      }
+      newErasedLocations[newErasedLocations.length - 1] = i;
+      erasedLocations = newErasedLocations;
+
+      LOG.info("Using zeros for stream " + i);
+      inputs[i] = new FSDataInputStream(
+        new RaidUtils.ZeroInputStream(curPos + limit));
+      inputs[i].seek(curPos);
+      RaidUtils.readTillEnd(inputs[i], readBufs[i], true);
+    }
+    return erasedLocations;
+  }
+
+  void performDecode(byte[][] readBufs, byte[][] writeBufs,
+                     int idx, int[] inputs,
+                     int[] erasedLocations, int[] decoded) {
+    for (int i = 0; i < decoded.length; i++) {
+      decoded[i] = 0;
+    }
+    for (int i = 0; i < inputs.length; i++) {
+      inputs[i] = readBufs[i][idx] & 0x000000FF;
+    }
+    reedSolomonCode.decode(inputs, erasedLocations, decoded);
+    for (int i = 0; i < decoded.length; i++) {
+      writeBufs[i][idx] = (byte)decoded[i];
+    }
+  }
+
+}
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonEncoder.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonEncoder.java
new file mode 100644
index 0000000..4119784
--- /dev/null
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/ReedSolomonEncoder.java
@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hadoop.raid;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.Progressable;
+
+public class ReedSolomonEncoder extends Encoder {
+  public static final Log LOG = LogFactory.getLog(
+                                  "org.apache.hadoop.raid.ReedSolomonEncoder");
+  private ErasureCode reedSolomonCode;
+
+  public ReedSolomonEncoder(
+    Configuration conf, int stripeSize, int paritySize) {
+    super(conf, stripeSize, paritySize);
+    this.reedSolomonCode = new ReedSolomonCode(stripeSize, paritySize);
+  }
+
+  protected void encodeStripe(
+    InputStream[] blocks,
+    long stripeStartOffset,
+    long blockSize,
+    OutputStream[] outs,
+    Progressable reporter) throws IOException {
+
+    int[] data = new int[stripeSize];
+    int[] code = new int[paritySize];
+
+    for (long encoded = 0; encoded < blockSize; encoded += bufSize) {
+      // Read some data from each block = bufSize.
+      for (int i = 0; i < blocks.length; i++) {
+        RaidUtils.readTillEnd(blocks[i], readBufs[i], true);
+      }
+
+      // Encode the data read.
+      for (int j = 0; j < bufSize; j++) {
+        performEncode(readBufs, writeBufs, j, data, code);
+      }
+
+      // Now that we have some data to write, send it to the temp files.
+      for (int i = 0; i < paritySize; i++) {
+        outs[i].write(writeBufs[i], 0, bufSize);
+      }
+
+      if (reporter != null) {
+        reporter.progress();
+      }
+    }
+  }
+
+  void performEncode(byte[][] readBufs, byte[][] writeBufs, int idx,
+                          int[] data, int[] code) {
+    for (int i = 0; i < paritySize; i++) {
+      code[i] = 0;
+    }
+    for (int i = 0; i < stripeSize; i++) {
+      data[i] = readBufs[i][idx] & 0x000000FF;
+    }
+    reedSolomonCode.encode(data, code);
+    for (int i = 0; i < paritySize; i++) {
+      writeBufs[i][idx] = (byte)code[i];
+    }
+  }
+
+  @Override
+  public Path getParityTempPath() {
+    return new Path(RaidNode.rsTempPrefix(conf));
+  }
+
+}
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/XOREncoder.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/XOREncoder.java
index 3d113e6..469a171 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/XOREncoder.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/XOREncoder.java
@@ -55,4 +55,9 @@
       parityIn.close();
     }
   }
+
+  @Override
+  public Path getParityTempPath() {
+    return new Path(RaidNode.unraidTmpDirectory(conf));
+  }
 }
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java
index 90e1ad4..2df302a 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java
@@ -49,14 +49,26 @@
 
   private Path srcPath;            // the specified src path
   private String policyName;       // name of policy
-  private String destinationPath;  // A destination path for this policy
+  private ErasureCodeType codeType;// the erasure code used
   private String description;      // A verbose description of this policy
   private Configuration conf;      // Hadoop configuration
 
   private Properties properties;   // Policy-dependent properties
 
   private ReentrantReadWriteLock plock; // protects policy operations.
-  
+  public static enum ErasureCodeType {
+    XOR, RS;
+    public static ErasureCodeType fromString(String s) {
+      if (XOR.toString().equalsIgnoreCase(s)) {
+        return XOR;
+      }
+      if (RS.toString().equalsIgnoreCase(s)) {
+        return RS;
+      }
+      return null;
+    }
+  }
+
   /**
    * Create the empty object
    */
@@ -82,6 +94,31 @@
   }
 
   /**
+   * Copy fields from another PolicyInfo
+   */
+  public void copyFrom(PolicyInfo other) {
+    if (other.conf != null) {
+      this.conf = other.conf;
+    }
+    if (other.policyName != null && other.policyName.length() > 0) {
+      this.policyName = other.policyName;
+    }
+    if (other.description != null && other.description.length() > 0) {
+      this.description = other.description;
+    }
+    if (other.codeType != null) {
+      this.codeType = other.codeType;
+    }
+    if (other.srcPath != null) {
+      this.srcPath = other.srcPath;
+    }
+    for (Object key : other.properties.keySet()) {
+      String skey = (String) key;
+      this.properties.setProperty(skey, other.properties.getProperty(skey));
+    }
+  }
+
+  /**
    * Sets the input path on which this policy has to be applied
    */
   public void setSrcPath(String in) throws IOException {
@@ -90,10 +127,10 @@
   }
 
   /**
-   * Set the destination path of this policy.
+   * Set the erasure code type used in this policy
    */
-  public void setDestinationPath(String des) {
-    this.destinationPath = des;
+  public void setErasureCode(String code) {
+    this.codeType = ErasureCodeType.fromString(code);
   }
 
   /**
@@ -130,8 +167,8 @@
   /**
    * Get the destination path of this policy.
    */
-  public String getDestinationPath() {
-    return this.destinationPath;
+  public ErasureCodeType getErasureCode() {
+    return this.codeType;
   }
 
   /**
@@ -167,7 +204,7 @@
     StringBuffer buff = new StringBuffer();
     buff.append("Policy Name:\t" + policyName + " --------------------\n");
     buff.append("Source Path:\t" + srcPath + "\n");
-    buff.append("Dest Path:\t" + destinationPath + "\n");
+    buff.append("Erasure Code:\t" + codeType + "\n");
     for (Enumeration<?> e = properties.propertyNames(); e.hasMoreElements();) {
       String name = (String) e.nextElement(); 
       buff.append( name + ":\t" + properties.getProperty(name) + "\n");
@@ -195,7 +232,7 @@
   public void write(DataOutput out) throws IOException {
     Text.writeString(out, srcPath.toString());
     Text.writeString(out, policyName);
-    Text.writeString(out, destinationPath);
+    Text.writeString(out, codeType.toString());
     Text.writeString(out, description);
     out.writeInt(properties.size());
     for (Enumeration<?> e = properties.propertyNames(); e.hasMoreElements();) {
@@ -208,7 +245,7 @@
   public void readFields(DataInput in) throws IOException {
     this.srcPath = new Path(Text.readString(in));
     this.policyName = Text.readString(in);
-    this.destinationPath = Text.readString(in);
+    this.codeType = ErasureCodeType.fromString(Text.readString(in));
     this.description = Text.readString(in);
     for (int n = in.readInt(); n>0; n--) {
       String name = Text.readString(in);
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java b/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java
index ad8a33c..9926eef 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/hdfs/TestRaidDfs.java
@@ -73,7 +73,8 @@
   RaidNode cnode = null;
   String jobTrackerName = null;
 
-  private void mySetup() throws Exception {
+  private void mySetup(String erasureCode, int stripeLength,
+      int rsParityLength) throws Exception {
 
     new File(TEST_DIR).mkdirs(); // Make sure data directory exists
     conf = new Configuration();
@@ -81,6 +82,7 @@
     conf.set("raid.config.file", CONFIG_FILE);
     conf.setBoolean("raid.config.reload", true);
     conf.setLong("raid.config.reload.interval", RELOAD_INTERVAL);
+    conf.setInt(RaidNode.RS_PARITY_LENGTH_KEY, rsParityLength);
 
     // scan all policies once every 5 second
     conf.setLong("raid.policy.rescan.interval", 5000);
@@ -92,8 +94,9 @@
     conf.set("raid.classname", "org.apache.hadoop.raid.LocalRaidNode");
 
     conf.set("raid.server.address", "localhost:0");
-    conf.setInt("hdfs.raid.stripeLength", 3);
-    conf.set("hdfs.raid.locations", "/destraid");
+    conf.setInt("hdfs.raid.stripeLength", stripeLength);
+    conf.set("xor".equals(erasureCode) ? RaidNode.RAID_LOCATION_KEY :
+             RaidNode.RAIDRS_LOCATION_KEY, "/destraid");
 
     dfs = new MiniDFSCluster(conf, NUM_DATANODES, true, null);
     dfs.waitActive();
@@ -108,7 +111,7 @@
     String str = "<configuration> " +
                    "<srcPath prefix=\"/user/dhruba/raidtest\"> " +
                      "<policy name = \"RaidTest1\"> " +
-                        "<destPath> /destraid</destPath> " +
+                        "<erasureCode>" + erasureCode + "</erasureCode> " +
                         "<property> " +
                           "<name>targetReplication</name> " +
                           "<value>1</value> " + 
@@ -231,47 +234,28 @@
   }
 
   /**
-   * Create a file, corrupt a block in it and ensure that the file can be
-   * read through DistributedRaidFileSystem.
+   * Create a file, corrupt several blocks in it and ensure that the file can be
+   * read through DistributedRaidFileSystem by ReedSolomon coding.
    */
-  public void testRaidDfs() throws Exception {
+  public void testRaidDfsRs() throws Exception {
     LOG.info("Test testRaidDfs started.");
 
     long blockSize = 8192L;
     int numBlocks = 8;
-    int repl = 1;
-    mySetup();
+    int stripeLength = 3;
+    mySetup("rs", stripeLength, 3);
 
     // Create an instance of the RaidNode
     Configuration localConf = new Configuration(conf);
     localConf.set(RaidNode.RAID_LOCATION_KEY, "/destraid");
     cnode = RaidNode.createRaidNode(null, localConf);
-
-    Path file = new Path("/user/dhruba/raidtest/file");
     Path destPath = new Path("/destraid/user/dhruba/raidtest");
-    int[][] corrupt = {{0}, {4}, {7}}; // first, last and middle block
+    int[][] corrupt = {{1, 2, 3}, {1, 4, 7}, {3, 6, 7}};
     try {
-      long crc = createTestFilePartialLastBlock(fileSys, file, repl,
-                  numBlocks, blockSize);
-      long length = fileSys.getFileStatus(file).getLen();
-      waitForFileRaided(LOG, fileSys, file, destPath);
-      LocatedBlocks locations = getBlockLocations(file);
-
       for (int i = 0; i < corrupt.length; i++) {
-        int blockNumToCorrupt = corrupt[i][0];
-        LOG.info("Corrupt block " + blockNumToCorrupt + " of file");
-        corruptBlock(file, locations.get(blockNumToCorrupt).getBlock(),
-          NUM_DATANODES, false);
-        validateFile(getRaidFS(), file, length, crc);
-      }
-
-      // Corrupt one more block. This is expected to fail.
-      LOG.info("Corrupt one more block of file");
-      corruptBlock(file, locations.get(1).getBlock(), NUM_DATANODES, false);
-      try {
-        validateFile(getRaidFS(), file, length, crc);
-        fail("Expected exception ChecksumException not thrown!");
-      } catch (org.apache.hadoop.fs.ChecksumException e) {
+        Path file = new Path("/user/dhruba/raidtest/file" + i);
+        corruptBlockAndValidate(
+            file, destPath, corrupt[0], blockSize, numBlocks);
       }
     } catch (Exception e) {
       LOG.info("testRaidDfs Exception " + e +
@@ -288,7 +272,7 @@
    * Test DistributedRaidFileSystem.readFully()
    */
   public void testReadFully() throws Exception {
-    mySetup();
+    mySetup("xor", 3, 1);
 
     try {
       Path file = new Path("/user/raid/raidtest/file1");
@@ -328,7 +312,7 @@
     long blockSize = 8192L;
     int numBlocks = 8;
     int repl = 1;
-    mySetup();
+    mySetup("xor", 3, 1);
 
     Path file = new Path("/user/dhruba/raidtest/file");
     Path destPath = new Path("/destraid/user/dhruba/raidtest");
@@ -351,6 +335,41 @@
       myTearDown();
     }
   }
+  /**
+   * Create a file, corrupt a block in it and ensure that the file can be
+   * read through DistributedRaidFileSystem by XOR code.
+   */
+  public void testRaidDfsXor() throws Exception {
+    LOG.info("Test testRaidDfs started.");
+
+    long blockSize = 8192L;
+    int numBlocks = 8;
+    int stripeLength = 3;
+    mySetup("xor", stripeLength, 1);
+
+    // Create an instance of the RaidNode
+    Configuration localConf = new Configuration(conf);
+    localConf.set(RaidNode.RAID_LOCATION_KEY, "/destraid");
+    cnode = RaidNode.createRaidNode(null, localConf);
+
+    Path destPath = new Path("/destraid/user/dhruba/raidtest");
+    int[][] corrupt = {{0}, {4}, {7}}; // first, last and middle block
+    try {
+      for (int i = 0; i < corrupt.length; i++) {
+        Path file = new Path("/user/dhruba/raidtest/" + i);
+        corruptBlockAndValidate(
+            file, destPath, corrupt[0], blockSize, numBlocks);
+      }
+    } catch (Exception e) {
+      LOG.info("testRaidDfs Exception " + e +
+                StringUtils.stringifyException(e));
+      throw e;
+    } finally {
+      if (cnode != null) { cnode.stop(); cnode.join(); }
+      myTearDown();
+    }
+    LOG.info("Test testRaidDfs completed.");
+  }
 
   //
   // creates a file and populate it with random data. Returns its crc.
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java
index d883385..12151c0 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java
@@ -48,6 +48,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.mapred.MiniMRCluster;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
 import org.apache.hadoop.hdfs.protocol.Block;
@@ -56,6 +57,7 @@
 import org.apache.hadoop.hdfs.RaidDFSUtil;
 import org.apache.hadoop.hdfs.TestDatanodeBlockScanner;
 import org.apache.hadoop.hdfs.TestRaidDfs;
+import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.raid.RaidNode;
 import org.apache.hadoop.raid.RaidUtils;
 
@@ -520,7 +522,7 @@
     assertTrue(corrupted);
   }
 
-  void reportCorruptBlocks(FileSystem fs, Path file, int[] idxs,
+  static void reportCorruptBlocks(FileSystem fs, Path file, int[] idxs,
     long blockSize) throws IOException {
 
     FSDataInputStream in = fs.open(file);
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java
index 386cb6c..93658a5 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java
@@ -32,6 +32,8 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.mapred.Reporter;
+
 import org.apache.hadoop.raid.protocol.PolicyInfo;
 
 public class TestDirectoryTraversal extends TestCase {
@@ -89,6 +91,7 @@
   }
 
   public void testSuspension() throws IOException {
+    LOG.info("Starting testSuspension");
     mySetup();
 
     try {
@@ -107,7 +110,7 @@
       Path raid = new Path("/raid");
       DirectoryTraversal.FileFilter filter =
         new RaidFilter.TimeBasedFilter(conf,
-          RaidNode.getDestinationPath(conf), 1, System.currentTimeMillis(), 0);
+          RaidNode.xorDestinationPath(conf), 1, System.currentTimeMillis(), 0);
       List<FileStatus> selected = dt.getFilteredFiles(filter, limit);
       for (FileStatus f: selected) {
         LOG.info(f.getPath());
@@ -124,6 +127,50 @@
     }
   }
 
+  public void testFileFilter() throws IOException {
+    mySetup();
+
+    try {
+      Path topDir = new Path(TEST_DIR + "/testFileFilter");
+      int targetRepl = 1;
+      createTestTree(topDir);
+      Path file = new Path(topDir.toString() + "/a/f1");
+      FileStatus stat = fs.getFileStatus(file);
+      PolicyInfo info = new PolicyInfo("testFileFilter", conf);
+      info.setSrcPath(topDir.toString());
+      info.setErasureCode("rs");
+      info.setDescription("test policy");
+      info.setProperty("targetReplication", "1");
+      info.setProperty("metaReplication", "1");
+
+      DirectoryTraversal.FileFilter timeBasedXORFilter =
+        new RaidFilter.TimeBasedFilter(conf,
+          RaidNode.xorDestinationPath(conf), targetRepl,
+            System.currentTimeMillis(), 0);
+      DirectoryTraversal.FileFilter timeBasedRSFilter =
+        new RaidFilter.TimeBasedFilter(conf,
+          RaidNode.rsDestinationPath(conf), targetRepl,
+            System.currentTimeMillis(), 0);
+      DirectoryTraversal.FileFilter preferenceForRSFilter =
+        new RaidFilter.PreferenceFilter(
+          conf, RaidNode.rsDestinationPath(conf),
+          RaidNode.xorDestinationPath(conf), 1, System.currentTimeMillis(), 0);
+
+      assertTrue(timeBasedXORFilter.check(stat));
+      assertTrue(timeBasedRSFilter.check(stat));
+      assertTrue(preferenceForRSFilter.check(stat));
+
+      RaidNode.doRaid(
+        conf, info, stat, new RaidNode.Statistics(), Reporter.NULL);
+
+      assertTrue(timeBasedXORFilter.check(stat));
+      assertFalse(timeBasedRSFilter.check(stat));
+      assertFalse(preferenceForRSFilter.check(stat));
+    } finally {
+      myTearDown();
+    }
+  }
+
   /**
    * Creates a test directory tree.
    *            top
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidFilter.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidFilter.java
index 9b1cdd9..4b1fe67 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidFilter.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidFilter.java
@@ -65,6 +65,7 @@
 
     PolicyInfo info1 = new PolicyInfo("p1", conf);
     info1.setSrcPath(src1.toString());
+    info1.setErasureCode("xor");
     info1.setDescription("test policy");
     info1.setProperty("targetReplication", "1");
     info1.setProperty("metaReplication", "1");
@@ -72,6 +73,7 @@
 
     PolicyInfo info2 = new PolicyInfo("p2", conf);
     info2.setSrcPath(src2.toString());
+    info2.setErasureCode("xor");
     info2.setDescription("test policy");
     info2.setProperty("targetReplication", "1");
     info2.setProperty("metaReplication", "1");
@@ -103,7 +105,7 @@
 
       RaidFilter.Statistics stats = new RaidFilter.Statistics();
       RaidFilter.TimeBasedFilter filter = new RaidFilter.TimeBasedFilter(
-        conf, RaidNode.getDestinationPath(conf), info1, all,
+        conf, RaidNode.xorDestinationPath(conf), info1, all,
         System.currentTimeMillis(), stats);
       System.out.println("Stats " + stats);
 
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidHar.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidHar.java
index 513d5aa..3439804 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidHar.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidHar.java
@@ -113,7 +113,7 @@
     String str = "<configuration> " +
                    "<srcPath prefix=\"/user/test/raidtest\"> " +
                      "<policy name = \"RaidTest1\"> " +
-                        "<destPath> /destraid</destPath> " +
+                        "<erasureCode>xor</erasureCode> " +
                         "<property> " +
                           "<name>targetReplication</name> " +
                           "<value>" + targetReplication + "</value> " +
@@ -247,6 +247,9 @@
             }
           }
           if (count == 1  && listPaths.length == 1) {
+            Path partfile = new Path(harPath, "part-0");
+            assertEquals(fileSys.getFileStatus(partfile).getReplication(),
+              targetReplication);
             break;
           }
         } catch (FileNotFoundException e) {
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java
index 1336b41..1dd6841 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidNode.java
@@ -48,6 +48,7 @@
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.raid.protocol.PolicyInfo;
 import org.apache.hadoop.raid.protocol.PolicyList;
+import org.apache.hadoop.raid.protocol.PolicyInfo.ErasureCodeType;
 
 /**
   * Test the generation of parity blocks for files with different block
@@ -122,12 +123,22 @@
       policies = new java.util.ArrayList<String>();
     }
 
+    public void addPolicy(String name, String path, String parent) {
+      String str =
+        "<srcPath prefix=\"" + path + "\"> " +
+          "<policy name = \"" + name + "\"> " +
+             "<parentPolicy>" + parent + "</parentPolicy>" +
+          "</policy>" +
+        "</srcPath>";
+      policies.add(str);
+    }
+
     public void addPolicy(String name, short srcReplication,
                           long targetReplication, long metaReplication, long stripeLength) {
       String str =
         "<srcPath prefix=\"/user/dhruba/raidtest\"> " +
           "<policy name = \"" + name + "\"> " +
-             "<destPath> /destraid</destPath> " +
+             "<erasureCode>xor</erasureCode> " +
              "<property> " +
                "<name>srcReplication</name> " +
                "<value>" + srcReplication + "</value> " +
@@ -169,7 +180,7 @@
       String str =
         "<srcPath prefix=\"" + path + "\"> " +
           "<policy name = \"" + name + "\"> " +
-             "<destPath> /destraid</destPath> " +
+             "<erasureCode>xor</erasureCode> " +
              "<property> " +
                "<name>srcReplication</name> " +
                "<value>" + srcReplication + "</value> " +
@@ -280,7 +291,6 @@
 
       // create an instance of the RaidNode
       cnode = RaidNode.createRaidNode(null, conf);
-      int times = 10;
 
       FileStatus[] listPaths = null;
 
@@ -345,7 +355,7 @@
                                           StringUtils.stringifyException(e));
       throw e;
     } finally {
-      shell.close();
+      if (shell != null) shell.close();
       if (cnode != null) { cnode.stop(); cnode.join(); }
       LOG.info("doTestPathFilter delete file " + file1);
       fileSys.delete(file1, true);
@@ -388,8 +398,14 @@
       long firstmodtime = 0;
       // wait till file is raided
       while (true) {
-        Thread.sleep(20000L);                  // waiting
-        listPaths = fileSys.listStatus(destPath);
+        Thread.sleep(1000);                  // waiting
+        try {
+          listPaths = fileSys.listStatus(destPath);
+        } catch (FileNotFoundException e) {
+          LOG.warn("File not found " + destPath);
+          // The directory have been deleted by the purge thread.
+          continue;
+        }
         int count = 0;
         if (listPaths != null && listPaths.length == 1) {
           for (FileStatus s : listPaths) {
@@ -554,7 +570,7 @@
                      long crc, long corruptOffset) throws IOException {
     // recover the file assuming that we encountered a corruption at offset 0
     String[] args = new String[3];
-    args[0] = "recover";
+    args[0] = "-recover";
     args[1] = file1.toString();
     args[2] = Long.toString(corruptOffset);
     Path recover1 = shell.recover(args[0], args, 1)[0];
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidPurge.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidPurge.java
index a341877..accef0b 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidPurge.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidPurge.java
@@ -47,9 +47,11 @@
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.mapred.MiniMRCluster;
 import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.hdfs.TestRaidDfs;
 import org.apache.hadoop.raid.protocol.PolicyInfo;
 import org.apache.hadoop.raid.protocol.PolicyList;
+import org.apache.hadoop.hdfs.TestRaidDfs;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.raid.protocol.PolicyInfo;
 
 /**
  * If a file gets deleted, then verify that the parity file gets deleted too.
@@ -61,6 +63,7 @@
       "test-raid.xml").getAbsolutePath();
   final static long RELOAD_INTERVAL = 1000;
   final static Log LOG = LogFactory.getLog("org.apache.hadoop.raid.TestRaidNode");
+  final Random rand = new Random();
 
   {
     ((Log4JLogger)RaidNode.LOG).getLogger().setLevel(Level.ALL);
@@ -69,8 +72,11 @@
 
   Configuration conf;
   String namenode = null;
+  String hftp = null;
   MiniDFSCluster dfs = null;
+  MiniMRCluster mr = null;
   FileSystem fileSys = null;
+  String jobTrackerName = null;
 
   /**
    * create mapreduce and dfs clusters
@@ -97,7 +103,7 @@
     }
 
     conf.set("raid.server.address", "localhost:0");
-    
+
     // create a dfs and map-reduce cluster
     final int taskTrackers = 4;
     final int jobTrackerPort = 60050;
@@ -106,25 +112,31 @@
     dfs.waitActive();
     fileSys = dfs.getFileSystem();
     namenode = fileSys.getUri().toString();
+    mr = new MiniMRCluster(taskTrackers, namenode, 3);
+    jobTrackerName = "localhost:" + mr.getJobTrackerPort();
+    hftp = "hftp://localhost.localdomain:" + dfs.getNameNodePort();
+
+    FileSystem.setDefaultUri(conf, namenode);
+    conf.set("mapred.job.tracker", jobTrackerName);
   }
     
   /**
    * create raid.xml file for RaidNode
    */
-  private void mySetup(String srcPath, long targetReplication,
-                long metaReplication, long stripeLength)
-    throws Exception {
-    mySetup(srcPath, targetReplication, metaReplication, stripeLength, 1);
+  private void mySetup(long targetReplication,
+    long metaReplication, long stripeLength) throws Exception {
+    int harDelay = 1; // 1 day.
+    mySetup(targetReplication, metaReplication, stripeLength, harDelay);
   }
 
-  private void mySetup(String srcPath, long targetReplication,
-                long metaReplication, long stripeLength, int harDelay)
-    throws Exception {
+  private void mySetup(long targetReplication,
+    long metaReplication, long stripeLength, int harDelay) throws Exception {
     FileWriter fileWriter = new FileWriter(CONFIG_FILE);
     fileWriter.write("<?xml version=\"1.0\"?>\n");
     String str = "<configuration> " +
-                   "<srcPath prefix=\"" + srcPath + "\"> " +
+                   "<srcPath prefix=\"/user/dhruba/raidtest\"> " +
                      "<policy name = \"RaidTest1\"> " +
+                        "<erasureCode>xor</erasureCode> " +
                         "<destPath> /destraid</destPath> " +
                         "<property> " +
                           "<name>targetReplication</name> " +
@@ -154,7 +166,7 @@
                         "<property> " +
                           "<name>time_before_har</name> " +
                           "<value> " + harDelay + "</value> " +
-                          "<description> time before har'ing parity files" +
+                          "<description> amount of time waited before har'ing parity files" +
                           "</description> " +
                         "</property> " +
                      "</policy>" +
@@ -168,6 +180,7 @@
    * stop clusters created earlier
    */
   private void stopClusters() throws Exception {
+    if (mr != null) { mr.shutdown(); }
     if (dfs != null) { dfs.shutdown(); }
   }
 
@@ -178,7 +191,6 @@
   public void testPurge() throws Exception {
     LOG.info("Test testPurge  started.");
 
-    String srcPaths    []  = { "/user/dhruba/raidtest", "/user/dhruba/raid*" };
     long blockSizes    []  = {1024L};
     long stripeLengths []  = {5};
     long targetReplication = 1;
@@ -188,13 +200,11 @@
 
     createClusters(true);
     try {
-      for (String srcPath : srcPaths ) {
-        for (long blockSize : blockSizes) {
-          for (long stripeLength : stripeLengths) {
-            doTestPurge(iter, srcPath, targetReplication, metaReplication,
-                stripeLength, blockSize, numBlock);
-            iter++;
-          }
+      for (long blockSize : blockSizes) {
+        for (long stripeLength : stripeLengths) {
+           doTestPurge(iter, targetReplication, metaReplication,
+                       stripeLength, blockSize, numBlock);
+           iter++;
         }
       }
     } finally {
@@ -207,12 +217,12 @@
    * Create parity file, delete original file and then validate that
    * parity file is automatically deleted.
    */
-  private void doTestPurge(int iter, String srcPath, long targetReplication,
+  private void doTestPurge(int iter, long targetReplication,
                           long metaReplication, long stripeLength,
                           long blockSize, int numBlock) throws Exception {
     LOG.info("doTestPurge started---------------------------:" +  " iter " + iter +
              " blockSize=" + blockSize + " stripeLength=" + stripeLength);
-    mySetup(srcPath, targetReplication, metaReplication, stripeLength);
+    mySetup(targetReplication, metaReplication, stripeLength);
     Path dir = new Path("/user/dhruba/raidtest/");
     Path file1 = new Path(dir + "/file" + iter);
     RaidNode cnode = null;
@@ -287,7 +297,7 @@
     LOG.info("testPurgeHar started");
     int harDelay = 0;
     createClusters(true);
-    mySetup("/user/dhruba/raidtest", 1, 1, 5, harDelay);
+    mySetup(1, 1, 5, harDelay);
     Path dir = new Path("/user/dhruba/raidtest/");
     Path destPath = new Path("/destraid/user/dhruba/raidtest");
     Path file1 = new Path(dir + "/file");
@@ -324,7 +334,14 @@
       boolean found = false;
       FileStatus[] listPaths = null;
       while (!found || listPaths == null || listPaths.length > 1) {
-        listPaths = fileSys.listStatus(destPath);
+        try {
+          listPaths = fileSys.listStatus(destPath);
+        } catch (FileNotFoundException e) {
+          // If the parent directory is deleted because the har is deleted
+          // and the parent is empty, try again.
+          Thread.sleep(1000);
+          continue;
+        }
         if (listPaths != null) {
           for (FileStatus s: listPaths) {
             LOG.info("testPurgeHar waiting for parity file to be recreated" +
@@ -348,4 +365,153 @@
       stopClusters();
     }
   }
+
+  /**
+   * Create parity file, delete original file's directory and then validate that
+   * parity directory is automatically deleted.
+   */
+  public void testPurgeDirectory() throws Exception {
+    long stripeLength = 5;
+    long blockSize = 8192;
+    long targetReplication = 1;
+    long metaReplication   = 1;
+    int  numBlock          = 9;
+
+    createClusters(true);
+    mySetup(targetReplication, metaReplication, stripeLength);
+    Path dir = new Path("/user/dhruba/raidtest/");
+    Path file1 = new Path(dir + "/file1");
+    RaidNode cnode = null;
+    try {
+      TestRaidNode.createOldFile(fileSys, file1, 1, numBlock, blockSize);
+
+      // create an instance of the RaidNode
+      Configuration localConf = new Configuration(conf);
+      localConf.set(RaidNode.RAID_LOCATION_KEY, "/destraid");
+      cnode = RaidNode.createRaidNode(null, localConf);
+
+      Path destPath = new Path("/destraid/user/dhruba/raidtest");
+      TestRaidDfs.waitForFileRaided(LOG, fileSys, file1, destPath);
+
+      // delete original directory.
+      assertTrue("Unable to delete original directory " + file1 ,
+                 fileSys.delete(file1.getParent(), true));
+      LOG.info("deleted file " + file1);
+
+      // wait till parity file and directory are automatically deleted
+      long start = System.currentTimeMillis();
+      while (fileSys.exists(destPath) &&
+            System.currentTimeMillis() - start < 120000) {
+        LOG.info("testPurgeDirectory waiting for parity files to be removed.");
+        Thread.sleep(1000);                  // keep waiting
+      }
+      assertFalse(fileSys.exists(destPath));
+
+    } catch (Exception e) {
+      LOG.info("testPurgeDirectory Exception " + e +
+                                          StringUtils.stringifyException(e));
+      throw e;
+    } finally {
+      if (cnode != null) { cnode.stop(); cnode.join(); }
+      LOG.info("testPurgeDirectory delete file " + file1);
+      fileSys.delete(file1, true);
+      stopClusters();
+    }
+  }
+
+  /**
+   * Test that an XOR parity file is removed when a RS parity file is detected.
+   */
+  public void testPurgePreference() throws Exception {
+    createClusters(true);
+    Path dir = new Path("/user/test/raidtest/");
+    Path file1 = new Path(dir + "/file1");
+
+    PolicyInfo infoXor = new PolicyInfo("testPurgePreference", conf);
+    infoXor.setSrcPath("/user/test/raidtest");
+    infoXor.setErasureCode("xor");
+    infoXor.setDescription("test policy");
+    infoXor.setProperty("targetReplication", "2");
+    infoXor.setProperty("metaReplication", "2");
+
+    PolicyInfo infoRs = new PolicyInfo("testPurgePreference", conf);
+    infoRs.setSrcPath("/user/test/raidtest");
+    infoRs.setErasureCode("rs");
+    infoRs.setDescription("test policy");
+    infoRs.setProperty("targetReplication", "1");
+    infoRs.setProperty("metaReplication", "1");
+    try {
+      TestRaidNode.createOldFile(fileSys, file1, 1, 9, 8192L);
+      FileStatus stat = fileSys.getFileStatus(file1);
+
+      // Create the parity files.
+      RaidNode.doRaid(
+        conf, infoXor, stat, new RaidNode.Statistics(), Reporter.NULL);
+      RaidNode.doRaid(
+        conf, infoRs, stat, new RaidNode.Statistics(), Reporter.NULL);
+      Path xorParity =
+        new Path(RaidNode.DEFAULT_RAID_LOCATION, "user/test/raidtest/file1");
+      Path rsParity =
+        new Path(RaidNode.DEFAULT_RAIDRS_LOCATION, "user/test/raidtest/file1");
+      assertTrue(fileSys.exists(xorParity));
+      assertTrue(fileSys.exists(rsParity));
+
+      // Check purge of a single parity file.
+      RaidNode cnode = RaidNode.createRaidNode(conf);
+      FileStatus raidRsStat =
+        fileSys.getFileStatus(new Path(RaidNode.DEFAULT_RAIDRS_LOCATION));
+      cnode.purgeMonitor.recursePurge(infoRs.getErasureCode(), fileSys, fileSys,
+         RaidNode.DEFAULT_RAIDRS_LOCATION, raidRsStat);
+
+      // Calling purge under the RS path has no effect.
+      assertTrue(fileSys.exists(xorParity));
+      assertTrue(fileSys.exists(rsParity));
+
+      FileStatus raidStat =
+         fileSys.getFileStatus(new Path(RaidNode.DEFAULT_RAID_LOCATION));
+      cnode.purgeMonitor.recursePurge(infoXor.getErasureCode(), fileSys, fileSys,
+         RaidNode.DEFAULT_RAID_LOCATION, raidStat);
+      // XOR parity must have been purged by now.
+      assertFalse(fileSys.exists(xorParity));
+      assertTrue(fileSys.exists(rsParity));
+
+      // Now check the purge of a parity har.
+      // Delete the RS parity for now.
+      fileSys.delete(rsParity);
+      // Recreate the XOR parity.
+      Path xorHar =
+        new Path(RaidNode.DEFAULT_RAID_LOCATION, "user/test/raidtest/raidtest" +
+          RaidNode.HAR_SUFFIX);
+      RaidNode.doRaid(
+        conf, infoXor, stat, new RaidNode.Statistics(), Reporter.NULL);
+      assertTrue(fileSys.exists(xorParity));
+      assertFalse(fileSys.exists(xorHar));
+
+      // Create the har.
+      long cutoff = System.currentTimeMillis();
+      cnode.recurseHar(infoXor, fileSys, raidStat,
+        RaidNode.DEFAULT_RAID_LOCATION, fileSys, cutoff,
+        RaidNode.tmpHarPathForCode(conf, infoXor.getErasureCode()));
+
+      // Call purge to get rid of the parity file. The har should remain.
+      cnode.purgeMonitor.recursePurge(infoXor.getErasureCode(), fileSys, fileSys,
+         RaidNode.DEFAULT_RAID_LOCATION, raidStat);
+      // XOR har should exist but xor parity file should have been purged.
+      assertFalse(fileSys.exists(xorParity));
+      assertTrue(fileSys.exists(xorHar));
+
+      // Now create the RS parity.
+      RaidNode.doRaid(
+        conf, infoRs, stat, new RaidNode.Statistics(), Reporter.NULL);
+      cnode.purgeMonitor.recursePurge(infoXor.getErasureCode(), fileSys, fileSys,
+         RaidNode.DEFAULT_RAID_LOCATION, raidStat);
+      // XOR har should get deleted.
+      assertTrue(fileSys.exists(rsParity));
+      assertFalse(fileSys.exists(xorParity));
+      assertFalse(fileSys.exists(xorHar));
+
+    } finally {
+      stopClusters();
+    }
+  }
 }
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java
index 427f915..c93631d 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidShell.java
@@ -58,6 +58,7 @@
 import org.apache.hadoop.hdfs.DistributedRaidFileSystem;
 import org.apache.hadoop.hdfs.TestRaidDfs;
 import org.apache.hadoop.hdfs.RaidDFSUtil;
+import org.apache.hadoop.hdfs.TestDatanodeBlockScanner;
 import org.apache.hadoop.raid.RaidNode;
 
 
@@ -120,16 +121,10 @@
       int[] corruptBlockIdxs = new int[]{0, 4, 6};
       for (int idx: corruptBlockIdxs) {
         LOG.info("Corrupting block " + locations.get(idx).getBlock());
-        TestRaidDfs.corruptBlock(file1, locations.get(idx).getBlock(),
-                                  NUM_DATANODES, false); // corrupt block
-        long offset = idx * blockSize;
-        try {
-          readAndDiscard(fileSys, file1, offset, blockSize);
-          fail("Expected checksumexception not thrown");
-        } catch (ChecksumException e) {
-          LOG.info("Block at offset " + offset + " got expected exception");
-        }
+        corruptBlock(locations.get(idx).getBlock().getBlockName());
       }
+      TestBlockFixer.reportCorruptBlocks(fileSys, file1, corruptBlockIdxs,
+        srcStat.getBlockSize());
 
       String fileUriPath = file1.toUri().getPath();
       waitForCorruptBlocks(corruptBlockIdxs.length, dfs, file1);
@@ -150,14 +145,9 @@
       long parityCrc = getCRC(fileSys, parityFile);
       locations = RaidDFSUtil.getBlockLocations(
         dfs, parityFile.toUri().getPath(), 0, parityStat.getLen());
-      TestRaidDfs.corruptBlock(parityFile, locations.get(0).getBlock(),
-                                NUM_DATANODES, false); // corrupt block
-      try {
-        readAndDiscard(fileSys, parityFile, 0, blockSize);
-        fail("Expected checksumexception not thrown");
-      } catch (ChecksumException e) {
-        LOG.info("Parity Block at offset 0 got expected exception");
-      }
+      corruptBlock(locations.get(0).getBlock().getBlockName());
+      TestBlockFixer.reportCorruptBlocks(fileSys, parityFile, new int[]{0},
+        srcStat.getBlockSize());
       waitForCorruptBlocks(1, dfs, parityFile);
 
       args[1] = parityFile.toUri().getPath();
@@ -294,14 +284,11 @@
     return crc.getValue();
   }
 
-  private static void readAndDiscard(
-    FileSystem fs, Path p, long offset, long length) throws IOException {
-    FSDataInputStream in = fs.open(p);
-    in.seek(offset);
-    long count = 0;
-    for (int b = 0; b >= 0 && count < length; count++) {
-      b = in.read();
+  void corruptBlock(String blockName) throws IOException {
+    boolean corrupted = false;
+    for (int i = 0; i < NUM_DATANODES; i++) {
+      corrupted |= TestDatanodeBlockScanner.corruptReplica(blockName, i);
     }
+    assertTrue(corrupted);
   }
 }
-
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestReedSolomonDecoder.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestReedSolomonDecoder.java
new file mode 100644
index 0000000..03d5a6e
--- /dev/null
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestReedSolomonDecoder.java
@@ -0,0 +1,146 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hadoop.raid;
+
+import java.io.File;
+import java.io.IOException;
+
+import junit.framework.TestCase;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FilterFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
+import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.DistributedRaidFileSystem;
+import org.apache.hadoop.hdfs.RaidDFSUtil;
+import org.apache.hadoop.hdfs.TestDatanodeBlockScanner;
+import org.apache.hadoop.hdfs.TestRaidDfs;
+import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.raid.RaidNode;
+
+
+public class TestReedSolomonDecoder extends TestCase {
+  final static Log LOG = LogFactory.getLog(
+                            "org.apache.hadoop.raid.TestReedSolomonDecoder");
+  final static String TEST_DIR = new File(System.getProperty("test.build.data",
+      "build/contrib/raid/test/data")).getAbsolutePath();
+  final static int NUM_DATANODES = 3;
+
+  Configuration conf;
+  MiniDFSCluster dfs = null;
+  FileSystem fileSys = null;
+
+  public void testDecoder() throws Exception {
+    mySetup();
+    int stripeSize = 10;
+    int paritySize = 4;
+    long blockSize = 8192;
+    Path file1 = new Path("/user/raidtest/file1");
+    Path recoveredFile1 = new Path("/user/raidtest/file1.recovered");
+    Path parityFile1 = new Path("/rsraid/user/raidtest/file1");
+    long crc1 = TestRaidDfs.createTestFilePartialLastBlock(fileSys, file1,
+                                                          1, 25, blockSize);
+    FileStatus file1Stat = fileSys.getFileStatus(file1);
+
+    conf.setInt("raid.rsdecoder.bufsize", 512);
+    conf.setInt("raid.rsencoder.bufsize", 512);
+
+    try {
+      // First encode the file.
+      ReedSolomonEncoder encoder = new ReedSolomonEncoder(
+        conf, stripeSize, paritySize);
+      short parityRepl = 1;
+      encoder.encodeFile(fileSys, file1, fileSys, parityFile1, parityRepl,
+        Reporter.NULL);
+
+      // Ensure there are no corrupt files yet.
+      DistributedFileSystem dfs = (DistributedFileSystem)fileSys;
+      String[] corruptFiles = RaidDFSUtil.getCorruptFiles(conf);
+      assertEquals(corruptFiles.length, 0);
+
+      // Now corrupt the file.
+      long corruptOffset = blockSize * 5;
+      FileStatus srcStat = fileSys.getFileStatus(file1);
+      LocatedBlocks locations = RaidDFSUtil.getBlockLocations(dfs,
+          file1.toUri().getPath(), 0, srcStat.getLen());
+      corruptBlock(locations.get(5).getBlock().getBlockName());
+      corruptBlock(locations.get(6).getBlock().getBlockName());
+      TestBlockFixer.reportCorruptBlocks(dfs, file1, new int[]{5, 6},
+          srcStat.getBlockSize());
+
+      // Ensure file is corrupted.
+      corruptFiles = RaidDFSUtil.getCorruptFiles(conf);
+      assertEquals(corruptFiles.length, 1);
+      assertEquals(corruptFiles[0], file1.toString());
+
+      // Fix the file.
+      ReedSolomonDecoder decoder = new ReedSolomonDecoder(
+        conf, stripeSize, paritySize);
+      decoder.decodeFile(fileSys, file1, fileSys, parityFile1,
+                corruptOffset, recoveredFile1);
+      assertTrue(TestRaidDfs.validateFile(
+                    fileSys, recoveredFile1, file1Stat.getLen(), crc1));
+    } finally {
+      myTearDown();
+    }
+  }
+
+  void corruptBlock(String blockName) throws IOException {
+    boolean corrupted = false;
+    for (int i = 0; i < NUM_DATANODES; i++) {
+      corrupted |= TestDatanodeBlockScanner.corruptReplica(blockName, i);
+    }
+    assertTrue(corrupted);
+  }
+
+  private void mySetup() throws Exception {
+
+    new File(TEST_DIR).mkdirs(); // Make sure data directory exists
+    conf = new Configuration();
+
+    // make all deletions not go through Trash
+    conf.set("fs.shell.delete.classname", "org.apache.hadoop.hdfs.DFSClient");
+
+    conf.setBoolean("dfs.permissions", false);
+
+    dfs = new MiniDFSCluster(conf, NUM_DATANODES, true, null);
+    dfs.waitActive();
+    fileSys = dfs.getFileSystem();
+    String namenode = fileSys.getUri().toString();
+    FileSystem.setDefaultUri(conf, namenode);
+  }
+
+  private void myTearDown() throws Exception {
+    if (dfs != null) { dfs.shutdown(); }
+  }
+}
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestReedSolomonEncoder.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestReedSolomonEncoder.java
new file mode 100644
index 0000000..9815591
--- /dev/null
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestReedSolomonEncoder.java
@@ -0,0 +1,104 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hadoop.raid;
+
+import java.io.File;
+
+import junit.framework.TestCase;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FilterFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
+import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
+import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.DistributedRaidFileSystem;
+import org.apache.hadoop.hdfs.TestRaidDfs;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.raid.RaidNode;
+
+
+public class TestReedSolomonEncoder extends TestCase {
+  final static Log LOG = LogFactory.getLog(
+                            "org.apache.hadoop.raid.TestReedSolomonEncoder");
+  final static String TEST_DIR = new File(System.getProperty("test.build.data",
+      "build/contrib/raid/test/data")).getAbsolutePath();
+  final static int NUM_DATANODES = 3;
+
+  Configuration conf;
+  String namenode = null;
+  MiniDFSCluster dfs = null;
+  FileSystem fileSys = null;
+
+  public void testEncoder() throws Exception {
+    mySetup();
+    int stripeSize = 10;
+    int paritySize = 4;
+    long blockSize = 8192;
+    Path file1 = new Path("/user/raidtest/file1");
+    Path parityFile1 = new Path("/rsraid/user/raidtest/file1");
+    long crc1 = TestRaidDfs.createTestFilePartialLastBlock(fileSys, file1,
+                                                          1, 25, blockSize);
+    try {
+      ReedSolomonEncoder encoder = new ReedSolomonEncoder(
+        conf, stripeSize, paritySize);
+      short parityRepl = 1;
+      encoder.encodeFile(fileSys, file1, fileSys, parityFile1, parityRepl,
+        Reporter.NULL);
+
+      FileStatus parityStat = fileSys.getFileStatus(parityFile1);
+      assertEquals(4*8192*3, parityStat.getLen());
+
+    } finally {
+      myTearDown();
+    }
+  }
+
+  private void mySetup() throws Exception {
+
+    new File(TEST_DIR).mkdirs(); // Make sure data directory exists
+    conf = new Configuration();
+
+    // make all deletions not go through Trash
+    conf.set("fs.shell.delete.classname", "org.apache.hadoop.hdfs.DFSClient");
+
+    dfs = new MiniDFSCluster(conf, NUM_DATANODES, true, null);
+    dfs.waitActive();
+    fileSys = dfs.getFileSystem();
+    namenode = fileSys.getUri().toString();
+
+    FileSystem.setDefaultUri(conf, namenode);
+
+  }
+
+  private void myTearDown() throws Exception {
+    if (dfs != null) { dfs.shutdown(); }
+  }
+}