@Test
@SuppressWarnings("unchecked")
public void testLocalResolver() throws IOException {

  // Mock the subcluster mapping
  Configuration conf = new Configuration();
  Router router = mock(Router.class);
  StateStoreService stateStore = mock(StateStoreService.class);
  MembershipStore membership = mock(MembershipStore.class);
  when(router.getStateStore()).thenReturn(stateStore);
  when(stateStore.getRegisteredRecordStore(any(Class.class)))
      .thenReturn(membership);
  GetNamenodeRegistrationsResponse response =
      GetNamenodeRegistrationsResponse.newInstance();
  // Set the mapping for each client
  List<MembershipState> records = new LinkedList<>();
  records.add(newMembershipState("client0", "subcluster0"));
  records.add(newMembershipState("client1", "subcluster1"));
  records.add(newMembershipState("client2", "subcluster2"));
  response.setNamenodeMemberships(records);
  when(membership.getNamenodeRegistrations(
      any(GetNamenodeRegistrationsRequest.class))).thenReturn(response);
  // Mock the client resolution: it will be anything in sb
  StringBuilder sb = new StringBuilder("clientX");
  LocalResolver localResolver = new LocalResolver(conf, router);
  LocalResolver spyLocalResolver = spy(localResolver);
  doAnswer(new Answer<String>() {
    @Override
    public String answer(InvocationOnMock invocation) throws Throwable {
      return sb.toString();
    }
  }).when(spyLocalResolver).getClientAddr();
  // Add the mocks to the resolver
  MultipleDestinationMountTableResolver resolver =
      new MultipleDestinationMountTableResolver(conf, router);
  resolver.addResolver(DestinationOrder.LOCAL, spyLocalResolver);
  // We point /local to subclusters 0, 1, 2 with the local order
  Map<String, String> mapLocal = new HashMap<>();
  mapLocal.put("subcluster0", "/local");
  mapLocal.put("subcluster1", "/local");
  mapLocal.put("subcluster2", "/local");
  MountTable localEntry = MountTable.newInstance("/local", mapLocal);
  localEntry.setDestOrder(DestinationOrder.LOCAL);
  resolver.addEntry(localEntry);
  // Test first with the default destination
  PathLocation dest = resolver.getDestinationForPath("/local/file0.txt");
  assertDestination("subcluster0", dest);
  // We change the client location and verify
  setClient(sb, "client2");
  dest = resolver.getDestinationForPath("/local/file0.txt");
  assertDestination("subcluster2", dest);
  setClient(sb, "client1");
  dest = resolver.getDestinationForPath("/local/file0.txt");
  assertDestination("subcluster1", dest);
  setClient(sb, "client0");
  dest = resolver.getDestinationForPath("/local/file0.txt");
  assertDestination("subcluster0", dest);
}