@Test(timeout = 1000 * 60) public void fallbackOnLastChunk() throws Exception { Future<WriteSummary> expected; Future<WriteSummary> actualLocal; Future<WriteSummary> actualUfs; long blockSize = CHUNK_SIZE * 1024 + CHUNK_SIZE / 3; try (DataWriter writer = create(blockSize, CHUNK_SIZE * 1024)) { expected = writeData(writer, blockSize); expected.get(); actualLocal = getLocalWrite(mBuffer); actualUfs = getUfsWrite(mClient); } assertEquals(blockSize, expected.get().getBytes()); assertEquals(CHUNK_SIZE * 1024, actualLocal.get().getBytes()); assertEquals(blockSize - CHUNK_SIZE * 1024, actualUfs.get().getBytes()); assertEquals(expected.get().getChecksum(), actualLocal.get().getChecksum() + actualUfs.get().getChecksum()); }
public static UfsFallbackLocalFileDataWriter create(FileSystemContext context, WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options) throws IOException { try { LocalFileDataWriter localFilePacketWriter = LocalFileDataWriter.create(context, address, blockId, options); return new UfsFallbackLocalFileDataWriter(localFilePacketWriter, null, context, address, blockId, blockSize, options); } catch (ResourceExhaustedException e) { LOG.warn("Fallback to create new block {} in UFS due to a failure of insufficient space on " + "the local worker: {}", blockId, e.getMessage()); } GrpcDataWriter grpcDataWriter = GrpcDataWriter .create(context, address, blockId, blockSize, RequestType.UFS_FALLBACK_BLOCK, options); return new UfsFallbackLocalFileDataWriter(null, grpcDataWriter, context, address, blockId, blockSize, options); }
UfsFallbackLocalFileDataWriter implements DataWriter { public static UfsFallbackLocalFileDataWriter create(FileSystemContext context, WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options) throws IOException { try { LocalFileDataWriter localFilePacketWriter = LocalFileDataWriter.create(context, address, blockId, options); return new UfsFallbackLocalFileDataWriter(localFilePacketWriter, null, context, address, blockId, blockSize, options); } catch (ResourceExhaustedException e) { LOG.warn("Fallback to create new block {} in UFS due to a failure of insufficient space on " + "the local worker: {}", blockId, e.getMessage()); } GrpcDataWriter grpcDataWriter = GrpcDataWriter .create(context, address, blockId, blockSize, RequestType.UFS_FALLBACK_BLOCK, options); return new UfsFallbackLocalFileDataWriter(null, grpcDataWriter, context, address, blockId, blockSize, options); } }
UfsFallbackLocalFileDataWriter implements DataWriter { public static UfsFallbackLocalFileDataWriter create(FileSystemContext context, WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options) throws IOException { try { LocalFileDataWriter localFilePacketWriter = LocalFileDataWriter.create(context, address, blockId, options); return new UfsFallbackLocalFileDataWriter(localFilePacketWriter, null, context, address, blockId, blockSize, options); } catch (ResourceExhaustedException e) { LOG.warn("Fallback to create new block {} in UFS due to a failure of insufficient space on " + "the local worker: {}", blockId, e.getMessage()); } GrpcDataWriter grpcDataWriter = GrpcDataWriter .create(context, address, blockId, blockSize, RequestType.UFS_FALLBACK_BLOCK, options); return new UfsFallbackLocalFileDataWriter(null, grpcDataWriter, context, address, blockId, blockSize, options); } UfsFallbackLocalFileDataWriter(DataWriter localFileDataWriter,
      GrpcDataWriter grpcDataWriter, FileSystemContext context,
      final WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options); }
UfsFallbackLocalFileDataWriter implements DataWriter { public static UfsFallbackLocalFileDataWriter create(FileSystemContext context, WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options) throws IOException { try { LocalFileDataWriter localFilePacketWriter = LocalFileDataWriter.create(context, address, blockId, options); return new UfsFallbackLocalFileDataWriter(localFilePacketWriter, null, context, address, blockId, blockSize, options); } catch (ResourceExhaustedException e) { LOG.warn("Fallback to create new block {} in UFS due to a failure of insufficient space on " + "the local worker: {}", blockId, e.getMessage()); } GrpcDataWriter grpcDataWriter = GrpcDataWriter .create(context, address, blockId, blockSize, RequestType.UFS_FALLBACK_BLOCK, options); return new UfsFallbackLocalFileDataWriter(null, grpcDataWriter, context, address, blockId, blockSize, options); } UfsFallbackLocalFileDataWriter(DataWriter localFileDataWriter,
      GrpcDataWriter grpcDataWriter, FileSystemContext context,
      final WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options); static UfsFallbackLocalFileDataWriter create(FileSystemContext context,
      WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options); @Override void writeChunk(ByteBuf chunk); @Override void flush(); @Override int chunkSize(); @Override long pos(); @Override void cancel(); @Override void close(); }
UfsFallbackLocalFileDataWriter implements DataWriter { public static UfsFallbackLocalFileDataWriter create(FileSystemContext context, WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options) throws IOException { try { LocalFileDataWriter localFilePacketWriter = LocalFileDataWriter.create(context, address, blockId, options); return new UfsFallbackLocalFileDataWriter(localFilePacketWriter, null, context, address, blockId, blockSize, options); } catch (ResourceExhaustedException e) { LOG.warn("Fallback to create new block {} in UFS due to a failure of insufficient space on " + "the local worker: {}", blockId, e.getMessage()); } GrpcDataWriter grpcDataWriter = GrpcDataWriter .create(context, address, blockId, blockSize, RequestType.UFS_FALLBACK_BLOCK, options); return new UfsFallbackLocalFileDataWriter(null, grpcDataWriter, context, address, blockId, blockSize, options); } UfsFallbackLocalFileDataWriter(DataWriter localFileDataWriter,
      GrpcDataWriter grpcDataWriter, FileSystemContext context,
      final WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options); static UfsFallbackLocalFileDataWriter create(FileSystemContext context,
      WorkerNetAddress address, long blockId, long blockSize, OutStreamOptions options); @Override void writeChunk(ByteBuf chunk); @Override void flush(); @Override int chunkSize(); @Override long pos(); @Override void cancel(); @Override void close();  }