@Test
  public void testAsyncIndexInitDuringEntryDestroyAndQueryOnPersistentPR() {
    Host host = Host.getHost(0);
    VM vm0 = host.getVM(0);
    hooked = false;
    name = "PartionedPortfoliosPR";
    // Create Overflow Persistent Partition Region
    vm0.invoke(
        new CacheSerializableRunnable("Create local region with synchronous index maintenance") {
          @Override
          public void run2() throws CacheException {
            Cache cache = getCache();
            Region partitionRegion = null;
            IndexManager.testHook = null;
            try {
              DiskStore ds = cache.findDiskStore("disk");
              if (ds == null) {
                ds = cache.createDiskStoreFactory().setDiskDirs(getDiskDirs()).create("disk");
              }
              AttributesFactory attr = new AttributesFactory();
              attr.setValueConstraint(PortfolioData.class);
              attr.setIndexMaintenanceSynchronous(true);
              EvictionAttributesImpl evicAttr =
                  new EvictionAttributesImpl().setAction(EvictionAction.OVERFLOW_TO_DISK);
              evicAttr.setAlgorithm(EvictionAlgorithm.LRU_ENTRY).setMaximum(1);
              attr.setEvictionAttributes(evicAttr);
              attr.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
              attr.setPartitionAttributes(
                  new PartitionAttributesFactory().setTotalNumBuckets(1).create());
              attr.setDiskStoreName("disk");
              RegionFactory regionFactory = cache.createRegionFactory(attr.create());
              partitionRegion = regionFactory.create(name);
            } catch (IllegalStateException ex) {
              LogWriterUtils.getLogWriter().warning("Creation caught IllegalStateException", ex);
            }
            assertNotNull("Region " + name + " not in cache", cache.getRegion(name));
            assertNotNull("Region ref null", partitionRegion);
            assertTrue("Region ref claims to be destroyed", !partitionRegion.isDestroyed());
            // Create Indexes
            try {
              Index index =
                  cache.getQueryService().createIndex("statusIndex", "p.ID",
                      SEPARATOR + name + " p");
              assertNotNull(index);
            } catch (Exception e1) {
              e1.printStackTrace();
              fail("Index creation failed");
            }
          }
        });

    // Start changing the value in Region which should turn into a deadlock if
    // the fix is not there
    AsyncInvocation<Void> asyncInv1 =
        vm0.invokeAsync(new CacheSerializableRunnable("Change value in region") {

          @Override
          public void run2() throws CacheException {
            // Do a put in region.
            Region r = getCache().getRegion(name);

            for (int i = 0; i < 100; i++) {
              r.put(i, new PortfolioData(i));
            }

            assertNull(IndexManager.testHook);
            IndexManager.testHook = new IndexManagerTestHook();

            // Destroy one of the values.
            getCache().getLogger().fine("Destroying the value");
            r.destroy(1);

            IndexManager.testHook = null;
          }
        });

    AsyncInvocation<Void> asyncInv2 =
        vm0.invokeAsync(new CacheSerializableRunnable("Run query on region") {

          @Override
          public void run2() throws CacheException {
            Query statusQuery = getCache().getQueryService()
                .newQuery("select * from " + SEPARATOR + name + " p where p.ID > -1");

            while (!hooked) {
              Wait.pause(100);
            }
            try {
              getCache().getLogger().fine("Querying the region");
              SelectResults results = (SelectResults) statusQuery.execute();
              assertEquals(100, results.size());
            } catch (Exception e) {
              e.printStackTrace();
            }
          }
        });

    // If we take more than 30 seconds then its a deadlock.
    ThreadUtils.join(asyncInv2, 30 * 1000);
    ThreadUtils.join(asyncInv1, 30 * 1000);
  }
Test creates a persistent-overflow region and performs updates in region which has compact
indexes and queries the indexes simultaneously.

This test's main purpose is to look for deadlock problems caused by NOT taking early locks on
compact indexes and remove the requirement of the system property
gemfire.index.acquireCompactIndexLocksWithRegionEntryLocks.
geode-core/src/distributedTest/java/org/apache/geode/cache/query/internal/index/ConcurrentIndexOperationsOnOverflowRegionDUnitTest.java