@Test
  public void testMetadataIsSameOnAllServersAndClients() {
    int locatorPort = DUnitEnv.get().getLocatorPort();
    String locators = "localhost[" + locatorPort + "]";

    int redundantCopies = 3;
    int totalNumberOfBuckets = 4;

    vm0.invoke(() -> createServer(locators, null, -1, redundantCopies, totalNumberOfBuckets));
    vm1.invoke(() -> createServer(locators, null, -1, redundantCopies, totalNumberOfBuckets));
    vm2.invoke(() -> createServer(locators, null, -1, redundantCopies, totalNumberOfBuckets));
    vm3.invoke(() -> createServer(locators, null, -1, redundantCopies, totalNumberOfBuckets));
    createClient(100, true, false, false, locatorPort);

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

    doManyPuts(partitionedRegion);

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

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);

    Map<String, ClientPartitionAdvisor> clientPRMetadata =
        clientMetadataService.getClientPRMetadata_TEST_ONLY();

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

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

    await().alias("expected no metadata to be refreshed").untilAsserted(() -> {
      assertThat(clientBucketMap).hasSize(totalNumberOfBuckets);
    });

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

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> verifyMetadata(clientBucketMap));
    }

    vm0.invoke(this::stopServer);
    vm1.invoke(this::stopServer);

    vm0.invoke(this::startServer);
    vm1.invoke(this::startServer);

    doManyPuts(partitionedRegion);

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

    await().atMost(2, SECONDS).untilAsserted(() -> {
      Map<String, ClientPartitionAdvisor> clientPRMetadata_await =
          clientMetadataService.getClientPRMetadata_TEST_ONLY();

      assertThat(clientPRMetadata_await)
          .hasSize(1)
          .containsKey(partitionedRegion.getFullPath());

      ClientPartitionAdvisor prMetadata_await =
          clientPRMetadata_await.get(partitionedRegion.getFullPath());
      Map<Integer, List<BucketServerLocation66>> clientBucketMap_await =
          prMetadata_await.getBucketServerLocationsMap_TEST_ONLY();

      assertThat(clientBucketMap_await).hasSize(totalNumberOfBuckets);

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

    clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);

    clientPRMetadata = clientMetadataService.getClientPRMetadata_TEST_ONLY();

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

    prMetadata = clientPRMetadata.get(partitionedRegion.getFullPath());

    Map<Integer, List<BucketServerLocation66>> clientBucketMap2 =
        prMetadata.getBucketServerLocationsMap_TEST_ONLY();

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

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

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> verifyMetadata(clientBucketMap));
    }

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

    doManyPuts(partitionedRegion);

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

    vm2.invoke(this::waitForLocalBucketsCreation);
    vm3.invoke(this::waitForLocalBucketsCreation);

    clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);
    clientPRMetadata = clientMetadataService.getClientPRMetadata_TEST_ONLY();

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

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

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

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

    await().alias("verification of metadata on all members").untilAsserted(() -> {
      vm2.invoke(() -> verifyMetadata(clientBucketMap));
      vm3.invoke(() -> verifyMetadata(clientBucketMap));
    });
  }
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