/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache;

import static java.util.Arrays.asList;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.cache.RegionShortcut.LOCAL;
import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.internal.lang.SystemProperty.GEMFIRE_PREFIX;
import static org.apache.geode.management.ManagementService.getExistingManagementService;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.dunit.VM.getController;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.getVMId;
import static org.apache.geode.test.dunit.rules.DistributedRule.getDistributedSystemProperties;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.client.AllConnectionsInUseException;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.ClientMetadataService;
import org.apache.geode.cache.client.internal.ClientPartitionAdvisor;
import org.apache.geode.cache.client.internal.InternalClientCache;
import org.apache.geode.cache.execute.FunctionAdapter;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.LocatorLauncher;
import org.apache.geode.distributed.ServerLauncher;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.internal.cache.BucketAdvisor.ServerBucketProfile;
import org.apache.geode.internal.cache.execute.InternalFunctionInvocationTargetException;
import org.apache.geode.internal.cache.execute.util.TypedFunctionService;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.membership.MembershipEvent;
import org.apache.geode.management.membership.UniversalMembershipListenerAdapter;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.DUnitEnv;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.ClientServerTest;
import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;

@Category(ClientServerTest.class)
@SuppressWarnings("serial")
public class PartitionedRegionSingleHopDUnitTest implements Serializable {

  private static final String PARTITIONED_REGION_NAME = "single_hop_pr";
  private static final String ORDER_REGION_NAME = "ORDER";
  private static final String CUSTOMER_REGION_NAME = "CUSTOMER";
  private static final String SHIPMENT_REGION_NAME = "SHIPMENT";
  private static final String REPLICATE_REGION_NAME = "rr";

  private static final int LOCAL_MAX_MEMORY_DEFAULT = -1;

  private static final ServerLauncher DUMMY_SERVER = mock(ServerLauncher.class);
  private static final LocatorLauncher DUMMY_LOCATOR = mock(LocatorLauncher.class);
  private static final InternalClientCache DUMMY_CLIENT = mock(InternalClientCache.class);
  private static final InternalCache DUMMY_CACHE = mock(InternalCache.class);
  private static final AtomicReference<ServerLauncher> SERVER = new AtomicReference<>();
  private static final AtomicReference<LocatorLauncher> LOCATOR = new AtomicReference<>();
  private static final AtomicReference<InternalClientCache> CLIENT = new AtomicReference<>();
  private static final AtomicReference<InternalCache> CACHE = new AtomicReference<>();

  private static final AtomicReference<CountDownLatch> LATCH = new AtomicReference<>();

  private String diskStoreName;

  private VM vm0;
  private VM vm1;
  private VM vm2;
  private VM vm3;

  @Rule
  public DistributedRule distributedRule = new DistributedRule();
  @Rule
  public DistributedRestoreSystemProperties restoreProps = new DistributedRestoreSystemProperties();
  @Rule
  public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

  @Before
  public void setUp() {
    addIgnoredException("Connection refused");

    diskStoreName = "disk";

    vm0 = getVM(0);
    vm1 = getVM(1);
    vm2 = getVM(2);
    vm3 = getVM(3);

    for (VM vm : asList(getController(), vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> {
        CLIENT.set(DUMMY_CLIENT);
        CACHE.set(DUMMY_CACHE);
        SERVER.set(DUMMY_SERVER);
        LOCATOR.set(DUMMY_LOCATOR);
      });
    }
  }

  @After
  public void tearDown() {
    for (VM vm : asList(getController(), vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> {
        CLIENT.getAndSet(DUMMY_CLIENT).close();
        CACHE.getAndSet(DUMMY_CACHE).close();
        SERVER.getAndSet(DUMMY_SERVER).stop();
        LOCATOR.getAndSet(DUMMY_LOCATOR).stop();
      });
    }
  }

  /**
   * 2 peers 2 servers 1 accessor.No client.Should work without any exceptions.
   */
  @Test
  public void testNoClient() throws Exception {
    vm0.invoke(() -> createServer(-1, 1, 4));
    vm1.invoke(() -> createServer(-1, 1, 4));
    vm2.invoke(() -> createAccessorPeer(1, 4));
    vm3.invoke(() -> createAccessorPeer(1, 4));
    createAccessorServer(1, 4);

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

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

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

    for (VM vm : asList(getController(), vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> {
        ClientMetadataService clientMetadataService =
            getInternalCache(SERVER.get()).getClientMetadataService();

        assertThat(clientMetadataService.getClientPRMetadata_TEST_ONLY()).isEmpty();
        assertThat(clientMetadataService.getClientPartitionAttributesMap()).isEmpty();
      });
    }
  }

