/*
 * 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;

import static java.util.Arrays.asList;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.geode.cache.RegionShortcut.PARTITION;
import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
import static org.apache.geode.cache.partition.PartitionRegionHelper.getPartitionRegionInfo;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.internal.cache.ColocationHelper.getAllColocationRegions;
import static org.apache.geode.internal.cache.partitioned.colocation.ColocationLoggerFactory.COLOCATION_LOGGER_FACTORY_PROPERTY;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Disconnect.disconnectAllFromDS;
import static org.apache.geode.test.dunit.DistributedTestUtils.getLocators;
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.VM.toArray;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.verify;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;

import junitparams.Parameters;
import junitparams.naming.TestCaseName;
import org.apache.logging.log4j.Logger;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.stubbing.Answer;

import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.control.RebalanceResults;
import org.apache.geode.cache.partition.PartitionRegionInfo;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.DistributionMessageObserver;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InitialImageOperation.RequestImageMessage;
import org.apache.geode.internal.cache.InternalCache;
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.internal.cache.partitioned.colocation.ColocationLogger;
import org.apache.geode.internal.cache.partitioned.colocation.ColocationLoggerFactory;
import org.apache.geode.internal.cache.partitioned.colocation.SingleThreadColocationLogger;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.SerializableCallable;
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.RegionsTest;
import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;
import org.apache.geode.test.junit.runners.GeodeParamsRunner;

@Category(RegionsTest.class)
@RunWith(GeodeParamsRunner.class)
@SuppressWarnings("serial")
public class PersistentColocatedPartitionedRegionDistributedTest implements Serializable {

  private static final long TIMEOUT_MILLIS = GeodeAwaitility.getTimeout().toMillis();

  private static final long DEFAULT_RECOVERY_DELAY = -1;
  private static final int DEFAULT_REDUNDANT_COPIES = 0;
  private static final long DEFAULT_STARTUP_RECOVERY_DELAY = 0;

  private static final int NUM_BUCKETS_TO_CREATE = 15;

  private static final String PATTERN_FOR_MISSING_CHILD_LOG =
      "(?s)Persistent data recovery for region .*is prevented by offline colocated region.*";

  private static volatile InternalCache cache;
  private static volatile CountDownLatch latch;

  private final transient List<AsyncInvocation<Void>> asyncInvocations = new ArrayList<>();

  private String locators;
  private String regionName;
  private String childRegionName1;
  private String childRegionName2;
  private String diskStoreName1;
  private String diskStoreName2;

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

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

  @Before
  public void setUp() {
    locators = getLocators();
    regionName = getClass().getSimpleName() + "_region";
    childRegionName1 = "region2";
    childRegionName2 = "region3";
    diskStoreName1 = "disk1";
    diskStoreName2 = "disk2";

    vm0 = getVM(0);
    vm1 = getVM(1);
    vm2 = getVM(2);
  }

  @After
  public void tearDown() {
    for (VM vm : toArray(vm0, vm1, vm2, getController())) {
      vm.invoke(() -> {
        DistributionMessageObserver.setInstance(null);
        tearDownPartitionedRegionObserver();

        while (latch != null && latch.getCount() > 0) {
          latch.countDown();
        }

        closeCache();
      });
    }
    disconnectAllFromDS();
  }

  /**
   * Testing that we can colocate persistent PRs
   */
  @Test
  public void testColocatedPRs() throws Exception {
    for (VM vm : toArray(vm0, vm1, vm2)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR(regionName, childRegionName2, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
      createData(childRegionName2, "c");
    });

    Map<Integer, Set<Integer>> bucketIdsInVM = new HashMap<>();
    for (VM vm : toArray(vm0, vm1, vm2)) {
      bucketIdsInVM.put(vm.getId(), vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(getBucketIds(childRegionName1)).isEqualTo(bucketIds);
        assertThat(getBucketIds(childRegionName2)).isEqualTo(bucketIds);
        return bucketIds;
      }));
    }

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

    for (VM vm : toArray(vm0, vm1, vm2)) {
      addAsync(vm.invokeAsync(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR(regionName, childRegionName2, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      }));
    }
    awaitAllAsync();

    // The secondary buckets can be recovered asynchronously, so wait for them to come back.
    for (VM vm : toArray(vm0, vm1)) {
      Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
      vm.invoke(() -> {
        waitForBuckets(regionName, bucketIds);
        waitForBuckets(childRegionName1, bucketIds);
      });
    }

    vm0.invoke(() -> {
      validateData(regionName, "a");
      validateData(childRegionName1, "b");

      // region 3 didn't have persistent data, so it nothing should be recovered
      validateData(childRegionName2, null);
      // Make sure can do a put in all of the buckets in region 3
      createData(childRegionName2, "c");
      // Now all of those buckets should exist.
      validateData(childRegionName2, "c");
    });

    // The region 3 buckets should be restored in the appropriate places.
    for (VM vm : toArray(vm0, vm1, vm2)) {
      Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
      assertThat(vm.invoke(() -> getBucketIds(childRegionName2))).isEqualTo(bucketIds);
    }
  }

  /**
   * Testing that missing colocated persistent PRs are logged as warning
   */
  @Test
  public void testMissingColocatedParentPR() {
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
    });

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        assertThat(getBucketIds(childRegionName1)).isEqualTo(bucketIds);
      });
    }

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

    // The following should fail immediately with ISE on vm0,
    // it's not necessary to also try the operation on vm1.
    vm0.invoke("createPRsMissingParentRegion", () -> {
      createCache();

      Throwable thrown =
          catchThrowable(
              () -> createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
                  DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES,
                  DEFAULT_STARTUP_RECOVERY_DELAY));

      assertThat(thrown)
          .isInstanceOf(IllegalStateException.class)
          .hasMessageContaining("Region specified in 'colocated-with'");
    });
  }

  /**
   * Testing that parent colocated persistent PRs only missing on local member throws exception
   */
  @Test
  public void testMissingColocatedParentPRWherePRConfigExists() throws Exception {
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
    });

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        assertThat(getBucketIds(childRegionName1)).isEqualTo(bucketIds);
      });
    }

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

    vm1.invoke(() -> {
      latch = new CountDownLatch(1);
    });

    AsyncInvocation<Void> createPRsInVM0 = vm0.invokeAsync(() -> {
      createCache();
      createDiskStore(diskStoreName1);
      createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
          DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);

      vm1.invoke(() -> latch.countDown());

      createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
          DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
    });

    vm1.invoke(() -> {
      // this thread delays the attempted creation on the local member of colocated child region
      // when parent doesn't exist. The delay is so that both parent and child regions will be
      // created on another member and the PR root config will have an entry for the parent region.
      createCache();
      createDiskStore(diskStoreName1);

      latch.await(TIMEOUT_MILLIS, MILLISECONDS);

      Throwable thrown =
          catchThrowable(
              () -> createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
                  DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES,
                  DEFAULT_STARTUP_RECOVERY_DELAY));

      assertThat(thrown)
          .isInstanceOf(IllegalStateException.class)
          .hasMessageMatching("Region specified in 'colocated-with'.*");

      createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
          DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
          DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
    });

    createPRsInVM0.await();
  }

  /**
   * Testing that missing colocated child persistent PRs are logged as warning
   */
  @Test
  public void testMissingColocatedChildPRDueToDelayedStart() throws Exception {
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
    });

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        assertThat(getBucketIds(childRegionName1)).isEqualTo(bucketIds);
      });
    }

    AsyncInvocation<Void> createMissingChildPRInVM0 =
        vm0.invokeAsync(() -> validateColocationLogger_withChildRegion(1));
    AsyncInvocation<Void> createMissingChildPRInVM1 =
        vm1.invokeAsync(() -> validateColocationLogger_withChildRegion(1));

    createMissingChildPRInVM0.await();
    createMissingChildPRInVM1.await();
  }

  /**
   * Testing that missing colocated child persistent PRs are logged as warning
   */
  @Test
  public void testMissingColocatedChildPR() throws Exception {
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
    });

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        assertThat(getBucketIds(childRegionName1)).isEqualTo(bucketIds);
      });
    }

    AsyncInvocation<Void> createPRWithMissingChildInVM0 =
        vm0.invokeAsync(() -> validateColocationLogger_withMissingChildRegion(2));
    AsyncInvocation<Void> createPRWithMissingChildInVM1 =
        vm1.invokeAsync(() -> validateColocationLogger_withMissingChildRegion(2));

    createPRWithMissingChildInVM0.await();
    createPRWithMissingChildInVM1.await();
  }

  /**
   * Test that when there is more than one missing colocated child persistent PRs for a region all
   * missing regions are logged in the warning.
   */
  @Test
  public void testMultipleColocatedChildPRsMissing() throws Exception {
    int childPRCount = 2;

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);

        for (int i = 2; i < childPRCount + 2; ++i) {
          createChildPR_withPersistence(regionName, "region" + i, diskStoreName1,
              DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES,
              DEFAULT_STARTUP_RECOVERY_DELAY);
        }
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
      createData(childRegionName1, "c");
    });

    Map<Integer, Set<Integer>> bucketIdsInVM = new HashMap<>();
    for (VM vm : toArray(vm0, vm1)) {
      bucketIdsInVM.put(vm.getId(), vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        return bucketIds;
      }));
    }

    for (int i = 2; i < childPRCount + 2; ++i) {
      String childRegionName = "region" + i;
      for (VM vm : toArray(vm0, vm1)) {
        Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
        assertThat(vm.invoke(() -> getBucketIds(childRegionName))).isEqualTo(bucketIds);
      }
    }

    AsyncInvocation<Void> createPRWithMissingChildInVM0 =
        vm0.invokeAsync(() -> validateColocationLogger_withMissingChildRegion(2));
    AsyncInvocation<Void> createPRWithMissingChildInVM1 =
        vm1.invokeAsync(() -> validateColocationLogger_withMissingChildRegion(2));

    createPRWithMissingChildInVM0.await();
    createPRWithMissingChildInVM1.await();
  }

  /**
   * Test that when there is more than one missing colocated child persistent PRs for a region all
   * missing regions are logged in the warning. Verifies that as regions are created they no longer
   * appear in the warning.
   */
  @Test // GEODE-7513
  public void testMultipleColocatedChildPRsMissingWithSequencedStart() throws Exception {
    int childPRCount = 2;

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);

        for (int i = 2; i < childPRCount + 2; ++i) {
          createChildPR_withPersistence(regionName, "region" + i, diskStoreName1,
              DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES,
              DEFAULT_STARTUP_RECOVERY_DELAY);
        }
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
      createData(childRegionName1, "c");
    });

    Map<Integer, Set<Integer>> bucketIdsInVM = new HashMap<>();
    for (VM vm : toArray(vm0, vm1)) {
      bucketIdsInVM.put(vm.getId(), vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        return bucketIds;
      }));
    }

    for (int i = 2; i < childPRCount + 2; ++i) {
      String childRegionName = "region" + i;
      for (VM vm : toArray(vm0, vm1)) {
        Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
        assertThat(vm.invoke(() -> getBucketIds(childRegionName))).isEqualTo(bucketIds);
      }
    }

    int expectedLogMessagesCount = 2;

    AsyncInvocation<Void> createMultipleChildPRGenerationsInVM0 =
        vm0.invokeAsync(
            () -> validateColocationLogger_withMultipleChildRegionGenerations(childPRCount,
                expectedLogMessagesCount));
    AsyncInvocation<Void> createMultipleChildPRGenerationsInVM1 =
        vm1.invokeAsync(
            () -> validateColocationLogger_withMultipleChildRegionGenerations(childPRCount,
                expectedLogMessagesCount));

    createMultipleChildPRGenerationsInVM0.await();
    createMultipleChildPRGenerationsInVM1.await();
  }

  /**
   * Testing that all missing persistent PRs in a colocation hierarchy are logged as warnings
   */
  @Test
  public void testHierarchyOfColocatedChildPRsMissing() throws Exception {
    int childPRGenerations = 2;

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);

        for (int i = 3; i < childPRGenerations + 2; ++i) {
          createChildPR_withPersistence("region" + (i - 1), "region" + i, diskStoreName1,
              DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        }
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
      createData(childRegionName2, "c");
    });

    Map<Integer, Set<Integer>> bucketIdsInVM = new HashMap<>();
    for (VM vm : toArray(vm0, vm1)) {
      bucketIdsInVM.put(vm.getId(), vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        return bucketIds;
      }));
    }

    for (int i = 2; i < childPRGenerations + 2; ++i) {
      String childRegionName = "region" + i;
      for (VM vm : toArray(vm0, vm1)) {
        Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
        assertThat(vm.invoke(() -> getBucketIds(childRegionName))).isEqualTo(bucketIds);
      }
    }

    // Expected warning logs only on the child region, because without the child there's nothing
    // known about the remaining hierarchy

    AsyncInvocation<Void> createPRWithMissingChildInVM0 =
        vm0.invokeAsync(() -> validateColocationLogger_withMissingChildRegion(childPRGenerations));
    AsyncInvocation<Void> createPRWithMissingChildInVM1 =
        vm1.invokeAsync(() -> validateColocationLogger_withMissingChildRegion(childPRGenerations));

    createPRWithMissingChildInVM0.await();
    createPRWithMissingChildInVM1.await();
  }

  /**
   * Testing that all missing persistent PRs in a colocation hierarchy are logged as warnings
   */
  @Test // GEODE-7513
  public void testHierarchyOfColocatedChildPRsMissingGrandchild() throws Exception {
    int childPRGenerationsCount = 3;

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence(regionName, childRegionName1, diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);

        for (int i = 3; i < childPRGenerationsCount + 2; ++i) {
          createChildPR_withPersistence("region" + (i - 1), "region" + i, diskStoreName1,
              DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        }
      });
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
      createData(childRegionName2, "c");
    });

    Map<Integer, Set<Integer>> bucketIdsInVM = new HashMap<>();
    for (VM vm : toArray(vm0, vm1)) {
      bucketIdsInVM.put(vm.getId(), vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        return bucketIds;
      }));
    }

    for (int i = 2; i < childPRGenerationsCount + 2; ++i) {
      String childRegionName = "region" + i;
      for (VM vm : toArray(vm0, vm1)) {
        Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
        assertThat(vm.invoke(() -> getBucketIds(childRegionName))).isEqualTo(bucketIds);
      }
    }

    // Expected warning logs only on the child region, because without the child
    // there's nothing known about the remaining hierarchy

    int expectedLogMessagesCount = childPRGenerationsCount * (childPRGenerationsCount + 1) / 2;

    AsyncInvocation<Void> createChildPRGenerationsInVM0 = vm0.invokeAsync(
        () -> validateColocationLogger_withChildRegionGenerations(childPRGenerationsCount,
            expectedLogMessagesCount));
    AsyncInvocation<Void> createChildPRGenerationsInVM1 = vm1.invokeAsync(
        () -> validateColocationLogger_withChildRegionGenerations(childPRGenerationsCount,
            expectedLogMessagesCount));

    createChildPRGenerationsInVM0.await();
    createChildPRGenerationsInVM1.await();
  }

  /**
   * Testing that all missing persistent PRs in a colocation tree hierarchy are logged as warnings.
   * This test is a combines the "multiple children" and "hierarchy of children" tests. This is the
   * colocation tree for this test
   *
   * <pre>
   *                  Parent
   *                /         \
   *             /               \
   *         Gen1_C1            Gen1_C2
   *         /    \              /    \
   *  Gen2_C1_1  Gen2_C1_2  Gen2_C2_1  Gen2_C2_2
   * </pre>
   */
  @Test
  public void testFullTreeOfColocatedChildPRsWithMissingRegions() throws Exception {
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence("Parent", diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence("Parent", "Gen1_C1", diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence("Parent", "Gen1_C2", diskStoreName1, DEFAULT_RECOVERY_DELAY,
            DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence("Gen1_C1", "Gen2_C1_1", diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence("Gen1_C1", "Gen2_C1_2", diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence("Gen1_C2", "Gen2_C2_1", diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
        createChildPR_withPersistence("Gen1_C2", "Gen2_C2_2", diskStoreName1,
            DEFAULT_RECOVERY_DELAY, DEFAULT_REDUNDANT_COPIES, DEFAULT_STARTUP_RECOVERY_DELAY);
      });
    }

    vm0.invoke(() -> {
      createData("Parent", "a");
      createData("Gen1_C1", "b");
      createData("Gen1_C2", "c");
      createData("Gen2_C1_1", "c");
      createData("Gen2_C1_2", "c");
      createData("Gen2_C2_1", "c");
      createData("Gen2_C2_2", "c");
    });

    Map<Integer, Set<Integer>> bucketIdsInVM = new HashMap<>();
    for (VM vm : toArray(vm0, vm1)) {
      bucketIdsInVM.put(vm.getId(), vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds("Parent");
        assertThat(bucketIds).isNotEmpty();
        return bucketIds;
      }));
    }

    for (String region : asList("Gen1_C1", "Gen1_C2", "Gen2_C1_1", "Gen2_C1_2", "Gen2_C2_1",
        "Gen2_C2_2")) {
      for (VM vm : toArray(vm0, vm1)) {
        Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
        assertThat(vm.invoke(() -> getBucketIds(region))).isEqualTo(bucketIds);
      }
    }

    AsyncInvocation<Void> createChildPRTreeInVM0 =
        vm0.invokeAsync(this::validateColocationLogger_withChildRegionTree);
    AsyncInvocation<Void> createChildPRTreeInVM1 =
        vm1.invokeAsync(this::validateColocationLogger_withChildRegionTree);

    createChildPRTreeInVM0.await();
    createChildPRTreeInVM1.await();
  }

  /**
   * Testing what happens we we recreate colocated persistent PRs by creating one PR everywhere and
   * then the other PR everywhere.
   */
  @Test
  public void testColocatedPRsRecoveryOnePRAtATime() throws Exception {
    for (VM vm : toArray(vm0, vm1, vm2)) {
      vm.invoke(() -> {
        createCache();
        createDiskStore(diskStoreName1);
        createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY, 1,
            DEFAULT_STARTUP_RECOVERY_DELAY);
      });
    }

    for (VM vm : toArray(vm0, vm1, vm2)) {
      vm.invoke(() -> createChildPR_withRecovery(regionName, childRegionName1,
          DEFAULT_RECOVERY_DELAY, 1, DEFAULT_STARTUP_RECOVERY_DELAY));
    }

    vm0.invoke(() -> {
      createData(regionName, "a");
      createData(childRegionName1, "b");
    });

    Map<Integer, Set<Integer>> bucketIdsInVM = new HashMap<>();
    for (VM vm : toArray(vm0, vm1, vm2)) {
      bucketIdsInVM.put(vm.getId(), vm.invoke(() -> {
        Set<Integer> bucketIds = getBucketIds(regionName);
        assertThat(bucketIds).isNotEmpty();
        return bucketIds;
      }));
    }

    for (VM vm : toArray(vm0, vm1, vm2)) {
      vm.invoke(() -> {
        Set<Integer> primaryBucketIds = getPrimaryBucketIds(regionName);
        assertThat(getPrimaryBucketIds(childRegionName1)).isEqualTo(primaryBucketIds);
      });
    }

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

    for (VM vm : toArray(vm0, vm1, vm2)) {
      addAsync(
          vm.invokeAsync(() -> {
            createCache();
            createDiskStore(diskStoreName1);
            createPR_withPersistence(regionName, diskStoreName1, DEFAULT_RECOVERY_DELAY, 1,
                DEFAULT_STARTUP_RECOVERY_DELAY);
          }));
    }
    awaitAllAsync();

    for (VM vm : toArray(vm0, vm1, vm2)) {
      vm.invoke(() -> createChildPR_withRecovery(regionName, childRegionName1,
          DEFAULT_RECOVERY_DELAY, 1, DEFAULT_STARTUP_RECOVERY_DELAY));
    }

    for (VM vm : toArray(vm0, vm1, vm2)) {
      Set<Integer> bucketIds = bucketIdsInVM.get(vm.getId());
      vm.invoke(() -> {
        assertThat(getBucketIds(regionName)).isEqualTo(bucketIds);
        assertThat(getBucketIds(childRegionName1)).isEqualTo(bucketIds);
      });
    }

    // primary can differ
    for (VM vm : toArray(vm0, vm1, vm2)) {
      vm.invoke(() -> {
        Set<Integer> primaryBucketIds = getPrimaryBucketIds(regionName);
        assertThat(getPrimaryBucketIds(childRegionName1)).isEqualTo(primaryBucketIds);
      });
    }

    vm0.invoke(() -> {
