void encodeStripe ( InputStream [ ] blocks , long blockSize , OutputStream [ ] outs , CRC32 [ ] crcOuts , Progressable reporter , boolean computeSrcChecksum , List < Integer > errorLocations ) throws IOException { configureBuffers ( blockSize ) ; int boundedBufferCapacity = 1 ; ParallelStreamReader parallelReader = new ParallelStreamReader ( reporter , blocks , bufSize , parallelism , boundedBufferCapacity , blockSize , computeSrcChecksum , outs ) ; parallelReader . start ( ) ; try { for ( long encoded = 0 ; encoded < blockSize ; encoded += bufSize ) { ParallelStreamReader . ReadResult readResult = null ; try { readResult = parallelReader . getReadResult ( ) ; } catch ( InterruptedException e ) { throw new IOException ( "Interrupted while waiting for read result" ) ; } // Cannot tolerate any IO errors. IOException readEx = readResult . getException ( ) ; if ( readEx != null ) { if ( errorLocations != null ) { errorLocations . clear ( ) ; for ( int idx : readResult . getErrorIdx ( ) ) { errorLocations . add ( idx ) ; } } throw readEx ; } code . encodeBulk ( readResult . readBufs , writeBufs ) ; reporter . progress ( ) ; // Assume each byte is independently encoded int toWrite = ( int ) Math . min ( blockSize - encoded , bufSize ) ; // Now that we have some data to write, send it to the temp files. for ( int i = 0 ; i < codec . parityLength ; i ++ ) { outs [ i ] . write ( writeBufs [ i ] , 0 , toWrite ) ; if ( crcOuts != null && crcOuts [ i ] != null ) { crcOuts [ i ] . update ( writeBufs [ i ] , 0 , toWrite ) ; } reporter . progress ( ) ; } } if ( computeSrcChecksum ) { parallelReader . collectSrcBlocksChecksum ( checksumStore ) ; } } finally { parallelReader . shutdown ( ) ; } }
Wraps around encodeStripeImpl in order to configure buffers . Having buffers of the right size is extremely important . If the the buffer size is not a divisor of the block size we may end up reading across block boundaries .