  /**
   * 2 AccessorServers, 2 Peers 1 Client connected to 2 AccessorServers. Hence metadata should not
   * be fetched.
   */
  @Test
  public void testClientConnectedToAccessors() {
    int port0 = vm0.invoke(() -> createAccessorServer(1, 4));
    int port1 = vm1.invoke(() -> createAccessorServer(1, 4));
    vm2.invoke(() -> createAccessorPeer(1, 4));
    vm3.invoke(() -> createAccessorPeer(1, 4));
    createClient(250, true, true, true, port0, port1);

    putIntoPartitionedRegions();
    doGets();

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();

    assertThat(clientMetadataService.getClientPRMetadata_TEST_ONLY()).isEmpty();
    assertThat(clientMetadataService.getClientPartitionAttributesMap()).isEmpty();
  }

  /**
   * 1 server 2 accesorservers 2 peers.i client connected to the server Since only 1 server hence
   * Metadata should not be fetched.
   */
  @Test
  public void testClientConnectedTo1Server() {
    int port0 = vm0.invoke(() -> createServer(-1, 1, 4));
    vm1.invoke(() -> createAccessorPeer(1, 4));
    vm2.invoke(() -> createAccessorPeer(1, 4));
    vm3.invoke(() -> createAccessorServer(1, 4));
    createClient(250, true, true, true, port0);

    putIntoPartitionedRegions();
    doGets();

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();

    assertThat(clientMetadataService.getClientPRMetadata_TEST_ONLY()).isEmpty();
    assertThat(clientMetadataService.getClientPartitionAttributesMap()).isEmpty();
  }

