private void processParityHarPartBlock(FileSystem dfs, Path partFile,
      long blockOffset,
      FileStatus partFileStat,
      HarIndex harIndex,
      File localBlockFile,
      Progressable progress)
  throws IOException {
    String partName = partFile.toUri().getPath(); // Temporarily.
    partName = partName.substring(1 + partName.lastIndexOf(Path.SEPARATOR));

    OutputStream out = new FileOutputStream(localBlockFile);

    try {
      // A HAR part file block could map to several parity files. We need to
      // use all of them to recover this block.
      final long blockEnd = Math.min(blockOffset + 
          partFileStat.getBlockSize(),
          partFileStat.getLen());
      for (long offset = blockOffset; offset < blockEnd; ) {
        HarIndex.IndexEntry entry = harIndex.findEntry(partName, offset);
        if (entry == null) {
          String msg = "Lost index file has no matching index entry for " +
          partName + ":" + offset;
          LOG.warn(msg);
          throw new IOException(msg);
        }
        Path parityFile = new Path(entry.fileName);
        Encoder encoder = null;
        for (Codec codec : Codec.getCodecs()) {
          if (isParityFile(parityFile, codec)) {
            encoder = new Encoder(getConf(), codec);
          }
        }
        if (encoder == null) {
          String msg = "Could not figure out codec correctly for " + parityFile;
          LOG.warn(msg);
          throw new IOException(msg);
        }
        Path srcFile = RaidUtils.sourcePathFromParityPath(parityFile, dfs);
        if (null == srcFile) {
          String msg = "Can not find the source path for parity file: " +
                          parityFile;
          LOG.warn(msg);
          throw new IOException(msg);
        }
        FileStatus srcStat = dfs.getFileStatus(srcFile);
        if (srcStat.getModificationTime() != entry.mtime) {
          String msg = "Modification times of " + parityFile + " and " +
          srcFile + " do not match.";
          LOG.warn(msg);
          throw new IOException(msg);
        }
        long lostOffsetInParity = offset - entry.startOffset;
        LOG.info(partFile + ":" + offset + " maps to " +
            parityFile + ":" + lostOffsetInParity +
            " and will be recovered from " + srcFile);
        encoder.recoverParityBlockToStream(dfs, srcStat,
            srcStat.getBlockSize(), parityFile,
            lostOffsetInParity, out, progress);
        // Finished recovery of one parity block. Since a parity block has the
        // same size as a source block, we can move offset by source block 
        // size.
        offset += srcStat.getBlockSize();
        LOG.info("Recovered " + srcStat.getBlockSize() + " part file bytes ");
        if (offset > blockEnd) {
          String msg =
            "Recovered block spills across part file blocks. Cannot continue";
          throw new IOException(msg);
        }
        progress.progress();
      }
    } finally {
      out.close();
    }
  }
private void processParityHarPartBlock ( FileSystem dfs , Path partFile , long blockOffset , FileStatus partFileStat , HarIndex harIndex , File localBlockFile , Progressable progress ) throws IOException { String partName = partFile . toUri ( ) . getPath ( ) ; partName = partName . substring ( 1 + partName . lastIndexOf ( Path . SEPARATOR ) ) ; OutputStream out = new FileOutputStream ( localBlockFile ) ; try { final long blockEnd = Math . min ( blockOffset + partFileStat . getBlockSize ( ) , partFileStat . getLen ( ) ) ; for ( long offset = blockOffset ; offset < blockEnd ; ) { HarIndex . IndexEntry entry = harIndex . findEntry ( partName , offset ) ; if ( entry == null ) { String msg = "Lost index file has no matching index entry for " + partName + ":" + offset ; LOG . warn ( msg ) ; throw new IOException ( msg ) ; } Path parityFile = new Path ( entry . fileName ) ; Encoder encoder = null ; for ( Codec codec : Codec . getCodecs ( ) ) { if ( isParityFile ( parityFile , codec ) ) { encoder = new Encoder ( getConf ( ) , codec ) ; } } if ( encoder == null ) { String msg = "Could not figure out codec correctly for " + parityFile ; LOG . warn ( msg ) ; throw new IOException ( msg ) ; } Path srcFile = RaidUtils . sourcePathFromParityPath ( parityFile , dfs ) ; if ( null == srcFile ) { String msg = "Can not find the source path for parity file: " + parityFile ; LOG . warn ( msg ) ; throw new IOException ( msg ) ; } FileStatus srcStat = dfs . getFileStatus ( srcFile ) ; if ( srcStat . getModificationTime ( ) != entry . mtime ) { String msg = "Modification times of " + parityFile + " and " + srcFile + " do not match." ; LOG . warn ( msg ) ; throw new IOException ( msg ) ; } long lostOffsetInParity = offset - entry . startOffset ; LOG . info ( partFile + ":" + offset + " maps to " + parityFile + ":" + lostOffsetInParity + " and will be recovered from " + srcFile ) ; encoder . recoverParityBlockToStream ( dfs , srcStat , srcStat . getBlockSize ( ) , parityFile , lostOffsetInParity , out , progress ) ; offset += srcStat . getBlockSize ( ) ; LOG . info ( "Recovered " + srcStat . getBlockSize ( ) + " part file bytes " ) ; if ( offset > blockEnd ) { String msg = "Recovered block spills across part file blocks. Cannot continue" ; throw new IOException ( msg ) ; } progress . progress ( ) ; } } finally { out . close ( ) ; } }
This reconstructs a single part file block by recovering in sequence each
parity block in the part file block.