@Test public void getInStreamInAlluxioWhenCreateStreamIsFailed() throws Exception { int workerCount = 5; boolean persisted = false; int[] blockLocations = new int[]{2, 3, 4}; Map<Integer, Long> failedWorkers = ImmutableMap.of( 0, 3L, 1, 1L, 3, 2L); int expectedWorker = 2; WorkerNetAddress[] workers = new WorkerNetAddress[workerCount]; for (int i = 0; i < workers.length - 1; i++) { workers[i] = new WorkerNetAddress().setHost(String.format("worker-%d", i)); } workers[workers.length - 1] = new WorkerNetAddress().setHost(WORKER_HOSTNAME_LOCAL); when(mContext.acquireBlockWorkerClient(WORKER_NET_ADDRESS_LOCAL)) .thenThrow(new UnavailableException("failed to connect to " + WORKER_NET_ADDRESS_LOCAL.getHost())); BlockInfo info = new BlockInfo().setBlockId(BLOCK_ID) .setLocations(Arrays.stream(blockLocations).mapToObj(x -> new BlockLocation().setWorkerAddress(workers[x])).collect(Collectors.toList())); URIStatus dummyStatus = new URIStatus(new FileInfo().setPersisted(persisted) .setBlockIds(Collections.singletonList(BLOCK_ID)) .setFileBlockInfos(Collections.singletonList(new FileBlockInfo().setBlockInfo(info)))); BlockLocationPolicy mockPolicy = mock(BlockLocationPolicy.class); when(mockPolicy.getWorker(any())).thenAnswer(arg -> arg .getArgument(0, GetWorkerOptions.class).getBlockWorkerInfos().iterator().next() .getNetAddress()); InStreamOptions options = new InStreamOptions(dummyStatus, FileSystemOptions.openFileDefaults(sConf), sConf); options.setUfsReadLocationPolicy(mockPolicy); when(mMasterClient.getBlockInfo(BLOCK_ID)).thenReturn(info); when(mContext.getCachedWorkers()).thenReturn( Arrays.stream(workers) .map(x -> new BlockWorkerInfo(x, -1, -1)).collect((Collectors.toList()))); Map<WorkerNetAddress, Long> failedWorkerAddresses = failedWorkers.entrySet().stream() .map(x -> new AbstractMap.SimpleImmutableEntry<>(workers[x.getKey()], x.getValue())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); BlockInStream inStream = null; int i = 2; while (i-- > 0) { try { inStream = mBlockStore.getInStream(BLOCK_ID, options, failedWorkerAddresses); } catch (Exception e) { } } assertEquals(workers[expectedWorker], inStream.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();  }