@Test public void getInStreamLocal() throws Exception { WorkerNetAddress remote = new WorkerNetAddress().setHost("remote"); WorkerNetAddress local = new WorkerNetAddress().setHost(WORKER_HOSTNAME_LOCAL); OpenLocalBlockResponse response = OpenLocalBlockResponse.newBuilder().setPath("/tmp").build(); when(mWorkerClient.openLocalBlock(any(StreamObserver.class))).thenAnswer(invocation -> { mResponseObserver = invocation.getArgument(0, StreamObserver.class); return mStreamObserver; }); doAnswer(invocation -> { mResponseObserver.onNext(response); mResponseObserver.onCompleted(); return null; }).when(mStreamObserver).onNext(any(OpenLocalBlockRequest.class)); BlockInfo info = new BlockInfo().setBlockId(BLOCK_ID).setLocations(Arrays .asList(new BlockLocation().setWorkerAddress(remote), new BlockLocation().setWorkerAddress(local))); when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(info); assertEquals(local, mBlockStore.getInStream(BLOCK_ID, new InStreamOptions( new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))), sConf)) .getAddress()); }
public BlockInStream getInStream(long blockId, InStreamOptions options) throws IOException { return getInStream(blockId, options, ImmutableMap.of()); }
AlluxioBlockStore { public BlockInStream getInStream(long blockId, InStreamOptions options) throws IOException { return getInStream(blockId, options, ImmutableMap.of()); } }
AlluxioBlockStore { public BlockInStream getInStream(long blockId, InStreamOptions options) throws IOException { return getInStream(blockId, options, ImmutableMap.of()); } @VisibleForTesting  AlluxioBlockStore(FileSystemContext context, TieredIdentity tieredIdentity); }
AlluxioBlockStore { public BlockInStream getInStream(long blockId, InStreamOptions options) throws IOException { return getInStream(blockId, options, ImmutableMap.of()); } @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 BlockInStream getInStream(long blockId, InStreamOptions options) throws IOException { return getInStream(blockId, options, ImmutableMap.of()); } @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();  }