/*
 * 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.partitioned.fixed;

import static java.util.Arrays.asList;
import static java.util.Collections.emptySet;
import static java.util.Locale.US;
import static java.util.Objects.requireNonNull;
import static org.apache.geode.cache.FixedPartitionAttributes.createFixedPartition;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.APR;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.AUG;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.DEC;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.FEB;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.JAN;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.JUL;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.JUN;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.MAR;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.MAY;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.NOV;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.OCT;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Month.SEP;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Quarter.Q1;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Quarter.Q2;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Quarter.Q3;
import static org.apache.geode.internal.cache.partitioned.fixed.FixedPartitioningWithColocationAndPersistenceDUnitTest.Quarter.Q4;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.getVMId;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.Mockito.mock;

import java.io.File;
import java.io.Serializable;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;

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.cache.CacheFactory;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.FixedPartitionAttributes;
import org.apache.geode.cache.FixedPartitionResolver;
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.RegionShortcut;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.FixedPartitionAttributesImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalDataSet;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionDataStore;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceObserver;
import org.apache.geode.test.dunit.AsyncInvocation;
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.PartitioningTest;
import org.apache.geode.test.junit.categories.PersistenceTest;
import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;

@Category({PartitioningTest.class, PersistenceTest.class})
@SuppressWarnings("serial")
public class FixedPartitioningWithColocationAndPersistenceDUnitTest implements Serializable {

  private static final String QUARTERS_REGION = "quarters";
  private static final String CUSTOMERS_REGION = "customers";
  private static final String ORDERS_REGION = "orders";
  private static final String SHIPMENTS_REGION = "shipments";
  private static final String DISK_STORE = "diskStore";

  private static final InternalCache DUMMY_CACHE = mock(InternalCache.class);

  private static final AtomicReference<InternalCache> CACHE = new AtomicReference<>(DUMMY_CACHE);
  private static final AtomicReference<File> DISK_DIR = new AtomicReference<>();

  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() {
    vm0 = getVM(0);
    vm1 = getVM(1);
    vm2 = getVM(2);
    vm3 = getVM(3);

    for (VM memberVM : asList(vm0, vm1, vm2, vm3)) {
      memberVM.invoke(() -> {
        CACHE.set(DUMMY_CACHE);
        DISK_DIR.set(temporaryFolder.newFolder("diskDir-" + getVMId()).getAbsoluteFile());
      });
    }
  }

  @After
  public void tearDown() {
    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> {
        InternalResourceManager.setResourceObserver(null);
        closeCache();
        DISK_DIR.set(null);
      });
    }
  }

  /**
   * Validates that in colocation of FPRs child region cannot specify FixedPartitionAttributes
   */
  @Test
  public void testColocation_WithFPROnChildRegion() {
    vm0.invoke(() -> {
      createCache();

      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition("Customer100", true, 2))
          .redundantCopies(0)
          .localMaxMemory(40)
          .totalNumBuckets(8)
          .create(CUSTOMERS_REGION);

      Throwable thrown = catchThrowable(() -> new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition("Order100", true, 2))
          .redundantCopies(0)
          .localMaxMemory(40)
          .totalNumBuckets(8)
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION));

      assertThat(thrown)
          .isInstanceOf(IllegalStateException.class)
          .hasMessageContaining(
              "not be specified in PartitionAttributesFactory if colocated-with is specified");
    });
  }

  /**
   * Validates that in Customer-Order-shipment colocation, Order and shipment have the
   * FixedPartitionAttributes of the parent region Customer.
   *
   * <p>
   * Put happens for all 3 regions. Colocation of the data is achieved by using a partition-resolver
   * {@link CustomerFixedPartitionResolver#getRoutingObject(EntryOperation)}
   * Also the Fixed Partitioning is achieved using same partition-resolver
   * {@link CustomerFixedPartitionResolver#getPartitionName(EntryOperation, Set)}
   *
   * <p>
   * Validation is done for the same number of the buckets on a particular member for all 3 regions.
   */
  @Test
  public void testColocation_FPRs_ChildUsingAttributesOfParent() {
    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(this::createCache);
    }

    vm0.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("10", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .create(CUSTOMERS_REGION));

    vm1.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("20", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .create(CUSTOMERS_REGION));

    vm2.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("30", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .create(CUSTOMERS_REGION));

    vm3.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("40", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .create(CUSTOMERS_REGION));

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> {
        validateFixedPartitionAttributes(ORDERS_REGION);
        validateFixedPartitionAttributes(SHIPMENTS_REGION);
      });
    }

    vm0.invoke(() -> {
      putOrdersData(40);
      putCustomersData(40);
      putShipmentsData(40);

      validateColocatedData(10);
    });

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    }
  }

  @Test
  public void testColocation_FPR_Persistence_ChildUsingAttributesOfParent() {
    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(this::createCache);
    }

    vm0.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("10", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .persistent(true)
        .create(CUSTOMERS_REGION));

    vm1.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("20", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .persistent(true)
        .create(CUSTOMERS_REGION));

    vm2.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("30", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .persistent(true)
        .create(CUSTOMERS_REGION));

    vm3.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("40", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .persistent(true)
        .create(CUSTOMERS_REGION));

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> {
        validateFixedPartitionAttributes(ORDERS_REGION);
        validateFixedPartitionAttributes(SHIPMENTS_REGION);
      });
    }

    vm0.invoke(() -> {
      putOrdersData(40);
      putCustomersData(40);
      putShipmentsData(40);

      validateColocatedData(10);
    });

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    }
  }

  /**
   * This tests validates that Customer-Order-shipment colocation with failover scenario,
   */
  @Test
  public void testColocation_FPRs_ChildUsingAttributesOfParent_HA() {
    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(this::createCache);
    }

    vm0.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("10", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .create(CUSTOMERS_REGION));

    vm1.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("20", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .create(CUSTOMERS_REGION));

    vm2.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("30", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .create(CUSTOMERS_REGION));

    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(() -> {
        validateFixedPartitionAttributes(ORDERS_REGION);
        validateFixedPartitionAttributes(SHIPMENTS_REGION);
      });
    }

    vm0.invoke(() -> {
      putCustomersData(40);
      putOrdersData(40);
      putShipmentsData(40);

      validateColocatedData(10);
    });

    vm0.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    vm1.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    vm2.invoke(() -> validatePrimaryBucketsForColocation(20, 10));


    vm2.invoke(this::closeCache);

    for (VM vm : asList(vm0, vm1)) {
      vm.invoke(() -> validatePrimaryBucketsForColocationAfterCacheClosed(15, 5));
    }

    vm2.invoke(() -> {
      RecoveryFinishedObserver observer =
          new RecoveryFinishedObserver(CUSTOMERS_REGION, ORDERS_REGION, SHIPMENTS_REGION);
      InternalResourceManager.setResourceObserver(observer);

      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition("30", true, 5))
          .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
          .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .create(CUSTOMERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION);

      await().until(observer::isRecoveryFinished);
    });

    vm0.invoke(() -> validateColocatedData(10));

    vm2.invoke(() -> validatePrimaryBucketsForColocation(15, 5));

    vm3.invoke(() -> {
      RecoveryFinishedObserver observer =
          new RecoveryFinishedObserver(CUSTOMERS_REGION, ORDERS_REGION, SHIPMENTS_REGION);
      InternalResourceManager.setResourceObserver(observer);

      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition("40", true, 5))
          .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
          .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .create(CUSTOMERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION);

      await().until(observer::isRecoveryFinished);
    });

    vm0.invoke(() -> validateColocatedData(10));

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    }
  }

  @Test
  public void testColocation_FPR_Persistence_ChildUsingAttributesOfParent_HA() {
    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(this::createCache);
    }

    vm0.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("10", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .persistent(true)
        .create(CUSTOMERS_REGION));

    vm1.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("20", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("30", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .persistent(true)
        .create(CUSTOMERS_REGION));

    vm2.invoke(() -> new RegionBuilder<>()
        .addFixedPartitionAttributes(createFixedPartition("30", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("40", true, 5))
        .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
        .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
        .redundantCopies(2)
        .localMaxMemory(50)
        .totalNumBuckets(20)
        .partitionResolver(new CustomerFixedPartitionResolver<>())
        .persistent(true)
        .create(CUSTOMERS_REGION));

    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .persistent(false)
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(() -> new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .persistent(false)
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION));
    }

    for (VM vm : asList(vm0, vm1, vm2)) {
      vm.invoke(() -> {
        validateFixedPartitionAttributes(ORDERS_REGION);
        validateFixedPartitionAttributes(SHIPMENTS_REGION);
      });
    }

    vm0.invoke(() -> {
      putCustomersData(40);
      putOrdersData(40);
      putShipmentsData(40);

      validateColocatedData(10);
    });

    vm0.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    vm1.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    vm2.invoke(() -> validatePrimaryBucketsForColocation(20, 10));

    vm2.invoke(this::closeCache);

    for (VM vm : asList(vm0, vm1)) {
      vm.invoke(() -> validatePrimaryBucketsForColocationAfterCacheClosed(15, 5));
    }

    vm2.invoke(() -> {
      RecoveryFinishedObserver observer =
          new RecoveryFinishedObserver(CUSTOMERS_REGION, ORDERS_REGION, SHIPMENTS_REGION);
      InternalResourceManager.setResourceObserver(observer);

      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition("30", true, 5))
          .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
          .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .create(CUSTOMERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION);

      await().until(observer::isRecoveryFinished);
    });

    vm0.invoke(() -> validateColocatedData(10));

    vm2.invoke(() -> validatePrimaryBucketsForColocation(15, 5));

    vm3.invoke(() -> {
      RecoveryFinishedObserver observer =
          new RecoveryFinishedObserver(CUSTOMERS_REGION, ORDERS_REGION, SHIPMENTS_REGION);
      InternalResourceManager.setResourceObserver(observer);

      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition("40", true, 5))
          .addFixedPartitionAttributes(createFixedPartition("10", false, 5))
          .addFixedPartitionAttributes(createFixedPartition("20", false, 5))
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .create(CUSTOMERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(CUSTOMERS_REGION)
          .create(ORDERS_REGION);
      new RegionBuilder<>()
          .redundantCopies(2)
          .localMaxMemory(50)
          .totalNumBuckets(20)
          .partitionResolver(new CustomerFixedPartitionResolver<>())
          .colocatedWith(ORDERS_REGION)
          .create(SHIPMENTS_REGION);

      await().until(observer::isRecoveryFinished);
    });

    vm0.invoke(() -> validateColocatedData(10));

    for (VM vm : asList(vm0, vm1, vm2, vm3)) {
      vm.invoke(() -> validatePrimaryBucketsForColocation(15, 5));
    }
  }

  /**
   * Tests validate the behavior of FPR with persistence when one member is kept alive and other
   * members goes down and come up
   */
  @Test
  public void testFPR_Persistence_OneMemberAlive() {
    vm0.invoke(() -> {
      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q1.toString(), true, 3))
          .redundantCopies(0)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);
    });

    vm1.invoke(() -> {
      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q2.toString(), true, 3))
          .redundantCopies(0)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);
    });

    vm0.invoke(() -> {
      putQuartersData(Q1);
      putQuartersData(Q2);
    });

    vm1.invoke(() -> {
      closeCache();
      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q2.toString(), true, 3))
          .redundantCopies(0)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);

      validateQuartersData(Q2);
    });

    vm2.invoke(() -> {
      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q3.toString(), true, 3))
          .redundantCopies(0)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);
    });

    vm3.invoke(() -> {
      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q4.toString(), true, 3))
          .redundantCopies(0)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);
    });

    vm0.invoke(() -> putQuartersData());

    vm0.invoke(() -> validatePrimaryData(Q1));
    vm1.invoke(() -> validatePrimaryData(Q2));
    vm2.invoke(() -> validatePrimaryData(Q3));
    vm3.invoke(() -> validatePrimaryData(Q4));

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

  /**
   * Tests validate the behavior of FPR with persistence when all members goes down and comes up.
   */
  @Test
  public void testFPR_Persistence() {
    vm0.invoke(() -> {
      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q1.toString(), true, 3))
          .addFixedPartitionAttributes(createFixedPartition(Q2.toString(), false, 3))
          .redundantCopies(1)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);
    });

    vm1.invoke(() -> {
      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q2.toString(), true, 3))
          .addFixedPartitionAttributes(createFixedPartition(Q1.toString(), false, 3))
          .redundantCopies(1)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);
    });

    vm0.invoke(() -> {
      putQuartersData(Q1);
      putQuartersData(Q2);
    });

    vm0.invoke(() -> validatePrimaryData(27 * 2, Q1, Q2));
    vm1.invoke(() -> validatePrimaryData(27 * 2, Q2, Q1));

    for (VM vm : asList(vm0, vm1)) {
      vm.invoke(() -> validatePrimaryBucketsForQuarters(6, 3));
    }

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

    vm1.invoke(() -> {
      RecoveryFinishedObserver observer = new RecoveryFinishedObserver(QUARTERS_REGION);
      InternalResourceManager.setResourceObserver(observer);

      createCache();
      new RegionBuilder<>()
          .addFixedPartitionAttributes(createFixedPartition(Q2.toString(), true, 3))
          .addFixedPartitionAttributes(createFixedPartition(Q1.toString(), false, 3))
          .redundantCopies(1)
          .localMaxMemory(40)
          .totalNumBuckets(12)
          .partitionResolver(new QuarterFixedPartitionResolver<>())
          .persistent(true)
          .create(QUARTERS_REGION);

      await().until(observer::isRecoveryFinished);
    });

    vm1.invoke(() -> {
      validateQuartersData(Q1);
      validateQuartersData(Q2);

      validatePrimaryData(27 * 2, Q2, Q1);
      validatePrimaryBucketsForQuarters(6, 6);
    });

    vm0.invoke(() -> {
      RecoveryFinishedObserver observer = new RecoveryFinishedObserver(QUARTERS_REGION);
      InternalResourceManager.setResourceObserver(observer);

      createCache();
      new RegionBuilder<>()
