@Test
  public void testClientMetadataForPersistentPrs() throws Exception {
    LATCH.set(new CountDownLatch(4));

    int locatorPort = DUnitEnv.get().getLocatorPort();

    vm0.invoke(() -> createServer("disk", -1, 3, 4));
    vm1.invoke(() -> createServer("disk", -1, 3, 4));
    vm2.invoke(() -> createServer("disk", -1, 3, 4));
    vm3.invoke(() -> createServer("disk", -1, 3, 4));

    vm3.invoke(this::putIntoPartitionedRegions);

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(this::waitForLocalBucketsCreation);
    }

    createOldClient(100, true, false, false, locatorPort);

    ManagementService managementService = getExistingManagementService(CACHE.get());
    new MemberCrashedListener(LATCH.get()).registerMembershipListener(managementService);
    ClientMetadataService clientMetadataService = CACHE.get().getClientMetadataService();

    Region<Object, Object> partitionedRegion = getRegion(PARTITIONED_REGION_NAME);

    await().until(() -> {
      try {
        clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);
      } catch (AllConnectionsInUseException ignore) {
      }

      Map<ServerLocation, Set<Integer>> serverBucketMap =
          clientMetadataService.groupByServerToAllBuckets(partitionedRegion, true);

      return serverBucketMap != null;
    });

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> {
        SERVER.getAndSet(DUMMY_SERVER).stop();
      });
    }

    LATCH.get().await(getTimeout().toMillis(), MILLISECONDS);

    AsyncInvocation<Integer> createServerOnVM3 =
        vm3.invokeAsync(() -> createServer("disk", -1, 3, 4));
    AsyncInvocation<Integer> createServerOnVM2 =
        vm2.invokeAsync(() -> createServer("disk", -1, 3, 4));
    AsyncInvocation<Integer> createServerOnVM1 =
        vm1.invokeAsync(() -> createServer("disk", -1, 3, 4));
    AsyncInvocation<Integer> createServerOnVM0 =
        vm0.invokeAsync(() -> createServer("disk", -1, 3, 4));

    createServerOnVM3.await();
    createServerOnVM2.await();
    createServerOnVM1.await();
    createServerOnVM0.await();

    await().untilAsserted(() -> {
      try {
        clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);
      } catch (AllConnectionsInUseException ignore) {
      }
      Map<ServerLocation, Set<Integer>> serverBucketMap =
          clientMetadataService.groupByServerToAllBuckets(partitionedRegion, true);

      assertThat(serverBucketMap).hasSize(4);
    });
  }
2 servers, 2 clients.One client to one server. Put from c1 to s1. Now put from c2. So since
there will be a hop at least once, fetchservice has to be triggered. Now put again from
c2.There should be no hop at all.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionSingleHopDUnitTest.java