@Test public void chooseLocalAccessibleDomainSocket() throws Exception { List<BlockWorkerInfo> workers = new ArrayList<>(); workers.add(worker(Constants.GB, "node2", "rack2")); BlockWorkerInfo workerWithDomainSocket = worker(Constants.GB, "node3", "rack3"); String domainSocketPath = "/tmp/domain/uuid-node3"; workerWithDomainSocket.getNetAddress().setDomainSocketPath(domainSocketPath); workers.add(workerWithDomainSocket); PowerMockito.mockStatic(NettyUtils.class); when(NettyUtils.isDomainSocketAccessible(eq(workerWithDomainSocket.getNetAddress()), any(AlluxioConfiguration.class))).thenReturn(true); InstancedConfiguration conf = ConfigurationTestUtils.defaults(); conf.set(PropertyKey.WORKER_DATA_SERVER_DOMAIN_SOCKET_AS_UUID, true); List<WorkerNetAddress> addresses = workers.stream() .map(worker -> worker.getNetAddress()) .filter(Objects::nonNull) .collect(Collectors.toList()); Optional<Pair<WorkerNetAddress, Boolean>> chosen = BlockLocationUtils .nearest(TieredIdentityFactory.fromString("node=node1,rack=rack2", conf), addresses, conf); assertTrue(chosen.isPresent()); assertTrue(chosen.get().getSecond()); assertEquals(domainSocketPath, chosen.get().getFirst().getDomainSocketPath()); assertEquals("node3", chosen.get().getFirst().getHost()); }
public static Optional<Pair<WorkerNetAddress, Boolean>> nearest(TieredIdentity tieredIdentity, List<WorkerNetAddress> addresses, AlluxioConfiguration conf) { if (conf.getBoolean(PropertyKey.WORKER_DATA_SERVER_DOMAIN_SOCKET_AS_UUID)) { for (WorkerNetAddress addr : addresses) { if (NettyUtils.isDomainSocketAccessible(addr, conf)) { LOG.debug("Found local worker by file system inspection of path {}", addr.getDomainSocketPath()); return Optional.of(new Pair<>(addr, true)); } } } Optional<TieredIdentity> nearestIdentity = TieredIdentityUtils.nearest(tieredIdentity, addresses.stream().map(addr -> addr.getTieredIdentity()).collect(Collectors.toList()), conf); if (!nearestIdentity.isPresent()) { return Optional.empty(); } boolean isLocal = tieredIdentity.getTier(0).getTierName().equals(Constants.LOCALITY_NODE) && tieredIdentity.topTiersMatch(nearestIdentity.get()); Optional<WorkerNetAddress> dataSource = addresses.stream() .filter(addr -> addr.getTieredIdentity().equals(nearestIdentity.get())).findFirst(); if (!dataSource.isPresent()) { return Optional.empty(); } return Optional.of(new Pair<>(dataSource.get(), isLocal)); }
BlockLocationUtils { public static Optional<Pair<WorkerNetAddress, Boolean>> nearest(TieredIdentity tieredIdentity, List<WorkerNetAddress> addresses, AlluxioConfiguration conf) { if (conf.getBoolean(PropertyKey.WORKER_DATA_SERVER_DOMAIN_SOCKET_AS_UUID)) { for (WorkerNetAddress addr : addresses) { if (NettyUtils.isDomainSocketAccessible(addr, conf)) { LOG.debug("Found local worker by file system inspection of path {}", addr.getDomainSocketPath()); return Optional.of(new Pair<>(addr, true)); } } } Optional<TieredIdentity> nearestIdentity = TieredIdentityUtils.nearest(tieredIdentity, addresses.stream().map(addr -> addr.getTieredIdentity()).collect(Collectors.toList()), conf); if (!nearestIdentity.isPresent()) { return Optional.empty(); } boolean isLocal = tieredIdentity.getTier(0).getTierName().equals(Constants.LOCALITY_NODE) && tieredIdentity.topTiersMatch(nearestIdentity.get()); Optional<WorkerNetAddress> dataSource = addresses.stream() .filter(addr -> addr.getTieredIdentity().equals(nearestIdentity.get())).findFirst(); if (!dataSource.isPresent()) { return Optional.empty(); } return Optional.of(new Pair<>(dataSource.get(), isLocal)); } }
BlockLocationUtils { public static Optional<Pair<WorkerNetAddress, Boolean>> nearest(TieredIdentity tieredIdentity, List<WorkerNetAddress> addresses, AlluxioConfiguration conf) { if (conf.getBoolean(PropertyKey.WORKER_DATA_SERVER_DOMAIN_SOCKET_AS_UUID)) { for (WorkerNetAddress addr : addresses) { if (NettyUtils.isDomainSocketAccessible(addr, conf)) { LOG.debug("Found local worker by file system inspection of path {}", addr.getDomainSocketPath()); return Optional.of(new Pair<>(addr, true)); } } } Optional<TieredIdentity> nearestIdentity = TieredIdentityUtils.nearest(tieredIdentity, addresses.stream().map(addr -> addr.getTieredIdentity()).collect(Collectors.toList()), conf); if (!nearestIdentity.isPresent()) { return Optional.empty(); } boolean isLocal = tieredIdentity.getTier(0).getTierName().equals(Constants.LOCALITY_NODE) && tieredIdentity.topTiersMatch(nearestIdentity.get()); Optional<WorkerNetAddress> dataSource = addresses.stream() .filter(addr -> addr.getTieredIdentity().equals(nearestIdentity.get())).findFirst(); if (!dataSource.isPresent()) { return Optional.empty(); } return Optional.of(new Pair<>(dataSource.get(), isLocal)); } private  BlockLocationUtils(); }
BlockLocationUtils { public static Optional<Pair<WorkerNetAddress, Boolean>> nearest(TieredIdentity tieredIdentity, List<WorkerNetAddress> addresses, AlluxioConfiguration conf) { if (conf.getBoolean(PropertyKey.WORKER_DATA_SERVER_DOMAIN_SOCKET_AS_UUID)) { for (WorkerNetAddress addr : addresses) { if (NettyUtils.isDomainSocketAccessible(addr, conf)) { LOG.debug("Found local worker by file system inspection of path {}", addr.getDomainSocketPath()); return Optional.of(new Pair<>(addr, true)); } } } Optional<TieredIdentity> nearestIdentity = TieredIdentityUtils.nearest(tieredIdentity, addresses.stream().map(addr -> addr.getTieredIdentity()).collect(Collectors.toList()), conf); if (!nearestIdentity.isPresent()) { return Optional.empty(); } boolean isLocal = tieredIdentity.getTier(0).getTierName().equals(Constants.LOCALITY_NODE) && tieredIdentity.topTiersMatch(nearestIdentity.get()); Optional<WorkerNetAddress> dataSource = addresses.stream() .filter(addr -> addr.getTieredIdentity().equals(nearestIdentity.get())).findFirst(); if (!dataSource.isPresent()) { return Optional.empty(); } return Optional.of(new Pair<>(dataSource.get(), isLocal)); } private  BlockLocationUtils(); static Optional<Pair<WorkerNetAddress, Boolean>> nearest(TieredIdentity tieredIdentity,
      List<WorkerNetAddress> addresses, AlluxioConfiguration conf); }
