@ Override public void onNext ( OpenLocalBlockRequest request ) { RpcUtils . streamingRPCAndLog ( LOG , new RpcUtils . StreamingRpcCallable < OpenLocalBlockResponse > ( ) { @ Override public OpenLocalBlockResponse call ( ) throws Exception { Preconditions . checkState ( mRequest == null ) ; mRequest = request ; if ( mLockId == BlockLockManager . INVALID_LOCK_ID ) { mSessionId = IdUtils . createSessionId ( ) ; // TODO(calvin): Update the locking logic so this can be done better if ( mRequest . getPromote ( ) ) { try { mWorker . moveBlock ( mSessionId , mRequest . getBlockId ( ) , mStorageTierAssoc . getAlias ( 0 ) ) ; } catch ( BlockDoesNotExistException e ) { LOG . debug ( "Block {} to promote does not exist in Alluxio: {}" , mRequest . getBlockId ( ) , e . getMessage ( ) ) ; } catch ( Exception e ) { LOG . warn ( "Failed to promote block {}: {}" , mRequest . getBlockId ( ) , e . getMessage ( ) ) ; } } mLockId = mWorker . lockBlock ( mSessionId , mRequest . getBlockId ( ) ) ; mWorker . accessBlock ( mSessionId , mRequest . getBlockId ( ) ) ; } else { LOG . warn ( "Lock block {} without releasing previous block lock {}." , mRequest . getBlockId ( ) , mLockId ) ; throw new InvalidWorkerStateException ( ExceptionMessage . LOCK_NOT_RELEASED . getMessage ( mLockId ) ) ; } OpenLocalBlockResponse response = OpenLocalBlockResponse . newBuilder ( ) . setPath ( mWorker . readBlock ( mSessionId , mRequest . getBlockId ( ) , mLockId ) ) . build ( ) ; return response ; } @ Override public void exceptionCaught ( Throwable e ) { if ( mLockId != BlockLockManager . INVALID_LOCK_ID ) { try { mWorker . unlockBlock ( mLockId ) ; } catch ( BlockDoesNotExistException ee ) { LOG . error ( "Failed to unlock block {}." , mRequest . getBlockId ( ) , e ) ; } mLockId = BlockLockManager . INVALID_LOCK_ID ; } mResponseObserver . onError ( GrpcExceptionUtils . fromThrowable ( e ) ) ; } } , "OpenBlock" , true , false , mResponseObserver , "Session=%d, Request=%s" , mSessionId , mRequest ) ; }
Handles block open request .