geode-core/src/distributedTest/java/org/apache/geode/cache/query/internal/index/InitializeIndexEntryDestroyQueryDUnitTest.java - geode - Git at Google

 /*
  * 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.cache.query.internal.index;

 import static org.apache.geode.cache.query.Utils.createPortfolioData;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;

 import java.util.Arrays;
 import java.util.Properties;

 import junitparams.JUnitParamsRunner;
 import junitparams.Parameters;
 import org.apache.logging.log4j.Logger;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.junit.runner.RunWith;

 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.Scope;
 import org.apache.geode.cache.query.Index;
 import org.apache.geode.cache.query.Query;
 import org.apache.geode.cache.query.SelectResults;
 import org.apache.geode.cache.query.data.Portfolio;
 import org.apache.geode.cache.query.data.PortfolioData;
 import org.apache.geode.cache.query.internal.Undefined;
 import org.apache.geode.cache.query.partitioned.PRQueryDUnitHelper;
 import org.apache.geode.distributed.ConfigurationProperties;
 import org.apache.geode.logging.internal.log4j.api.LogService;
 import org.apache.geode.test.dunit.AsyncInvocation;
 import org.apache.geode.test.dunit.Host;
 import org.apache.geode.test.dunit.ThreadUtils;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
 import org.apache.geode.test.junit.categories.OQLIndexTest;

 /**
  * Test creates a local region. Creates and removes index in a parallel running thread. Then
  * destroys and puts back entries in separated thread in the same region and runs query in parallel
  * and checks for UNDEFINED values in result set of the query.
  */
 @Category({OQLIndexTest.class})
 @RunWith(JUnitParamsRunner.class)
 public class InitializeIndexEntryDestroyQueryDUnitTest extends JUnit4CacheTestCase {
   private static final Logger logger = LogService.getLogger();

   PRQueryDUnitHelper PRQHelp = new PRQueryDUnitHelper();

   final Portfolio portfolio = new Portfolio(1, 1);

   private int cnt = 0;

   private int cntDest = 100;

   public InitializeIndexEntryDestroyQueryDUnitTest() {
     super();
   }

   public void setCacheInVMs(VM... vms) {
     for (VM vm : vms) {
       vm.invoke(() -> PRQueryDUnitHelper.setCache(getCache()));
     }
   }

   private static Scope[] getScope() {
     return new Scope[] {Scope.LOCAL, null};
   }

   @Override
   public Properties getDistributedSystemProperties() {
     Properties properties = super.getDistributedSystemProperties();
     properties.put(ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER,
         "org.apache.geode.cache.query.data.**");
     return properties;
   }


   @Test
   @Parameters(method = "getScope")
   public void testAsyncIndexInitDuringEntryDestroyAndQueryOnPR(Scope scope) {
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     setCacheInVMs(vm0);
     String regionName = "PartionedPortfoliosPR";
     String query = "select * from /" + regionName + " p where p.status = 'active'";
     try {
       vm0.invoke(() -> createRegionInVM(regionName, scope));
       final PortfolioData[] portfolio = createPortfolioData(cnt, cntDest);
       vm0.invoke(
           PRQHelp.getCacheSerializableRunnableForPRPuts(regionName, portfolio, cnt, cntDest));
       AsyncInvocation asyInvk0 =
           vm0.invokeAsync(() -> consecutivelyCreateAndDestroyIndex(regionName));
       AsyncInvocation asyInvk1 =
           vm0.invokeAsync(() -> consecutivelyPutAndDestroyEntries(regionName));
       vm0.invoke(() -> executeAndValidateQueryResults(query));
       waitForAsyncThreadsToComplete(asyInvk0, asyInvk1);
     } finally {
       vm0.invoke(() -> clearIndexesAndDestroyRegion(regionName));
     }
   }

   @Test
   public void testConcurrentRemoveIndexAndQueryOnPR() {
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     setCacheInVMs(vm0);
     String name = "PartionedPortfoliosPR";
     String query =
         "select * from /" + name + " p where p.status = 'active' and p.ID > 0 and p.pk != ' ' ";
     try {
       vm0.invoke(() -> createRegionInVM(name, null));
       final PortfolioData[] portfolio = createPortfolioData(cnt, cntDest);
       vm0.invoke(PRQHelp.getCacheSerializableRunnableForPRPuts(name, portfolio, cnt, cntDest));
       vm0.invoke(() -> createIndex(name, "statusIndex", "p.status", "/" + name + " p"));
       vm0.invoke(() -> createIndex(name, "idIndex", "p.ID", "/" + name + " p"));
       vm0.invoke(() -> createIndex(name, "pkidIndex", "p.pk", "/" + name + " p"));
       vm0.invoke(() -> executeAndValidateQueryResults(query));
     } finally {
       vm0.invoke(() -> clearIndexesAndDestroyRegion(name));
     }

   }

   private void waitForAsyncThreadsToComplete(AsyncInvocation asyInvk0, AsyncInvocation asyInvk1) {
     ThreadUtils.join(asyInvk0, 1000 * 1000); // TODO: this is way too long: 16.67 minutes!
     if (asyInvk0.exceptionOccurred()) {
       logger.error("Asynchronous thread to create and destroy index failed",
           asyInvk0.getException());
       fail();
     }

     ThreadUtils.join(asyInvk1, 1000 * 1000); // TODO: this is way too long: 16.67 minutes!
     if (asyInvk1.exceptionOccurred()) {
       logger.error("Asychronous thread to create and destroy region entry failed with exception",
           asyInvk1.getException());
       fail();
     }
   }

   private void createRegionInVM(String regionName, Scope scope) {
     Cache cache = getCache();
     Region partitionRegion = null;
     try {
       RegionFactory regionFactory = cache.createRegionFactory();
       if (scope != null) {
         partitionRegion = regionFactory.setValueConstraint(PortfolioData.class)
             .setIndexMaintenanceSynchronous(false).setScope(scope).create(regionName);
       } else {
         partitionRegion = regionFactory.setValueConstraint(PortfolioData.class)
             .setIndexMaintenanceSynchronous(false).setDataPolicy(DataPolicy.PARTITION)
             .create(regionName);
       }
     } catch (IllegalStateException ex) {
       logger.warn("Creation caught IllegalStateException", ex);
     }
     assertNotNull("Region " + regionName + " not in cache", cache.getRegion(regionName));
     assertNotNull("Region ref null", partitionRegion);
     assertTrue("Region ref claims to be destroyed", !partitionRegion.isDestroyed());
   }

   private void consecutivelyCreateAndDestroyIndex(String regionName) {
     for (int i = 0; i < cntDest; i++) {
       // Create Index first to go in hook.
       Cache cache = getCache();
       Index index = null;
       try {
         index =
             cache.getQueryService().createIndex("statusIndex", "p.status", "/" + regionName + " p");
       } catch (Exception e1) {
         logger.error("Index creation failed", e1);
         fail();
       }
       Region region = cache.getRegion(regionName);
       await()
           .untilAsserted(
               () -> assertNotNull(cache.getQueryService().getIndex(region, "statusIndex")));
       getCache().getQueryService().removeIndex(index);
       await().untilAsserted(
           () -> assertEquals(null, getCache().getQueryService().getIndex(region, "statusIndex")));
     }
   }

   private void consecutivelyPutAndDestroyEntries(String regionName) {
     Region r = getCache().getRegion(regionName);

     for (int i = 0, j = 0; i < 500; i++, j++) {

       PortfolioData p = (PortfolioData) r.get(j);
       logger.debug("Going to destroy the value" + p);
       r.destroy(j);
       final int key = j;
       await()
           .untilAsserted(() -> assertEquals(null, r.get(key)));

       // Put the value back again.
       getCache().getLogger().fine("Putting the value back" + p);
       r.put(j, p);

       // Reset j
       if (j == cntDest - 1) {
         j = 0;
       }
     }
   }

   private void executeAndValidateQueryResults(String queryString) {
     Query query = getCache().getQueryService().newQuery(queryString);
     // Now run the query
     SelectResults results = null;
     for (int i = 0; i < 50; i++) {
       try {
         logger.debug("Querying the region");
         results = (SelectResults) query.execute();
       } catch (Exception e) {
         e.printStackTrace();
         logger.error("Query Execution failed", e);
         fail();
       }
       for (Object obj : results) {
         if (obj instanceof Undefined) {
           fail("Found an undefined element" + Arrays.toString(results.toArray()));
         }
       }
     }
   }

   private void clearIndexesAndDestroyRegion(String regionName) {
     Region region = getCache().getRegion(regionName);
     if (region != null) {
       getCache().getQueryService().removeIndexes(region);
       region.destroyRegion();
     }
   }

   private void createIndex(String regionName, String indexName, String expression,
       String regionPath) {
     try {
       getCache().getQueryService().createIndex(indexName, expression, regionPath);
     } catch (Exception e) {
       e.printStackTrace();
       logger.error("Exception while creating index : " + indexName, e);
       fail();
     }
     Region region = getCache().getRegion(regionName);
     await()
         .untilAsserted(
             () -> assertNotNull(getCache().getQueryService().getIndex(region, indexName)));
   }
 }
	/*
	* 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.cache.query.internal.index;

	import static org.apache.geode.cache.query.Utils.createPortfolioData;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertNotNull;
	import static org.junit.Assert.assertTrue;
	import static org.junit.Assert.fail;

	import java.util.Arrays;
	import java.util.Properties;

	import junitparams.JUnitParamsRunner;
	import junitparams.Parameters;
	import org.apache.logging.log4j.Logger;
	import org.junit.Test;
	import org.junit.experimental.categories.Category;
	import org.junit.runner.RunWith;

	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.DataPolicy;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.Scope;
	import org.apache.geode.cache.query.Index;
	import org.apache.geode.cache.query.Query;
	import org.apache.geode.cache.query.SelectResults;
	import org.apache.geode.cache.query.data.Portfolio;
	import org.apache.geode.cache.query.data.PortfolioData;
	import org.apache.geode.cache.query.internal.Undefined;
	import org.apache.geode.cache.query.partitioned.PRQueryDUnitHelper;
	import org.apache.geode.distributed.ConfigurationProperties;
	import org.apache.geode.logging.internal.log4j.api.LogService;
	import org.apache.geode.test.dunit.AsyncInvocation;
	import org.apache.geode.test.dunit.Host;
	import org.apache.geode.test.dunit.ThreadUtils;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
	import org.apache.geode.test.junit.categories.OQLIndexTest;

	/**
	* Test creates a local region. Creates and removes index in a parallel running thread. Then
	* destroys and puts back entries in separated thread in the same region and runs query in parallel
	* and checks for UNDEFINED values in result set of the query.
	*/
	@Category({OQLIndexTest.class})
	@RunWith(JUnitParamsRunner.class)
	public class InitializeIndexEntryDestroyQueryDUnitTest extends JUnit4CacheTestCase {
	private static final Logger logger = LogService.getLogger();

	PRQueryDUnitHelper PRQHelp = new PRQueryDUnitHelper();

	final Portfolio portfolio = new Portfolio(1, 1);

	private int cnt = 0;

	private int cntDest = 100;

	public InitializeIndexEntryDestroyQueryDUnitTest() {
	super();
	}

	public void setCacheInVMs(VM... vms) {
	for (VM vm : vms) {
	vm.invoke(() -> PRQueryDUnitHelper.setCache(getCache()));
	}
	}

	private static Scope[] getScope() {
	return new Scope[] {Scope.LOCAL, null};
	}

	@Override
	public Properties getDistributedSystemProperties() {
	Properties properties = super.getDistributedSystemProperties();
	properties.put(ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER,
	"org.apache.geode.cache.query.data.**");
	return properties;
	}


	@Test
	@Parameters(method = "getScope")
	public void testAsyncIndexInitDuringEntryDestroyAndQueryOnPR(Scope scope) {
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	setCacheInVMs(vm0);
	String regionName = "PartionedPortfoliosPR";
	String query = "select * from /" + regionName + " p where p.status = 'active'";
	try {
	vm0.invoke(() -> createRegionInVM(regionName, scope));
	final PortfolioData[] portfolio = createPortfolioData(cnt, cntDest);
	vm0.invoke(
	PRQHelp.getCacheSerializableRunnableForPRPuts(regionName, portfolio, cnt, cntDest));
	AsyncInvocation asyInvk0 =
	vm0.invokeAsync(() -> consecutivelyCreateAndDestroyIndex(regionName));
	AsyncInvocation asyInvk1 =
	vm0.invokeAsync(() -> consecutivelyPutAndDestroyEntries(regionName));
	vm0.invoke(() -> executeAndValidateQueryResults(query));
	waitForAsyncThreadsToComplete(asyInvk0, asyInvk1);
	} finally {
	vm0.invoke(() -> clearIndexesAndDestroyRegion(regionName));
	}
	}

	@Test
	public void testConcurrentRemoveIndexAndQueryOnPR() {
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	setCacheInVMs(vm0);
	String name = "PartionedPortfoliosPR";
	String query =
	"select * from /" + name + " p where p.status = 'active' and p.ID > 0 and p.pk != ' ' ";
	try {
	vm0.invoke(() -> createRegionInVM(name, null));
	final PortfolioData[] portfolio = createPortfolioData(cnt, cntDest);
	vm0.invoke(PRQHelp.getCacheSerializableRunnableForPRPuts(name, portfolio, cnt, cntDest));
	vm0.invoke(() -> createIndex(name, "statusIndex", "p.status", "/" + name + " p"));
	vm0.invoke(() -> createIndex(name, "idIndex", "p.ID", "/" + name + " p"));
	vm0.invoke(() -> createIndex(name, "pkidIndex", "p.pk", "/" + name + " p"));
	vm0.invoke(() -> executeAndValidateQueryResults(query));
	} finally {
	vm0.invoke(() -> clearIndexesAndDestroyRegion(name));
	}

	}

	private void waitForAsyncThreadsToComplete(AsyncInvocation asyInvk0, AsyncInvocation asyInvk1) {
	ThreadUtils.join(asyInvk0, 1000 * 1000); // TODO: this is way too long: 16.67 minutes!
	if (asyInvk0.exceptionOccurred()) {
	logger.error("Asynchronous thread to create and destroy index failed",
	asyInvk0.getException());
	fail();
	}

	ThreadUtils.join(asyInvk1, 1000 * 1000); // TODO: this is way too long: 16.67 minutes!
	if (asyInvk1.exceptionOccurred()) {
	logger.error("Asychronous thread to create and destroy region entry failed with exception",
	asyInvk1.getException());
	fail();
	}
	}

	private void createRegionInVM(String regionName, Scope scope) {
	Cache cache = getCache();
	Region partitionRegion = null;
	try {
	RegionFactory regionFactory = cache.createRegionFactory();
	if (scope != null) {
	partitionRegion = regionFactory.setValueConstraint(PortfolioData.class)
	.setIndexMaintenanceSynchronous(false).setScope(scope).create(regionName);
	} else {
	partitionRegion = regionFactory.setValueConstraint(PortfolioData.class)
	.setIndexMaintenanceSynchronous(false).setDataPolicy(DataPolicy.PARTITION)
	.create(regionName);
	}
	} catch (IllegalStateException ex) {
	logger.warn("Creation caught IllegalStateException", ex);
	}
	assertNotNull("Region " + regionName + " not in cache", cache.getRegion(regionName));
	assertNotNull("Region ref null", partitionRegion);
	assertTrue("Region ref claims to be destroyed", !partitionRegion.isDestroyed());
	}

	private void consecutivelyCreateAndDestroyIndex(String regionName) {
	for (int i = 0; i < cntDest; i++) {
	// Create Index first to go in hook.
	Cache cache = getCache();
	Index index = null;
	try {
	index =
	cache.getQueryService().createIndex("statusIndex", "p.status", "/" + regionName + " p");
	} catch (Exception e1) {
	logger.error("Index creation failed", e1);
	fail();
	}
	Region region = cache.getRegion(regionName);
	await()
	.untilAsserted(
	() -> assertNotNull(cache.getQueryService().getIndex(region, "statusIndex")));
	getCache().getQueryService().removeIndex(index);
	await().untilAsserted(
	() -> assertEquals(null, getCache().getQueryService().getIndex(region, "statusIndex")));
	}
	}

	private void consecutivelyPutAndDestroyEntries(String regionName) {
	Region r = getCache().getRegion(regionName);

	for (int i = 0, j = 0; i < 500; i++, j++) {

	PortfolioData p = (PortfolioData) r.get(j);
	logger.debug("Going to destroy the value" + p);
	r.destroy(j);
	final int key = j;
	await()
	.untilAsserted(() -> assertEquals(null, r.get(key)));

	// Put the value back again.
	getCache().getLogger().fine("Putting the value back" + p);
	r.put(j, p);

	// Reset j
	if (j == cntDest - 1) {
	j = 0;
	}
	}
	}

	private void executeAndValidateQueryResults(String queryString) {
	Query query = getCache().getQueryService().newQuery(queryString);
	// Now run the query
	SelectResults results = null;
	for (int i = 0; i < 50; i++) {
	try {
	logger.debug("Querying the region");
	results = (SelectResults) query.execute();
	} catch (Exception e) {
	e.printStackTrace();
	logger.error("Query Execution failed", e);
	fail();
	}
	for (Object obj : results) {
	if (obj instanceof Undefined) {
	fail("Found an undefined element" + Arrays.toString(results.toArray()));
	}
	}
	}
	}

	private void clearIndexesAndDestroyRegion(String regionName) {
	Region region = getCache().getRegion(regionName);
	if (region != null) {
	getCache().getQueryService().removeIndexes(region);
	region.destroyRegion();
	}
	}

	private void createIndex(String regionName, String indexName, String expression,
	String regionPath) {
	try {
	getCache().getQueryService().createIndex(indexName, expression, regionPath);
	} catch (Exception e) {
	e.printStackTrace();
	logger.error("Exception while creating index : " + indexName, e);
	fail();
	}
	Region region = getCache().getRegion(regionName);
	await()
	.untilAsserted(
	() -> assertNotNull(getCache().getQueryService().getIndex(region, indexName)));
	}
	}