@ Override public void onNext ( CreateLocalBlockRequest request ) { final String methodName = request . getOnlyReserveSpace ( ) ? "ReserveSpace" : "CreateBlock" ; RpcUtils . streamingRPCAndLog ( LOG , new RpcUtils . StreamingRpcCallable < CreateLocalBlockResponse > ( ) { @ Override public CreateLocalBlockResponse call ( ) throws Exception { if ( request . getOnlyReserveSpace ( ) ) { mBlockWorker . requestSpace ( mSessionId , request . getBlockId ( ) , request . getSpaceToReserve ( ) ) ; return CreateLocalBlockResponse . newBuilder ( ) . build ( ) ; } else { Preconditions . checkState ( mRequest == null ) ; mRequest = request ; if ( mSessionId == INVALID_SESSION_ID ) { mSessionId = IdUtils . createSessionId ( ) ; String path = mBlockWorker . createBlock ( mSessionId , request . getBlockId ( ) , mStorageTierAssoc . getAlias ( request . getTier ( ) ) , request . getSpaceToReserve ( ) ) ; CreateLocalBlockResponse response = CreateLocalBlockResponse . newBuilder ( ) . setPath ( path ) . build ( ) ; return response ; } else { LOG . warn ( "Create block {} without closing the previous session {}." , request . getBlockId ( ) , mSessionId ) ; throw new InvalidWorkerStateException ( ExceptionMessage . SESSION_NOT_CLOSED . getMessage ( mSessionId ) ) ; } } } @ Override public void exceptionCaught ( Throwable throwable ) { if ( mSessionId != INVALID_SESSION_ID ) { // In case the client is a UfsFallbackDataWriter, DO NOT clean the temp blocks. if ( throwable instanceof alluxio . exception . WorkerOutOfSpaceException && request . hasCleanupOnFailure ( ) && ! request . getCleanupOnFailure ( ) ) { mResponseObserver . onError ( GrpcExceptionUtils . fromThrowable ( throwable ) ) ; return ; } mBlockWorker . cleanupSession ( mSessionId ) ; mSessionId = INVALID_SESSION_ID ; } mResponseObserver . onError ( GrpcExceptionUtils . fromThrowable ( throwable ) ) ; } } , methodName , true , false , mResponseObserver , "Session=%d, Request=%s" , mSessionId , request ) ; }
Handles request to create local block . No exceptions should be thrown .