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 ( ) ; } }
This reconstructs a single part file block by recovering in sequence each parity block in the part file block .