@Test(timeout = 1000 * 60) public void writeFullFile() throws Exception { long checksumActual; Future<Long> checksumExpected; long length = CHUNK_SIZE * 1024 + CHUNK_SIZE / 3; try (DataWriter writer = create(length)) { checksumExpected = writeFile(writer, length, 0, length - 1); checksumExpected.get(); checksumActual = verifyWriteRequests(mClient, 0, length - 1); } assertEquals(checksumExpected.get().longValue(), checksumActual); }
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; } }
GrpcDataWriter implements DataWriter { 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; } } }
GrpcDataWriter implements DataWriter { 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; } } private  GrpcDataWriter(FileSystemContext context, final WorkerNetAddress address, long id,
      long length, long chunkSize, RequestType type, OutStreamOptions options,
      CloseableResource<BlockWorkerClient> client); }
GrpcDataWriter implements DataWriter { 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; } } private  GrpcDataWriter(FileSystemContext context, final WorkerNetAddress address, long id,
      long length, long chunkSize, RequestType type, OutStreamOptions options,
      CloseableResource<BlockWorkerClient> client); static GrpcDataWriter create(FileSystemContext context, WorkerNetAddress address,
      long id, long length, RequestType type, OutStreamOptions options); @Override long pos(); @Override void writeChunk(final ByteBuf buf); void writeFallbackInitRequest(long pos); @Override void cancel(); @Override void flush(); @Override void close(); @Override int chunkSize(); }
GrpcDataWriter implements DataWriter { 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; } } private  GrpcDataWriter(FileSystemContext context, final WorkerNetAddress address, long id,
      long length, long chunkSize, RequestType type, OutStreamOptions options,
      CloseableResource<BlockWorkerClient> client); static GrpcDataWriter create(FileSystemContext context, WorkerNetAddress address,
      long id, long length, RequestType type, OutStreamOptions options); @Override long pos(); @Override void writeChunk(final ByteBuf buf); void writeFallbackInitRequest(long pos); @Override void cancel(); @Override void flush(); @Override void close(); @Override int chunkSize();  }