code: @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);
    }
docs: Handles request to create local block. No exceptions should be thrown.
  
  @param request a create request