  /**
   * 4 servers, 1 client connected to all 4 servers. Put data, get data and make the metadata
   * stable. Now verify that metadata has all 8 buckets info. Now update and ensure the fetch
   * service is never called.
   */
  @Test
  public void testMetadataContents() {
    int port0 = vm0.invoke(() -> createServer(-1, 1, 4));
    int port1 = vm1.invoke(() -> createServer(-1, 1, 4));
    int port2 = vm2.invoke(() -> createServer(-1, 1, 4));
    int port3 = vm3.invoke(() -> createServer(-1, 1, 4));
    createClient(100, true, false, true, port0, port1, port2, port3);

    putIntoPartitionedRegions();
    doGets();

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.getRefreshTaskCount_TEST_ONLY()).isZero();
    });

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

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

    partitionedRegion.put(0, "update0");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();

    partitionedRegion.put(1, "update1");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();

    partitionedRegion.put(2, "update2");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();

    partitionedRegion.put(3, "update3");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();

    partitionedRegion.put(0, "update00");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();

    partitionedRegion.put(1, "update11");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();

    partitionedRegion.put(2, "update22");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();

    partitionedRegion.put(3, "update33");
    assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();
  }

  /**
   * 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.
   */
  @Test
  public void testMetadataServiceCallAccuracy() {
    int port0 = vm0.invoke(() -> createServer(-1, 1, 4));
    int port1 = vm1.invoke(() -> createServer(-1, 1, 4));
    vm2.invoke(() -> createClient(250, true, true, true, port0));
    createClient(250, true, true, true, port1);

    vm2.invoke(() -> doPuts(getRegion(PARTITIONED_REGION_NAME)));

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

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

    partitionedRegion.put(0, "create0");
    partitionedRegion.put(1, "create1");
    partitionedRegion.put(2, "create2");
    partitionedRegion.put(3, "create3");

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isTrue();
    });

    // make sure all fetch tasks are completed
    await().untilAsserted(() -> {
      assertThat(clientMetadataService.getRefreshTaskCount_TEST_ONLY()).isZero();
    });

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

    partitionedRegion.put(0, "create0");
    partitionedRegion.put(1, "create1");
    partitionedRegion.put(2, "create2");
    partitionedRegion.put(3, "create3");

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();
    });
  }

  @Test
  public void testMetadataServiceCallAccuracy_FromDestroyOp() {
    int port0 = vm0.invoke(() -> createServer(-1, 0, 4));
    int port1 = vm1.invoke(() -> createServer(-1, 0, 4));
    vm2.invoke(() -> createClient(250, true, true, true, port0));
    createClient(250, true, true, true, port1);

    vm2.invoke(() -> doPuts(getRegion(PARTITIONED_REGION_NAME)));

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

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

    partitionedRegion.destroy(0);
    partitionedRegion.destroy(1);
    partitionedRegion.destroy(2);
    partitionedRegion.destroy(3);

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isTrue();
    });
  }

  @Test
  public void testMetadataServiceCallAccuracy_FromGetOp() {
    int port0 = vm0.invoke(() -> createServer(-1, 0, 4));
    int port1 = vm1.invoke(() -> createServer(-1, 0, 4));
    vm2.invoke(() -> createClient(250, true, true, true, port0));
    createClient(250, true, true, true, port1);

    vm2.invoke(() -> doPuts(getRegion(PARTITIONED_REGION_NAME)));

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

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

    partitionedRegion.get(0);
    partitionedRegion.get(1);
    partitionedRegion.get(2);
    partitionedRegion.get(3);

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isTrue();
    });

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

    partitionedRegion.get(0);
    partitionedRegion.get(1);
    partitionedRegion.get(2);
    partitionedRegion.get(3);

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();
    });
  }

  @Test
  public void testSingleHopWithHA() {
    int port0 = vm0.invoke(() -> createServer(-1, 0, 8));
    int port1 = vm1.invoke(() -> createServer(-1, 0, 8));
    int port2 = vm2.invoke(() -> createServer(-1, 0, 8));
    int port3 = vm3.invoke(() -> createServer(-1, 0, 8));
    createClient(100, true, false, true, port0, port1, port2, port3);

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

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

    // put
    for (int i = 1; i <= 16; i++) {
      partitionedRegion.put(i, i);
    }

    // update
    for (int i = 1; i <= 16; i++) {
      partitionedRegion.put(i, i + 1);
    }

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isTrue();
    });

    // kill server
    vm0.invoke(this::stopServer);

    // again update
    for (int i = 1; i <= 16; i++) {
      partitionedRegion.put(i, i + 10);
    }
  }

  @Test
  public void testSingleHopWithHAWithLocator() {
    int locatorPort = vm3.invoke(this::startLocator);
    String locators = "localhost[" + locatorPort + "]";

    vm0.invoke(() -> createServer(locators, null, LOCAL_MAX_MEMORY_DEFAULT, 0, 8));
    vm1.invoke(() -> createServer(locators, null, LOCAL_MAX_MEMORY_DEFAULT, 0, 8));
    vm2.invoke(() -> createServer(locators, null, LOCAL_MAX_MEMORY_DEFAULT, 0, 8));
    createClient(250, true, true, false, locatorPort);

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

    // put
    for (int i = 1; i <= 16; i++) {
      partitionedRegion.put(i, i);
    }

    // update
    for (int i = 1; i <= 16; i++) {
      partitionedRegion.put(i, i + 1);
    }

    // kill server
    vm0.invoke(this::stopServer);

    // again update
    for (int i = 1; i <= 16; i++) {
      partitionedRegion.put(i, i + 10);
    }
  }

  @Test
  public void testNoMetadataServiceCall_ForGetOp() {
    int port0 = vm0.invoke(() -> createServer(-1, 0, 4));
    int port1 = vm1.invoke(() -> createServer(-1, 0, 4));
    vm2.invoke(() -> createClient(250, false, true, true, port0));
    createClient(250, false, true, true, port1);

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

    vm2.invoke(() -> doPuts(getRegion(PARTITIONED_REGION_NAME)));

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

    partitionedRegion.get(0);
    partitionedRegion.get(1);
    partitionedRegion.get(2);
    partitionedRegion.get(3);

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();
    });

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

    partitionedRegion.get(0);
    partitionedRegion.get(1);
    partitionedRegion.get(2);
    partitionedRegion.get(3);

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();
    });
  }

  @Test
  public void testNoMetadataServiceCall() {
    int port0 = vm0.invoke(() -> createServer(-1, 1, 4));
    int port1 = vm1.invoke(() -> createServer(-1, 1, 4));
    vm2.invoke(() -> createClient(250, false, true, true, port0));
    createClient(250, false, true, true, port1);

    vm2.invoke(() -> doPuts(getRegion(PARTITIONED_REGION_NAME)));

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();

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

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);
    partitionedRegion.put(0, "create0");
    boolean metadataRefreshed_get1 = clientMetadataService.isRefreshMetadataTestOnly();

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);
    partitionedRegion.put(1, "create1");
    boolean metadataRefreshed_get2 = clientMetadataService.isRefreshMetadataTestOnly();

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);
    partitionedRegion.put(2, "create2");
    boolean metadataRefreshed_get3 = clientMetadataService.isRefreshMetadataTestOnly();

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);
    partitionedRegion.put(3, "create3");
    boolean metadataRefreshed_get4 = clientMetadataService.isRefreshMetadataTestOnly();

    await().untilAsserted(() -> {
      assertThat(metadataRefreshed_get1).isFalse();
      assertThat(metadataRefreshed_get2).isFalse();
      assertThat(metadataRefreshed_get3).isFalse();
      assertThat(metadataRefreshed_get4).isFalse();
    });

    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

    partitionedRegion.put(0, "create0");
    partitionedRegion.put(1, "create1");
    partitionedRegion.put(2, "create2");
    partitionedRegion.put(3, "create3");

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();
    });
  }

  @Test
  public void testNoMetadataServiceCall_ForDestroyOp() {
    int port0 = vm0.invoke(() -> createServer(-1, 0, 4));
    int port1 = vm1.invoke(() -> createServer(-1, 0, 4));
    vm2.invoke(() -> createClient(250, false, true, true, port0));
    createClient(250, false, true, true, port1);

    vm2.invoke(() -> doPuts(getRegion(PARTITIONED_REGION_NAME)));

    ClientMetadataService clientMetadataService = CLIENT.get().getClientMetadataService();
    clientMetadataService.satisfyRefreshMetadata_TEST_ONLY(false);

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

    partitionedRegion.destroy(0);
    partitionedRegion.destroy(1);
    partitionedRegion.destroy(2);
    partitionedRegion.destroy(3);

    await().untilAsserted(() -> {
      assertThat(clientMetadataService.isRefreshMetadataTestOnly()).isFalse();
    });
  }

  @Test
  public void testServerLocationRemovalThroughPing() throws Exception {
    LATCH.set(new CountDownLatch(2));

    int redundantCopies = 3;
    int totalNumberOfBuckets = 4;

    int port0 = vm0.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port1 = vm1.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port2 = vm2.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port3 = vm3.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    createOldClient(100, true, false, true, port0, port1, port2, port3);

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

    putIntoPartitionedRegions();
    doGets();

    ClientMetadataService clientMetadataService = CACHE.get().getClientMetadataService();
    Map<String, ClientPartitionAdvisor> clientPRMetadata =
        clientMetadataService.getClientPRMetadata_TEST_ONLY();

    Region<Object, Object> partitionedRegion = getRegion(PARTITIONED_REGION_NAME);
    Region<Object, Object> customerRegion = getRegion(CUSTOMER_REGION_NAME);
    Region<Object, Object> orderRegion = getRegion(ORDER_REGION_NAME);
    Region<Object, Object> shipmentRegion = getRegion(SHIPMENT_REGION_NAME);

    await().untilAsserted(() -> {
      assertThat(clientPRMetadata)
          .hasSize(4)
          .containsKey(partitionedRegion.getFullPath())
          .containsKey(customerRegion.getFullPath())
          .containsKey(orderRegion.getFullPath())
          .containsKey(shipmentRegion.getFullPath());
    });

    ClientPartitionAdvisor prMetadata = clientPRMetadata.get(partitionedRegion.getFullPath());
    assertThat(prMetadata.getBucketServerLocationsMap_TEST_ONLY()).hasSize(totalNumberOfBuckets);

    for (Entry entry : prMetadata.getBucketServerLocationsMap_TEST_ONLY().entrySet()) {
      assertThat((Iterable<?>) entry.getValue()).hasSize(totalNumberOfBuckets);
    }

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

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

    doGets();

    verifyDeadServer(clientPRMetadata, customerRegion, port0, port1);
    verifyDeadServer(clientPRMetadata, partitionedRegion, port0, port1);
  }

  @Test
  public void testMetadataFetchOnlyThroughFunctions() {
    // Workaround for 52004
    addIgnoredException(InternalFunctionInvocationTargetException.class);

    int redundantCopies = 3;
    int totalNumberOfBuckets = 4;

    int port0 = vm0.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port1 = vm1.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port2 = vm2.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port3 = vm3.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    createClient(100, true, false, true, port0, port1, port2, port3);

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

    executeFunctions(partitionedRegion);

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

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

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

    await().untilAsserted(() -> {
      clientMetadataService.getClientPRMetadata((InternalRegion) partitionedRegion);
      assertThat(prMetadata.getBucketServerLocationsMap_TEST_ONLY()).hasSize(totalNumberOfBuckets);
    });
  }

  @Test
  public void testMetadataFetchOnlyThroughputAll() {
    int redundantCopies = 3;
    int totalNumberOfBuckets = 4;

    int port0 = vm0.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port1 = vm1.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port2 = vm2.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    int port3 = vm3.invoke(() -> createServer(-1, redundantCopies, totalNumberOfBuckets));
    createClient(100, true, false, true, port0, port1, port2, port3);

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

    doPutAlls(partitionedRegion);

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

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

  @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));
    });
  }

  @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();
