void updatePipeline ( String clientName , Block oldBlock , Block newBlock , List < DatanodeID > newNodes ) throws IOException { LOG . info ( "updatePipeline(block=" + oldBlock + ", newGenerationStamp=" + newBlock . getGenerationStamp ( ) + ", newLength=" + newBlock . getNumBytes ( ) + ", newNodes=" + newNodes + ")" ) ; writeLock ( ) ; try { // check the vadility of the block and lease holder name final INodeFileUnderConstruction pendingFile = checkUCBlock ( oldBlock , clientName ) ; final Block oldBlockInfo = pendingFile . getLastBlock ( ) ; // check new GS & length: this is not expected if ( newBlock . getGenerationStamp ( ) == oldBlockInfo . getGenerationStamp ( ) ) { // we will do nothing if the GS is the same, to make this method // idempotent, this should come from the avatar failover retry. String msg = "Update " + oldBlock + " (len = " + oldBlockInfo . getNumBytes ( ) + ") to a same generation stamp: " + newBlock + " (len = " + newBlock . getNumBytes ( ) + ")" ; LOG . warn ( msg ) ; return ; } if ( newBlock . getGenerationStamp ( ) < oldBlockInfo . getGenerationStamp ( ) || newBlock . getNumBytes ( ) < oldBlockInfo . getNumBytes ( ) ) { String msg = "Update " + oldBlock + " (len = " + oldBlockInfo . getNumBytes ( ) + ") to an older state: " + newBlock + " (len = " + newBlock . getNumBytes ( ) + ")" ; LOG . warn ( msg ) ; throw new IOException ( msg ) ; } // Remove old block from blocks map. This alawys have to be done // because the generation stamp of this block is changing. blocksMap . removeBlock ( oldBlockInfo ) ; // update Last block, add it to the blocks map BlockInfo newBlockInfo = blocksMap . addINode ( newBlock , pendingFile , pendingFile . getReplication ( ) ) ; // find the Datanode Descriptor objects DatanodeDescriptor [ ] descriptors = null ; if ( ! newNodes . isEmpty ( ) ) { descriptors = new DatanodeDescriptor [ newNodes . size ( ) ] ; for ( int i = 0 ; i < newNodes . size ( ) ; i ++ ) { descriptors [ i ] = getDatanode ( newNodes . get ( i ) ) ; } } // add locations into the the INodeUnderConstruction pendingFile . setLastBlock ( newBlockInfo , descriptors ) ; // persist blocks only if append is supported String src = leaseManager . findPath ( pendingFile ) ; if ( supportAppends ) { dir . persistBlocks ( src , pendingFile ) ; } } finally { writeUnlock ( ) ; } if ( supportAppends ) { getEditLog ( ) . logSync ( ) ; } LOG . info ( "updatePipeline(" + oldBlock + ") successfully to " + newBlock ) ; }
Update a pipeline for a block under construction