public static boolean downloadMissingBlob ( Map conf , BlobStore blobStore , String key , Set < NimbusInfo > nimbusInfos ) throws TTransportException { NimbusClient client ; ReadableBlobMeta rbm ; ClientBlobStore remoteBlobStore ; InputStreamWithMeta in ; boolean isSuccess = false ; LOG . debug ( "Download blob NimbusInfos {}" , nimbusInfos ) ; for ( NimbusInfo nimbusInfo : nimbusInfos ) { if ( isSuccess ) { break ; } try { client = new NimbusClient ( conf , nimbusInfo . getHost ( ) , nimbusInfo . getPort ( ) , null ) ; rbm = client . getClient ( ) . getBlobMeta ( key ) ; remoteBlobStore = new NimbusBlobStore ( ) ; remoteBlobStore . setClient ( conf , client ) ; in = remoteBlobStore . getBlob ( key ) ; blobStore . createBlob ( key , in , rbm . get_settable ( ) ) ; // if key already exists while creating the blob else update it Iterator < String > keyIterator = blobStore . listKeys ( ) ; while ( keyIterator . hasNext ( ) ) { if ( keyIterator . next ( ) . equals ( key ) ) { LOG . debug ( "Success creating key, {}" , key ) ; isSuccess = true ; break ; } } } catch ( IOException exception ) { throw new RuntimeException ( exception ) ; } catch ( KeyAlreadyExistsException kae ) { LOG . info ( "KeyAlreadyExistsException Key: {} {}" , key , kae ) ; } catch ( KeyNotFoundException knf ) { // Catching and logging KeyNotFoundException because, if // there is a subsequent update and delete, the non-leader // nimbodes might throw an exception. LOG . info ( "KeyNotFoundException Key: {} {}" , key , knf ) ; } catch ( Exception exp ) { // Logging an exception while client is connecting LOG . error ( "Exception " , exp ) ; } } if ( ! isSuccess ) { LOG . error ( "Could not download blob with key " + key ) ; } return isSuccess ; }
Download missing blobs from potential nimbodes