code: private void createUfsBlock(BlockWriteRequestContext context)
        throws Exception {
      BlockWriteRequest request = context.getRequest();
      Protocol.CreateUfsBlockOptions createUfsBlockOptions = request.getCreateUfsBlockOptions();
      UfsManager.UfsClient ufsClient = mUfsManager.get(createUfsBlockOptions.getMountId());
      alluxio.resource.CloseableResource<UnderFileSystem> ufsResource =
          ufsClient.acquireUfsResource();
      context.setUfsResource(ufsResource);
      String ufsString = MetricsSystem.escape(ufsClient.getUfsMountPointUri());
      String ufsPath = BlockUtils.getUfsBlockPath(ufsClient, request.getId());
      UnderFileSystem ufs = ufsResource.get();
      // Set the atomic flag to be true to ensure only the creation of this file is atomic on close.
      OutputStream ufsOutputStream =
          ufs.create(ufsPath,
              CreateOptions.defaults(ServerConfiguration.global()).setEnsureAtomic(true)
                  .setCreateParent(true));
      context.setOutputStream(ufsOutputStream);
      context.setUfsPath(ufsPath);
  
      String counterName = Metric.getMetricNameWithTags(WorkerMetrics.BYTES_WRITTEN_UFS,
          WorkerMetrics.TAG_UFS, ufsString);
      String meterName = Metric.getMetricNameWithTags(WorkerMetrics.BYTES_WRITTEN_UFS_THROUGHPUT,
          WorkerMetrics.TAG_UFS, ufsString);
      context.setCounter(MetricsSystem.counter(counterName));
      context.setMeter(MetricsSystem.meter(meterName));
    }
docs: Creates a UFS block and initialize it with bytes read from block store.
  
  @param context context of this request