public static GrpcDataWriter create(FileSystemContext context, WorkerNetAddress address,
      long id, long length, RequestType type, OutStreamOptions options)
      throws IOException {
    long chunkSize = context.getClusterConf()
        .getBytes(PropertyKey.USER_STREAMING_WRITER_CHUNK_SIZE_BYTES);
    CloseableResource<BlockWorkerClient> grpcClient = context.acquireBlockWorkerClient(address);
    try {
      return new GrpcDataWriter(context, address, id, length, chunkSize, type, options, grpcClient);
    } catch (Exception e) {
      grpcClient.close();
      throw e;
    }
  }
@Test(timeout = 1000 * 60)
  public void writeFileManyChunks() throws Exception {
    long checksumActual;
    Future<Long> checksumExpected;
    long length = CHUNK_SIZE * 30000 + CHUNK_SIZE / 3;
    try (DataWriter writer = create(Long.MAX_VALUE)) {
      checksumExpected = writeFile(writer, length, 10, length / 3);
      checksumExpected.get();
      checksumActual = verifyWriteRequests(mClient, 10, length / 3);
    }
    assertEquals(checksumExpected.get().longValue(), checksumActual);
  }