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));
  }
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 ( ) ; 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 ) ) ; }
Creates a UFS block and initialize it with bytes read from block store.

@param context context of this request