@Test
  public void testMetadataIsSameOnAllServersAndClientsHA() {
    int totalNumberOfBuckets = 4;

    int port0 = vm0.invoke(() -> createServer(-1, 2, totalNumberOfBuckets));
    int port1 = vm1.invoke(() -> createServer(-1, 2, totalNumberOfBuckets));
    createClient(100, true, false, true, port0, port1, port0, port1);

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

    doManyPuts(partitionedRegion);

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);
    Map<String, ClientPartitionAdvisor> clientPRMetadata =
        clientMetadataService.getClientPRMetadata_TEST_ONLY();

    await().untilAsserted(() -> {
      assertThat(clientPRMetadata)
          .hasSize(1)
          .containsKey(partitionedRegion.getFullPath());
    });

    vm0.invoke(() -> {
      PartitionedRegion pr = (PartitionedRegion) getRegion(PARTITIONED_REGION_NAME);
      pr.getRegionAdvisor().getAllClientBucketProfilesTest();
    });
    vm1.invoke(() -> {
      PartitionedRegion pr = (PartitionedRegion) getRegion(PARTITIONED_REGION_NAME);
      pr.getRegionAdvisor().getAllClientBucketProfilesTest();
    });

    ClientPartitionAdvisor prMetadata = clientPRMetadata.get(partitionedRegion.getFullPath());
    Map<Integer, List<BucketServerLocation66>> clientBucketMap =
        prMetadata.getBucketServerLocationsMap_TEST_ONLY();

    await().untilAsserted(() -> {
      assertThat(clientBucketMap).hasSize(totalNumberOfBuckets);
    });

    for (Entry<Integer, List<BucketServerLocation66>> entry : clientBucketMap.entrySet()) {
      assertThat(entry.getValue()).hasSize(2);
    }

    vm0.invoke(() -> verifyMetadata(clientBucketMap));
    vm1.invoke(() -> verifyMetadata(clientBucketMap));

    vm0.invoke(this::stopServer);

    doManyPuts(partitionedRegion);

    clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);

    assertThat(clientBucketMap).hasSize(totalNumberOfBuckets);

    for (Entry<Integer, List<BucketServerLocation66>> entry : clientBucketMap.entrySet()) {
      assertThat(entry.getValue()).hasSize(1);
    }

    assertThat(clientPRMetadata)
        .hasSize(1)
        .containsKey(partitionedRegion.getFullPath());
    assertThat(clientBucketMap)
        .hasSize(totalNumberOfBuckets);

    await().untilAsserted(() -> {
      for (Entry<Integer, List<BucketServerLocation66>> entry : clientBucketMap.entrySet()) {
        assertThat(entry.getValue()).hasSize(1);
      }
    });
  }
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