@Test public void getInStreamRemote() throws Exception { WorkerNetAddress remote1 = new WorkerNetAddress().setHost("remote1"); WorkerNetAddress remote2 = new WorkerNetAddress().setHost("remote2"); BlockInfo info = new BlockInfo().setBlockId(BLOCK_ID).setLocations(Arrays .asList(new BlockLocation().setWorkerAddress(remote1), new BlockLocation().setWorkerAddress(remote2))); when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(info); Set<WorkerNetAddress> results = new HashSet<>(); for (int i = 0; i < 40; i++) { results.add(mBlockStore.getInStream(BLOCK_ID, new InStreamOptions( new URIStatus(new FileInfo().setBlockIds(Lists.newArrayList(BLOCK_ID))), sConf)) .getAddress()); } assertEquals(Sets.newHashSet(remote1, remote2), results); }
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();  }