@Test public void getOutStreamLocal() throws Exception { File file = File.createTempFile("test", ".tmp"); CreateLocalBlockResponse response = CreateLocalBlockResponse.newBuilder() .setPath(file.getAbsolutePath()).build(); when(mWorkerClient.createLocalBlock(any(StreamObserver.class))) .thenAnswer(new Answer() { public Object answer(InvocationOnMock invocation) { StreamObserver<CreateLocalBlockResponse> observer = invocation.getArgument(0, StreamObserver.class); observer.onNext(response); return mStreamObserver; } }); OutStreamOptions options = OutStreamOptions.defaults(mClientContext).setBlockSizeBytes(BLOCK_LENGTH).setLocationPolicy( new MockBlockLocationPolicy(Lists.newArrayList(WORKER_NET_ADDRESS_LOCAL))) .setWriteType(WriteType.MUST_CACHE); BlockOutStream stream = mBlockStore.getOutStream(BLOCK_ID, BLOCK_LENGTH, options); assertEquals(WORKER_NET_ADDRESS_LOCAL, stream.getAddress()); }
public BlockOutStream getOutStream(long blockId, long blockSize, WorkerNetAddress address, OutStreamOptions options) throws IOException { if (blockSize == -1) { try (CloseableResource<BlockMasterClient> blockMasterClientResource = mContext.acquireBlockMasterClientResource()) { blockSize = blockMasterClientResource.get().getBlockInfo(blockId).getLength(); } } if (address == null) { throw new ResourceExhaustedException(ExceptionMessage.NO_SPACE_FOR_BLOCK_ON_WORKER .getMessage(FormatUtils.getSizeFromBytes(blockSize))); } LOG.debug("Create block outstream for {} of block size {} at address {}, using options: {}", blockId, blockSize, address, options); return BlockOutStream.create(mContext, blockId, blockSize, address, options); }
AlluxioBlockStore { public BlockOutStream getOutStream(long blockId, long blockSize, WorkerNetAddress address, OutStreamOptions options) throws IOException { if (blockSize == -1) { try (CloseableResource<BlockMasterClient> blockMasterClientResource = mContext.acquireBlockMasterClientResource()) { blockSize = blockMasterClientResource.get().getBlockInfo(blockId).getLength(); } } if (address == null) { throw new ResourceExhaustedException(ExceptionMessage.NO_SPACE_FOR_BLOCK_ON_WORKER .getMessage(FormatUtils.getSizeFromBytes(blockSize))); } LOG.debug("Create block outstream for {} of block size {} at address {}, using options: {}", blockId, blockSize, address, options); return BlockOutStream.create(mContext, blockId, blockSize, address, options); } }
AlluxioBlockStore { public BlockOutStream getOutStream(long blockId, long blockSize, WorkerNetAddress address, OutStreamOptions options) throws IOException { if (blockSize == -1) { try (CloseableResource<BlockMasterClient> blockMasterClientResource = mContext.acquireBlockMasterClientResource()) { blockSize = blockMasterClientResource.get().getBlockInfo(blockId).getLength(); } } if (address == null) { throw new ResourceExhaustedException(ExceptionMessage.NO_SPACE_FOR_BLOCK_ON_WORKER .getMessage(FormatUtils.getSizeFromBytes(blockSize))); } LOG.debug("Create block outstream for {} of block size {} at address {}, using options: {}", blockId, blockSize, address, options); return BlockOutStream.create(mContext, blockId, blockSize, address, options); } @VisibleForTesting  AlluxioBlockStore(FileSystemContext context, TieredIdentity tieredIdentity); }
AlluxioBlockStore { public BlockOutStream getOutStream(long blockId, long blockSize, WorkerNetAddress address, OutStreamOptions options) throws IOException { if (blockSize == -1) { try (CloseableResource<BlockMasterClient> blockMasterClientResource = mContext.acquireBlockMasterClientResource()) { blockSize = blockMasterClientResource.get().getBlockInfo(blockId).getLength(); } } if (address == null) { throw new ResourceExhaustedException(ExceptionMessage.NO_SPACE_FOR_BLOCK_ON_WORKER .getMessage(FormatUtils.getSizeFromBytes(blockSize))); } LOG.debug("Create block outstream for {} of block size {} at address {}, using options: {}", blockId, blockSize, address, options); return BlockOutStream.create(mContext, blockId, blockSize, address, options); } @VisibleForTesting  AlluxioBlockStore(FileSystemContext context, TieredIdentity tieredIdentity); static AlluxioBlockStore create(FileSystemContext context); BlockInfo getInfo(long blockId); BlockInStream getInStream(long blockId, InStreamOptions options); BlockInStream getInStream(long blockId, InStreamOptions options,
      Map<WorkerNetAddress, Long> failedWorkers); BlockInStream getInStream(BlockInfo info, InStreamOptions options,
      Map<WorkerNetAddress, Long> failedWorkers); BlockOutStream getOutStream(long blockId, long blockSize, WorkerNetAddress address,
      OutStreamOptions options); BlockOutStream getOutStream(long blockId, long blockSize, OutStreamOptions options); long getCapacityBytes(); long getUsedBytes(); }
AlluxioBlockStore { public BlockOutStream getOutStream(long blockId, long blockSize, WorkerNetAddress address, OutStreamOptions options) throws IOException { if (blockSize == -1) { try (CloseableResource<BlockMasterClient> blockMasterClientResource = mContext.acquireBlockMasterClientResource()) { blockSize = blockMasterClientResource.get().getBlockInfo(blockId).getLength(); } } if (address == null) { throw new ResourceExhaustedException(ExceptionMessage.NO_SPACE_FOR_BLOCK_ON_WORKER .getMessage(FormatUtils.getSizeFromBytes(blockSize))); } LOG.debug("Create block outstream for {} of block size {} at address {}, using options: {}", blockId, blockSize, address, options); return BlockOutStream.create(mContext, blockId, blockSize, address, options); } @VisibleForTesting  AlluxioBlockStore(FileSystemContext context, TieredIdentity tieredIdentity); static AlluxioBlockStore create(FileSystemContext context); BlockInfo getInfo(long blockId); BlockInStream getInStream(long blockId, InStreamOptions options); BlockInStream getInStream(long blockId, InStreamOptions options,
      Map<WorkerNetAddress, Long> failedWorkers); BlockInStream getInStream(BlockInfo info, InStreamOptions options,
      Map<WorkerNetAddress, Long> failedWorkers); BlockOutStream getOutStream(long blockId, long blockSize, WorkerNetAddress address,
      OutStreamOptions options); BlockOutStream getOutStream(long blockId, long blockSize, OutStreamOptions options); long getCapacityBytes(); long getUsedBytes();  }