BlockLocationUtils { public static Optional<Pair<WorkerNetAddress, Boolean>> nearest(TieredIdentity tieredIdentity, List<WorkerNetAddress> addresses, AlluxioConfiguration conf) { if (conf.getBoolean(PropertyKey.WORKER_DATA_SERVER_DOMAIN_SOCKET_AS_UUID)) { for (WorkerNetAddress addr : addresses) { if (NettyUtils.isDomainSocketAccessible(addr, conf)) { LOG.debug("Found local worker by file system inspection of path {}", addr.getDomainSocketPath()); return Optional.of(new Pair<>(addr, true)); } } } Optional<TieredIdentity> nearestIdentity = TieredIdentityUtils.nearest(tieredIdentity, addresses.stream().map(addr -> addr.getTieredIdentity()).collect(Collectors.toList()), conf); if (!nearestIdentity.isPresent()) { return Optional.empty(); } boolean isLocal = tieredIdentity.getTier(0).getTierName().equals(Constants.LOCALITY_NODE) && tieredIdentity.topTiersMatch(nearestIdentity.get()); Optional<WorkerNetAddress> dataSource = addresses.stream() .filter(addr -> addr.getTieredIdentity().equals(nearestIdentity.get())).findFirst(); if (!dataSource.isPresent()) { return Optional.empty(); } return Optional.of(new Pair<>(dataSource.get(), isLocal)); } private  BlockLocationUtils(); static Optional<Pair<WorkerNetAddress, Boolean>> nearest(TieredIdentity tieredIdentity,
      List<WorkerNetAddress> addresses, AlluxioConfiguration conf);  }