| /* |
| * 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.dunit; |
| |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.test.dunit.Assert.assertEquals; |
| import static org.apache.geode.test.dunit.Assert.assertFalse; |
| import static org.apache.geode.test.dunit.Assert.assertNotNull; |
| import static org.apache.geode.test.dunit.Assert.assertTrue; |
| import static org.apache.geode.test.dunit.Assert.fail; |
| |
| import java.util.Iterator; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.apache.logging.log4j.Logger; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.PartitionAttributesFactory; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.client.ClientCache; |
| import org.apache.geode.cache.client.ClientCacheFactory; |
| import org.apache.geode.cache.client.PoolFactory; |
| import org.apache.geode.cache.client.PoolManager; |
| import org.apache.geode.cache.client.ServerOperationException; |
| import org.apache.geode.cache.query.FunctionDomainException; |
| import org.apache.geode.cache.query.Index; |
| import org.apache.geode.cache.query.IndexInvalidException; |
| import org.apache.geode.cache.query.NameResolutionException; |
| import org.apache.geode.cache.query.Query; |
| import org.apache.geode.cache.query.QueryException; |
| import org.apache.geode.cache.query.QueryExecutionLowMemoryException; |
| import org.apache.geode.cache.query.QueryExecutionTimeoutException; |
| import org.apache.geode.cache.query.QueryInvocationTargetException; |
| import org.apache.geode.cache.query.QueryService; |
| import org.apache.geode.cache.query.SelectResults; |
| import org.apache.geode.cache.query.TypeMismatchException; |
| import org.apache.geode.cache.query.data.Portfolio; |
| import org.apache.geode.cache.query.internal.DefaultQuery; |
| import org.apache.geode.cache.query.internal.ExecutionContext; |
| import org.apache.geode.cache.server.CacheServer; |
| import org.apache.geode.cache30.CacheSerializableRunnable; |
| import org.apache.geode.cache30.ClientServerTestCase; |
| import org.apache.geode.distributed.internal.DistributionConfig; |
| import org.apache.geode.internal.AvailablePortHelper; |
| import org.apache.geode.internal.cache.DistributedRegion; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.control.HeapMemoryMonitor; |
| import org.apache.geode.internal.cache.control.InternalResourceManager; |
| import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType; |
| import org.apache.geode.internal.cache.control.MemoryEvent; |
| import org.apache.geode.internal.cache.control.ResourceListener; |
| import org.apache.geode.internal.cache.control.TestMemoryThresholdListener; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| import org.apache.geode.test.dunit.AsyncInvocation; |
| import org.apache.geode.test.dunit.DistributedTestUtils; |
| import org.apache.geode.test.dunit.IgnoredException; |
| import org.apache.geode.test.dunit.Invoke; |
| import org.apache.geode.test.dunit.NetworkUtils; |
| import org.apache.geode.test.dunit.SerializableCallable; |
| import org.apache.geode.test.dunit.ThreadUtils; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.junit.categories.OQLQueryTest; |
| |
| @Category({OQLQueryTest.class}) |
| public class ResourceManagerWithQueryMonitorDUnitTest extends ClientServerTestCase { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| private static int MAX_TEST_QUERY_TIMEOUT = 4000; |
| private static int TEST_QUERY_TIMEOUT = 1000; |
| private static final int CRITICAL_HEAP_USED = 950; |
| private static final int NORMAL_HEAP_USED = 500; |
| |
| @Override |
| public final void postSetUpClientServerTestCase() throws Exception { |
| Invoke.invokeInEveryVM(this.setHeapMemoryMonitorTestMode); |
| IgnoredException.addIgnoredException("above heap critical threshold"); |
| IgnoredException.addIgnoredException("below heap critical threshold"); |
| } |
| |
| @Override |
| protected void preTearDownClientServerTestCase() throws Exception { |
| Invoke.invokeInEveryVM(resetQueryMonitor); |
| Invoke.invokeInEveryVM(resetResourceManager); |
| } |
| |
| private SerializableCallable setHeapMemoryMonitorTestMode = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| HeapMemoryMonitor.setTestDisableMemoryUpdates(true); |
| return null; |
| } |
| }; |
| |
| private SerializableCallable resetResourceManager = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| InternalResourceManager irm = ((GemFireCacheImpl) getCache()).getInternalResourceManager(); |
| // Reset CRITICAL_UP by informing all that heap usage is now 1 byte (0 would disable). |
| irm.getHeapMonitor().updateStateAndSendEvent(NORMAL_HEAP_USED, "test"); |
| Set<ResourceListener> listeners = irm.getResourceListeners(ResourceType.HEAP_MEMORY); |
| Iterator<ResourceListener> it = listeners.iterator(); |
| while (it.hasNext()) { |
| ResourceListener<MemoryEvent> l = it.next(); |
| if (l instanceof TestMemoryThresholdListener) { |
| ((TestMemoryThresholdListener) l).resetThresholdCalls(); |
| } |
| } |
| irm.setCriticalHeapPercentage(0f); |
| irm.setEvictionHeapPercentage(0f); |
| irm.getHeapMonitor().setTestMaxMemoryBytes(0); |
| HeapMemoryMonitor.setTestDisableMemoryUpdates(false); |
| return null; |
| } |
| }; |
| |
| private SerializableCallable resetQueryMonitor = new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| InternalCache cache = getCache(); |
| if (cache.getQueryMonitor() != null) { |
| cache.getQueryMonitor().setLowMemory(false, 0); |
| } |
| DefaultQuery.testHook = null; |
| return null; |
| } |
| }; |
| |
| @Test |
| public void testRMAndNoTimeoutSet() throws Exception { |
| doCriticalMemoryHitTest("portfolios", false, 85/* crit threshold */, false, -1, true); |
| } |
| |
| @Test |
| public void testRMAndNoTimeoutSetParReg() throws Exception { |
| doCriticalMemoryHitTest("portfolios", true, 85/* crit threshold */, false, -1, true); |
| } |
| |
| @Test |
| public void testRMButDisabledQueryMonitorForLowMemAndNoTimeoutSet() throws Exception { |
| // verify that timeout is not set and that a query can execute properly |
| doCriticalMemoryHitTest("portfolios", false, 85/* crit threshold */, true, -1, true); |
| } |
| |
| @Test |
| public void testRMAndTimeoutSet() throws Exception { |
| // verify that we still receive critical heap cancelation |
| doCriticalMemoryHitTest("portfolios", false, 85/* crit threshold */, true, TEST_QUERY_TIMEOUT, |
| true); |
| } |
| |
| @Test |
| public void testRMAndTimeoutSetAndQueryTimesoutInstead() throws Exception { |
| // verify that timeout is set correctly and cancel query |
| doCriticalMemoryHitTest("portfolios", false, 85/* crit threshold */, true, TEST_QUERY_TIMEOUT, |
| false); |
| } |
| |
| @Test |
| public void testRMButDisabledQueryMonitorForLowMemAndTimeoutSet() throws Exception { |
| // verify that timeout is still working properly |
| doCriticalMemoryHitTest("portfolios", false, 85/* crit threshold */, true, TEST_QUERY_TIMEOUT, |
| true); |
| } |
| |
| // Query directly on member with RM and QM set |
| @Test |
| public void testRMAndNoTimeoutSetOnServer() throws Exception { |
| doCriticalMemoryHitTestOnServer("portfolios", false, 85/* crit threshold */, false, -1, true); |
| } |
| |
| // Query directly on member with RM and QM set |
| @Test |
| public void whenTimeoutIsSetAndAQueryIsExecutedThenTimeoutMustStopTheQueryBeforeCriticalMemory() |
| throws Exception { |
| // Timeout is set along with critical heap but it be called after the timeout expires |
| // Timeout is set to 1ms which is very unrealistic time period for a query to be able to fetch |
| // 200 entries from the region successfully, hence a timeout is expected. |
| executeQueryWithTimeoutSetAndCriticalThreshold("portfolios", false, 85/* crit threshold */, |
| false, 1, true); |
| } |
| |
| @Test |
| public void whenTimeoutIsSetAndAQueryIsExecutedFromClientThenTimeoutMustStopTheQueryBeforeCriticalMemory() |
| throws Exception { |
| // Timeout is set along with critical heap but it be called after the timeout expires |
| // Timeout is set to 1ms which is very unrealistic time period for a query to be able to fetch |
| // 200 entries from the region successfully, hence a timeout is expected. |
| executeQueryFromClientWithTimeoutSetAndCriticalThreshold("portfolios", false, |
| 85/* crit threshold */, false, 1, true); |
| } |
| |
| @Test |
| public void testRMAndNoTimeoutSetParRegOnServer() throws Exception { |
| doCriticalMemoryHitTestOnServer("portfolios", true, 85/* crit threshold */, false, -1, true); |
| } |
| |
| @Test |
| public void testRMButDisabledQueryMonitorForLowMemAndNoTimeoutSetOnServer() throws Exception { |
| // verify that timeout is not set and that a query can execute properly |
| doCriticalMemoryHitTestOnServer("portfolios", false, 85/* crit threshold */, true, -1, true); |
| } |
| |
| @Test |
| public void testRMAndTimeoutSetOnServer() throws Exception { |
| // verify that we still receive critical heap cancelation |
| doCriticalMemoryHitTestOnServer("portfolios", false, 85/* crit threshold */, true, |
| TEST_QUERY_TIMEOUT, true); |
| } |
| |
| @Test |
| public void testRMAndTimeoutSetAndQueryTimesoutInsteadOnServer() throws Exception { |
| // verify that timeout is set correctly and cancel query |
| doCriticalMemoryHitTestOnServer("portfolios", false, 85/* crit threshold */, true, |
| TEST_QUERY_TIMEOUT, false); |
| } |
| |
| @Test |
| public void testRMButDisabledQueryMonitorForLowMemAndTimeoutSetOnServer() throws Exception { |
| // verify that timeout is still working properly |
| doCriticalMemoryHitTestOnServer("portfolios", false, 85/* crit threshold */, true, |
| TEST_QUERY_TIMEOUT, true); |
| } |
| |
| @Test |
| public void testPRGatherCancellation() throws Exception { |
| doCriticalMemoryHitTestWithMultipleServers("portfolios", true, 85/* crit threshold */, false, |
| -1, true); |
| } |
| |
| @Test |
| public void testPRGatherCancellationWhileGatheringResults() throws Exception { |
| doCriticalMemoryHitDuringGatherTestWithMultipleServers("portfolios", true, |
| 85/* crit threshold */, false, -1, true); |
| } |
| |
| @Test |
| public void testPRGatherCancellationWhileAddingResults() throws Exception { |
| doCriticalMemoryHitAddResultsTestWithMultipleServers("portfolios", true, 85/* crit threshold */, |
| false, -1, true); |
| } |
| |
| @Test |
| public void testIndexCreationCancellationPR() throws Exception { |
| doCriticalMemoryHitWithIndexTest("portfolios", true, 85/* crit threshold */, false, -1, true, |
| "compact"); |
| } |
| |
| @Test |
| public void testIndexCreationCancellation() throws Exception { |
| doCriticalMemoryHitWithIndexTest("portfolios", false, 85/* crit threshold */, false, -1, true, |
| "compact"); |
| } |
| |
| @Test |
| public void testIndexCreationNoCancellationPR() throws Exception { |
| doCriticalMemoryHitWithIndexTest("portfolios", true, 85/* crit threshold */, true, -1, true, |
| "compact"); |
| } |
| |
| @Test |
| public void testHashIndexCreationCancellationPR() throws Exception { |
| doCriticalMemoryHitWithIndexTest("portfolios", true, 85/* crit threshold */, false, -1, true, |
| "hash"); |
| } |
| |
| @Test |
| public void testHashIndexCreationCancellation() throws Exception { |
| // need to add hook to canceled result set and very it is triggered for multiple servers |
| doCriticalMemoryHitWithIndexTest("portfolios", false, 85/* crit threshold */, false, -1, true, |
| "hash"); |
| } |
| |
| @Test |
| public void testHashIndexCreationNoCancellationPR() throws Exception { |
| // need to add hook to canceled result set and very it is triggered for multiple servers |
| doCriticalMemoryHitWithIndexTest("portfolios", true, 85/* crit threshold */, true, -1, true, |
| "hash"); |
| } |
| |
| private void doCriticalMemoryHitTest(final String regionName, boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, final boolean hitCriticalThreshold) |
| throws Exception { |
| // create region on the server |
| final VM server = VM.getVM(0); |
| final VM client = VM.getVM(1); |
| final int numObjects = 200; |
| try { |
| final int port = AvailablePortHelper.getRandomAvailableTCPPort(); |
| startCacheServer(server, port, criticalThreshold, disabledQueryMonitorForLowMem, queryTimeout, |
| regionName, createPR, 0); |
| |
| startClient(client, server, port, regionName); |
| populateData(server, regionName, numObjects); |
| |
| doTestCriticalHeapAndQueryTimeout(server, client, regionName, disabledQueryMonitorForLowMem, |
| queryTimeout, hitCriticalThreshold); |
| |
| // Pause for a second and then let's recover |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server); |
| } |
| |
| // Check to see if query execution is ok under "normal" or "healthy" conditions |
| client.invoke(new CacheSerializableRunnable("Executing query when system is 'Normal'") { |
| @Override |
| public void run2() { |
| try { |
| QueryService qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| SelectResults results = (SelectResults) query.execute(); |
| assertEquals(numObjects, results.size()); |
| } catch (QueryInvocationTargetException e) { |
| assertFalse(true); |
| } catch (NameResolutionException e) { |
| assertFalse(true); |
| } catch (TypeMismatchException e) { |
| assertFalse(true); |
| } catch (FunctionDomainException e) { |
| assertFalse(true); |
| } |
| } |
| }); |
| |
| // Execute a critical heap event/ query timeout test again |
| doTestCriticalHeapAndQueryTimeout(server, client, regionName, disabledQueryMonitorForLowMem, |
| queryTimeout, hitCriticalThreshold); |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server); |
| } |
| } finally { |
| stopServer(server); |
| } |
| } |
| |
| // test to verify what happens during index creation if memory threshold is hit |
| private void doCriticalMemoryHitWithIndexTest(final String regionName, boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold, |
| final String indexType) |
| throws Exception { |
| // create region on the server |
| final VM server1 = VM.getVM(0); |
| final VM server2 = VM.getVM(2); |
| final VM client = VM.getVM(1); |
| final int numObjects = 200; |
| try { |
| final int[] port = AvailablePortHelper.getRandomAvailableTCPPorts(2); |
| startCacheServer(server1, port[0], criticalThreshold, disabledQueryMonitorForLowMem, |
| queryTimeout, regionName, createPR, 0); |
| startCacheServer(server2, port[1], criticalThreshold, true, -1, regionName, createPR, 0); |
| |
| startClient(client, server1, port[0], regionName); |
| populateData(server1, regionName, numObjects); |
| |
| createCancelDuringGatherTestHook(server1); |
| server1.invoke(new SerializableCallable("create index") { |
| @Override |
| public Object call() { |
| QueryService qs = null; |
| try { |
| qs = getCache().getQueryService(); |
| Index index = null; |
| if (indexType.equals("compact")) { |
| index = qs.createIndex("newIndex", "ID", "/" + regionName); |
| } else if (indexType.equals("hash")) { |
| index = qs.createHashIndex("newIndex", "ID", "/" + regionName); |
| } |
| assertNotNull(index); |
| assertTrue(((CancelDuringGatherHook) DefaultQuery.testHook).triggeredOOME); |
| |
| if (hitCriticalThreshold && !disabledQueryMonitorForLowMem) { |
| throw new CacheException("Should have hit low memory") {}; |
| } |
| assertEquals(1, qs.getIndexes().size()); |
| } catch (Exception e) { |
| if (e instanceof IndexInvalidException) { |
| if (!hitCriticalThreshold || disabledQueryMonitorForLowMem) { |
| throw new CacheException("Should not have run into low memory exception") {}; |
| } |
| } else { |
| throw new CacheException(e) {}; |
| } |
| } |
| return 0; |
| } |
| }); |
| } finally { |
| stopServer(server1); |
| stopServer(server2); |
| } |
| } |
| |
| private void doCriticalMemoryHitAddResultsTestWithMultipleServers(final String regionName, |
| boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) |
| throws Exception { |
| // create region on the server |
| final VM server1 = VM.getVM(0); |
| final VM server2 = VM.getVM(1); |
| final VM client = VM.getVM(2); |
| final int numObjects = 200; |
| try { |
| final int[] port = AvailablePortHelper.getRandomAvailableTCPPorts(2); |
| startCacheServer(server1, port[0], criticalThreshold, disabledQueryMonitorForLowMem, |
| queryTimeout, regionName, createPR, 0); |
| startCacheServer(server2, port[1], criticalThreshold, true, -1, regionName, createPR, 0); |
| |
| startClient(client, server1, port[0], regionName); |
| populateData(server2, regionName, numObjects); |
| |
| createCancelDuringAddResultsTestHook(server1); |
| client.invoke(new SerializableCallable("executing query to be canceled during add results") { |
| @Override |
| public Object call() { |
| QueryService qs = null; |
| try { |
| qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| SelectResults results = (SelectResults) query.execute(); |
| if (hitCriticalThreshold && disabledQueryMonitorForLowMem == false) { |
| throw new CacheException("should have hit low memory") {}; |
| } |
| } catch (Exception e) { |
| handleException(e, hitCriticalThreshold, disabledQueryMonitorForLowMem, queryTimeout); |
| } |
| return 0; |
| } |
| }); |
| |
| verifyRejectedObjects(server1, disabledQueryMonitorForLowMem, queryTimeout, |
| hitCriticalThreshold); |
| // Pause for a second and then let's recover |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server1); |
| } |
| |
| // Check to see if query execution is ok under "normal" or "healthy" conditions |
| client.invoke(new CacheSerializableRunnable("Executing query when system is 'Normal'") { |
| @Override |
| public void run2() { |
| try { |
| QueryService qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| SelectResults results = (SelectResults) query.execute(); |
| assertEquals(numObjects, results.size()); |
| } catch (QueryInvocationTargetException e) { |
| assertFalse(true); |
| } catch (NameResolutionException e) { |
| assertFalse(true); |
| } catch (TypeMismatchException e) { |
| assertFalse(true); |
| } catch (FunctionDomainException e) { |
| assertFalse(true); |
| } |
| } |
| }); |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server1); |
| } |
| } finally { |
| stopServer(server1); |
| stopServer(server2); |
| } |
| } |
| |
| // tests low memory hit while gathering partition region results |
| private void doCriticalMemoryHitDuringGatherTestWithMultipleServers(final String regionName, |
| boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) |
| throws Exception { |
| // create region on the server |
| final VM server1 = VM.getVM(0); |
| final VM server2 = VM.getVM(1); |
| final VM client = VM.getVM(2); |
| final int numObjects = 200; |
| try { |
| final int[] port = AvailablePortHelper.getRandomAvailableTCPPorts(2); |
| startCacheServer(server1, port[0], criticalThreshold, disabledQueryMonitorForLowMem, |
| queryTimeout, regionName, createPR, 0); |
| startCacheServer(server2, port[1], criticalThreshold, true, -1, regionName, createPR, 0); |
| |
| startClient(client, server1, port[0], regionName); |
| populateData(server2, regionName, numObjects); |
| |
| createCancelDuringGatherTestHook(server1); |
| client.invoke(new SerializableCallable("executing query to be canceled by gather") { |
| @Override |
| public Object call() { |
| QueryService qs = null; |
| try { |
| qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| query.execute(); |
| } catch (ServerOperationException soe) { |
| if (soe.getRootCause() instanceof QueryException) { |
| QueryException e = (QueryException) soe.getRootCause(); |
| if (!isExceptionDueToLowMemory(e, CRITICAL_HEAP_USED)) { |
| throw new CacheException(soe) {}; |
| } else { |
| return 0; |
| } |
| } |
| } catch (Exception e) { |
| throw new CacheException(e) {}; |
| } |
| // assertTrue(((CancelDuringGatherHook)DefaultQuery.testHook).triggeredOOME); |
| throw new CacheException("should have hit low memory") {}; |
| } |
| }); |
| |
| verifyRejectedObjects(server1, disabledQueryMonitorForLowMem, queryTimeout, |
| hitCriticalThreshold); |
| // Pause for a second and then let's recover |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server1); |
| } |
| |
| // Check to see if query execution is ok under "normal" or "healthy" conditions |
| client.invoke(new CacheSerializableRunnable("Executing query when system is 'Normal'") { |
| @Override |
| public void run2() { |
| try { |
| QueryService qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| SelectResults results = (SelectResults) query.execute(); |
| assertEquals(numObjects, results.size()); |
| } catch (QueryInvocationTargetException e) { |
| assertFalse(true); |
| } catch (NameResolutionException e) { |
| assertFalse(true); |
| } catch (TypeMismatchException e) { |
| assertFalse(true); |
| } catch (FunctionDomainException e) { |
| assertFalse(true); |
| } |
| } |
| }); |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server1); |
| } |
| } finally { |
| stopServer(server1); |
| stopServer(server2); |
| } |
| } |
| |
| // Executes on client cache with multiple configured servers |
| private void doCriticalMemoryHitTestWithMultipleServers(final String regionName, boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) |
| throws Exception { |
| // create region on the server |
| final VM server1 = VM.getVM(0); |
| final VM server2 = VM.getVM(1); |
| final VM client = VM.getVM(2); |
| final int numObjects = 200; |
| |
| try { |
| final int[] port = AvailablePortHelper.getRandomAvailableTCPPorts(2); |
| startCacheServer(server1, port[0], criticalThreshold, disabledQueryMonitorForLowMem, |
| queryTimeout, regionName, createPR, 0); |
| startCacheServer(server2, port[1], criticalThreshold, true, -1, regionName, createPR, 0); |
| |
| startClient(client, server1, port[0], regionName); |
| populateData(server2, regionName, numObjects); |
| |
| doTestCriticalHeapAndQueryTimeout(server1, client, regionName, disabledQueryMonitorForLowMem, |
| queryTimeout, hitCriticalThreshold); |
| verifyRejectedObjects(server1, disabledQueryMonitorForLowMem, queryTimeout, |
| hitCriticalThreshold); |
| // Pause for a second and then let's recover |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server1); |
| } |
| |
| // Check to see if query execution is ok under "normal" or "healthy" conditions |
| client.invoke(new CacheSerializableRunnable("Executing query when system is 'Normal'") { |
| @Override |
| public void run2() { |
| try { |
| QueryService qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| SelectResults results = (SelectResults) query.execute(); |
| assertEquals(numObjects, results.size()); |
| } catch (QueryInvocationTargetException e) { |
| assertFalse(true); |
| } catch (NameResolutionException e) { |
| assertFalse(true); |
| } catch (TypeMismatchException e) { |
| assertFalse(true); |
| } catch (FunctionDomainException e) { |
| assertFalse(true); |
| } |
| } |
| }); |
| |
| // Execute a critical heap event/ query timeout test again |
| doTestCriticalHeapAndQueryTimeout(server1, client, regionName, disabledQueryMonitorForLowMem, |
| queryTimeout, hitCriticalThreshold); |
| verifyRejectedObjects(server1, disabledQueryMonitorForLowMem, queryTimeout, |
| hitCriticalThreshold); |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server1); |
| } |
| } finally { |
| stopServer(server1); |
| stopServer(server2); |
| } |
| } |
| |
| // Executes the query on the server with the RM and QM configured |
| private void doCriticalMemoryHitTestOnServer(final String regionName, boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) |
| throws Exception { |
| // create region on the server |
| final VM server = VM.getVM(0); |
| final int numObjects = 200; |
| try { |
| final int port = AvailablePortHelper.getRandomAvailableTCPPort(); |
| startCacheServer(server, port, criticalThreshold, disabledQueryMonitorForLowMem, queryTimeout, |
| regionName, createPR, 0); |
| |
| // startPeerClient(client, server, port, regionName); |
| populateData(server, regionName, numObjects); |
| |
| doTestCriticalHeapAndQueryTimeout(server, server, regionName, disabledQueryMonitorForLowMem, |
| queryTimeout, hitCriticalThreshold); |
| |
| // Pause for a second and then let's recover |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server); |
| } |
| |
| // Check to see if query execution is ok under "normal" or "healthy" conditions |
| server.invoke(new CacheSerializableRunnable("Executing query when system is 'Normal'") { |
| @Override |
| public void run2() { |
| try { |
| QueryService qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| SelectResults results = (SelectResults) query.execute(); |
| assertEquals(numObjects, results.size()); |
| } catch (QueryInvocationTargetException e) { |
| assertFalse(true); |
| } catch (NameResolutionException e) { |
| assertFalse(true); |
| } catch (TypeMismatchException e) { |
| assertFalse(true); |
| } catch (FunctionDomainException e) { |
| assertFalse(true); |
| } |
| } |
| }); |
| |
| // Execute a critical heap event/ query timeout test again |
| doTestCriticalHeapAndQueryTimeout(server, server, regionName, disabledQueryMonitorForLowMem, |
| queryTimeout, hitCriticalThreshold); |
| |
| // Recover from critical heap |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server); |
| } |
| } finally { |
| stopServer(server); |
| } |
| } |
| |
| |
| private void executeQueryFromClientWithTimeoutSetAndCriticalThreshold(final String regionName, |
| boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) |
| throws Exception { |
| // create region on the server |
| final VM server = VM.getVM(0); |
| final VM client = VM.getVM(1); |
| final int numObjects = 200; |
| try { |
| final int port = AvailablePortHelper.getRandomAvailableTCPPort(); |
| startCacheServer(server, port, criticalThreshold, disabledQueryMonitorForLowMem, queryTimeout, |
| regionName, createPR, 0); |
| startClient(client, server, port, regionName); |
| populateData(server, regionName, numObjects); |
| executeQueryWithCriticalHeapCalledAfterTimeout(server, client, regionName, queryTimeout, |
| hitCriticalThreshold); |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server); |
| } |
| |
| } finally { |
| stopServer(server); |
| } |
| } |
| |
| private void executeQueryWithTimeoutSetAndCriticalThreshold(final String regionName, |
| boolean createPR, |
| final int criticalThreshold, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) |
| throws Exception { |
| // create region on the server |
| final VM server = VM.getVM(0); |
| final int numObjects = 200; |
| try { |
| final int port = AvailablePortHelper.getRandomAvailableTCPPort(); |
| startCacheServer(server, port, criticalThreshold, disabledQueryMonitorForLowMem, queryTimeout, |
| regionName, createPR, 0); |
| populateData(server, regionName, numObjects); |
| executeQueryWithCriticalHeapCalledAfterTimeout(server, server, regionName, queryTimeout, |
| hitCriticalThreshold); |
| if (hitCriticalThreshold) { |
| vmRecoversFromCriticalHeap(server); |
| } |
| |
| } finally { |
| stopServer(server); |
| } |
| } |
| |
| // This helper method will set up a test hook |
| // Execute a query on the server, pause due to the test hook |
| // Execute a critical heap event |
| // release the test hook |
| // Check to see that the query either failed due to critical heap if query monitor is not disabled |
| // or it will fail due to time out, due to the sleeps we put in |
| // If timeout is disabled/not set, then the query should execute just fine |
| // The last part of the test is to execute another query with the system under duress and have it |
| // be rejected/cancelled if rm and qm are in use |
| private void doTestCriticalHeapAndQueryTimeout(VM server, VM client, final String regionName, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) { |
| createLatchTestHook(server); |
| |
| AsyncInvocation queryExecution = invokeClientQuery(client, regionName, |
| disabledQueryMonitorForLowMem, queryTimeout, hitCriticalThreshold); |
| |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| // We simulate a low memory/critical heap percentage hit |
| if (hitCriticalThreshold) { |
| vmHitsCriticalHeap(server); |
| } |
| |
| // Pause until query would time out if low memory was ignored |
| try { |
| Thread.sleep(MAX_TEST_QUERY_TIMEOUT); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| // release the hook to have the query throw either a low memory or query timeout |
| // unless otherwise configured |
| releaseHook(server); |
| |
| ThreadUtils.join(queryExecution, 60000); |
| // Make sure no exceptions were thrown during query testing |
| try { |
| assertEquals(0, queryExecution.getResult()); |
| } catch (Throwable e) { |
| e.printStackTrace(); |
| fail("queryExecution.getResult() threw Exception " + e.toString()); |
| } |
| } |
| |
| private void executeQueryWithCriticalHeapCalledAfterTimeout(VM server, VM client, |
| final String regionName, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) { |
| createLatchTestHook(server); |
| AsyncInvocation queryExecution = executeQueryWithTimeout(client, regionName, queryTimeout); |
| |
| // Wait till the timeout expires on the query |
| try { |
| Thread.sleep(queryTimeout + 1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| // We simulate a low memory/critical heap percentage hit |
| // But by design of this test the query must have been already terminated because of a 1ms |
| // timeout |
| if (hitCriticalThreshold) { |
| vmHitsCriticalHeap(server); |
| } |
| |
| releaseHook(server); |
| |
| ThreadUtils.join(queryExecution, 60000); |
| // Make sure no exceptions were thrown during query testing |
| try { |
| assertEquals(0, queryExecution.getResult()); |
| } catch (Throwable e) { |
| e.printStackTrace(); |
| fail("queryExecution.getResult() threw Exception " + e.toString()); |
| } |
| } |
| |
| private AsyncInvocation executeQueryWithTimeout(VM client, final String regionName, |
| final int queryTimeout) { |
| return client.invokeAsync(new SerializableCallable("execute query from client") { |
| @Override |
| public Object call() throws CacheException { |
| QueryService qs = null; |
| try { |
| qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| query.execute(); |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| if (e instanceof QueryExecutionTimeoutException) { |
| logger.info("Query Execution must be terminated by a timeout."); |
| return 0; |
| } |
| if (e instanceof ServerOperationException) { |
| ServerOperationException soe = (ServerOperationException) e; |
| if (soe.getRootCause() instanceof QueryException) { |
| QueryException qe = (QueryException) soe.getRootCause(); |
| if (isExceptionDueToTimeout(qe)) { |
| logger.info("Query Execution must be terminated by a timeout. Expected behavior"); |
| return 0; |
| } |
| } else if (soe.getRootCause() instanceof QueryExecutionTimeoutException) { |
| logger.info("Query Execution must be terminated by a timeout."); |
| return 0; |
| } |
| } |
| e.printStackTrace(); |
| throw new CacheException( |
| "The query should have been terminated by a timeout exception but instead hit a different exception :" |
| + e) {}; |
| } |
| return -1; |
| } |
| }); |
| |
| } |
| |
| private AsyncInvocation invokeClientQuery(VM client, final String regionName, |
| final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, |
| final boolean hitCriticalThreshold) { |
| return client.invokeAsync(new SerializableCallable("execute query from client") { |
| @Override |
| public Object call() throws CacheException { |
| QueryService qs = null; |
| try { |
| qs = getCache().getQueryService(); |
| Query query = qs.newQuery("Select * From /" + regionName); |
| query.execute(); |
| |
| if (disabledQueryMonitorForLowMem) { |
| if (queryTimeout != -1) { |
| // we should have timed out due to the way the test is written |
| // the query should have hit the configured timeouts |
| throw new CacheException("Should have reached the query timeout") {}; |
| } |
| } else { |
| if (hitCriticalThreshold) { |
| throw new CacheException("Exception should have been thrown due to low memory") {}; |
| } |
| } |
| } catch (Exception e) { |
| handleException(e, hitCriticalThreshold, disabledQueryMonitorForLowMem, queryTimeout); |
| } |
| |
| try { |
| Query query = qs.newQuery("Select * From /" + regionName); |
| query.execute(); |
| if (hitCriticalThreshold && disabledQueryMonitorForLowMem == false) { |
| throw new CacheException("Low memory should still be cancelling queries") {}; |
| } |
| } catch (Exception e) { |
| handleException(e, hitCriticalThreshold, disabledQueryMonitorForLowMem, queryTimeout); |
| } |
| return 0; |
| } |
| }); |
| |
| } |
| |
| private void handleException(Exception e, boolean hitCriticalThreshold, |
| boolean disabledQueryMonitorForLowMem, long queryTimeout) |
| throws CacheException { |
| if (e instanceof QueryExecutionLowMemoryException) { |
| if (!(hitCriticalThreshold && disabledQueryMonitorForLowMem == false)) { |
| // meaning the query should not be canceled due to low memory |
| throw new CacheException("Query should not have been canceled due to memory") {}; |
| } |
| } else if (e instanceof QueryExecutionTimeoutException) { |
| // if we have a queryTimeout set |
| if (queryTimeout == -1) { |
| // no time out set, this should not be thrown |
| throw new CacheException( |
| "Query failed due to unexplained reason, should not have been a time out or low memory " |
| + DefaultQuery.testHook.getClass().getName() + " " + e) {}; |
| } |
| } else if (e instanceof QueryException) { |
| if (isExceptionDueToLowMemory((QueryException) e, CRITICAL_HEAP_USED)) { |
| if (!(hitCriticalThreshold && disabledQueryMonitorForLowMem == false)) { |
| // meaning the query should not be canceled due to low memory |
| throw new CacheException("Query should not have been canceled due to memory") {}; |
| } |
| } else if (isExceptionDueToTimeout((QueryException) e)) { |
| |
| if (queryTimeout == -1) { |
| // no time out set, this should not be thrown |
| throw new CacheException( |
| "Query failed due to unexplained reason, should not have been a time out or low memory") {}; |
| } |
| } else { |
| throw new CacheException(e) {}; |
| } |
| } else if (e instanceof ServerOperationException) { |
| ServerOperationException soe = (ServerOperationException) e; |
| if (soe.getRootCause() instanceof QueryExecutionLowMemoryException) { |
| if (!(hitCriticalThreshold && disabledQueryMonitorForLowMem == false)) { |
| // meaning the query should not be canceled due to low memory |
| throw new CacheException("Query should not have been canceled due to memory") {}; |
| } |
| } else if (soe.getRootCause() instanceof QueryException) { |
| QueryException qe = (QueryException) soe.getRootCause(); |
| if (isExceptionDueToLowMemory(qe, CRITICAL_HEAP_USED)) { |
| if (!(hitCriticalThreshold && disabledQueryMonitorForLowMem == false)) { |
| // meaning the query should not be canceled due to low memory |
| throw new CacheException("Query should not have been canceled due to memory") {}; |
| } |
| } else if (isExceptionDueToTimeout(qe)) { |
| if (queryTimeout == -1) { |
| e.printStackTrace(); |
| // no time out set, this should not be thrown |
| throw new CacheException( |
| "Query failed due to unexplained reason, should not have been a time out or low memory") {}; |
| } |
| } else { |
| throw new CacheException(soe) {}; |
| } |
| } else if (soe.getRootCause() instanceof QueryExecutionTimeoutException) { |
| // if we have a queryTimeout set |
| if (queryTimeout == -1) { |
| // no time out set, this should not be thrown |
| throw new CacheException( |
| "Query failed due to unexplained reason, should not have been a time out or low memory " |
| + DefaultQuery.testHook.getClass().getName() + " " + soe.getRootCause()) {}; |
| } |
| } else { |
| throw new CacheException(soe) {}; |
| } |
| } else { |
| throw new CacheException(e) {}; |
| } |
| } |
| |
| |
| private void vmHitsCriticalHeap(VM vm) { |
| vm.invoke(new CacheSerializableRunnable("vm hits critical heap") { |
| @Override |
| public void run2() { |
| InternalResourceManager resourceManager = |
| (InternalResourceManager) getCache().getResourceManager(); |
| resourceManager.getHeapMonitor().updateStateAndSendEvent(CRITICAL_HEAP_USED, "test"); |
| } |
| }); |
| } |
| |
| private void vmRecoversFromCriticalHeap(VM vm) { |
| vm.invoke(new CacheSerializableRunnable("vm hits critical heap") { |
| @Override |
| public void run2() { |
| InternalResourceManager resourceManager = |
| (InternalResourceManager) getCache().getResourceManager(); |
| resourceManager.getHeapMonitor().updateStateAndSendEvent(NORMAL_HEAP_USED, "test"); |
| } |
| }); |
| } |
| |
| private void createLatchTestHook(VM vm) { |
| vm.invoke(new CacheSerializableRunnable("create latch test Hook") { |
| @Override |
| public void run2() { |
| DefaultQuery.TestHook hook = getPauseHook(); |
| DefaultQuery.testHook = hook; |
| } |
| }); |
| } |
| |
| private void createCancelDuringGatherTestHook(VM vm) { |
| vm.invoke(new CacheSerializableRunnable("create cancel during gather test Hook") { |
| @Override |
| public void run2() { |
| DefaultQuery.TestHook hook = getCancelDuringGatherHook(); |
| DefaultQuery.testHook = hook; |
| } |
| }); |
| } |
| |
| private void createCancelDuringAddResultsTestHook(VM vm) { |
| vm.invoke(new CacheSerializableRunnable("create cancel during gather test Hook") { |
| @Override |
| public void run2() { |
| DefaultQuery.TestHook hook = getCancelDuringAddResultsHook(); |
| DefaultQuery.testHook = hook; |
| } |
| }); |
| } |
| |
| |
| private void releaseHook(VM vm) { |
| vm.invoke(new CacheSerializableRunnable("release latch Hook") { |
| @Override |
| public void run2() { |
| PauseTestHook hook = (PauseTestHook) DefaultQuery.testHook; |
| hook.countDown(); |
| } |
| }); |
| } |
| |
| // Verify that PRQueryEvaluator dropped objects if low memory |
| private void verifyRejectedObjects(VM vm, final boolean disabledQueryMonitorForLowMem, |
| final int queryTimeout, final boolean hitCriticalThreshold) { |
| vm.invoke(new CacheSerializableRunnable("verify dropped objects") { |
| @Override |
| public void run2() { |
| if ((disabledQueryMonitorForLowMem == false && hitCriticalThreshold)) { |
| if (DefaultQuery.testHook instanceof PauseTestHook) { |
| PauseTestHook hook = (PauseTestHook) DefaultQuery.testHook; |
| assertTrue(hook.rejectedObjects); |
| } else if (DefaultQuery.testHook instanceof CancelDuringGatherHook) { |
| CancelDuringGatherHook hook = (CancelDuringGatherHook) DefaultQuery.testHook; |
| assertTrue(hook.rejectedObjects); |
| } else if (DefaultQuery.testHook instanceof CancelDuringAddResultsHook) { |
| CancelDuringAddResultsHook hook = (CancelDuringAddResultsHook) DefaultQuery.testHook; |
| assertTrue(hook.rejectedObjects); |
| } |
| } |
| } |
| |
| }); |
| } |
| |
| private void populateData(VM vm, final String regionName, final int numObjects) { |
| vm.invoke(new CacheSerializableRunnable("populate data for " + regionName) { |
| @Override |
| public void run2() { |
| Region region = getCache().getRegion(regionName); |
| for (int i = 0; i < numObjects; i++) { |
| region.put("key_" + i, new Portfolio(i)); |
| } |
| } |
| }); |
| } |
| |
| private void stopServer(VM server) { |
| server.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| GemFireCacheImpl cache = (GemFireCacheImpl) getCache(); |
| cache.MAX_QUERY_EXECUTION_TIME = -1; |
| return null; |
| } |
| }); |
| } |
| |
| private void startCacheServer(VM server, final int port, final int criticalThreshold, |
| final boolean disableQueryMonitorForLowMemory, |
| final int queryTimeout, |
| final String regionName, final boolean createPR, |
| final int prRedundancy) throws Exception { |
| |
| server.invoke(new SerializableCallable() { |
| @Override |
| public Object call() throws Exception { |
| getSystem(getServerProperties(disableQueryMonitorForLowMemory, queryTimeout)); |
| if (disableQueryMonitorForLowMemory == true) { |
| System.setProperty( |
| DistributionConfig.GEMFIRE_PREFIX + "Cache.DISABLE_QUERY_MONITOR_FOR_LOW_MEMORY", |
| "true"); |
| } else { |
| System.clearProperty( |
| DistributionConfig.GEMFIRE_PREFIX + "Cache.DISABLE_QUERY_MONITOR_FOR_LOW_MEMORY"); |
| } |
| |
| GemFireCacheImpl cache = (GemFireCacheImpl) getCache(); |
| |
| if (queryTimeout != -1) { |
| cache.MAX_QUERY_EXECUTION_TIME = queryTimeout; |
| } else { |
| cache.MAX_QUERY_EXECUTION_TIME = -1; |
| } |
| |
| if (criticalThreshold != 0) { |
| InternalResourceManager resourceManager = |
| (InternalResourceManager) cache.getResourceManager(); |
| HeapMemoryMonitor heapMonitor = resourceManager.getHeapMonitor(); |
| heapMonitor.setTestMaxMemoryBytes(1000); |
| HeapMemoryMonitor.setTestBytesUsedForThresholdSet(NORMAL_HEAP_USED); |
| resourceManager.setCriticalHeapPercentage(criticalThreshold); |
| } |
| |
| AttributesFactory factory = new AttributesFactory(); |
| if (createPR) { |
| PartitionAttributesFactory paf = new PartitionAttributesFactory(); |
| paf.setRedundantCopies(prRedundancy); |
| paf.setTotalNumBuckets(11); |
| factory.setPartitionAttributes(paf.create()); |
| } else { |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setDataPolicy(DataPolicy.REPLICATE); |
| } |
| Region region = createRootRegion(regionName, factory.create()); |
| if (createPR) { |
| assertTrue(region instanceof PartitionedRegion); |
| } else { |
| assertTrue(region instanceof DistributedRegion); |
| } |
| CacheServer cacheServer = getCache().addCacheServer(); |
| cacheServer.setPort(port); |
| cacheServer.start(); |
| |
| return null; |
| } |
| }); |
| } |
| |
| private void startClient(VM client, final VM server, final int port, final String regionName) { |
| |
| client.invoke(new CacheSerializableRunnable("Start client") { |
| @Override |
| public void run2() throws CacheException { |
| Properties props = getClientProps(); |
| getSystem(props); |
| |
| final ClientCacheFactory ccf = new ClientCacheFactory(props); |
| ccf.addPoolServer(NetworkUtils.getServerHostName(server.getHost()), port); |
| ClientCache cache = (ClientCache) getClientCache(ccf); |
| } |
| }); |
| } |
| |
| private void startPeerClient(VM client, final VM server, final int port, |
| final String regionName) { |
| |
| client.invoke(new CacheSerializableRunnable("Start peer client") { |
| @Override |
| public void run2() throws CacheException { |
| Properties props = getClientProps(); |
| getSystem(props); |
| |
| PoolFactory pf = PoolManager.createFactory(); |
| pf.addServer(NetworkUtils.getServerHostName(server.getHost()), port); |
| pf.create("pool1"); |
| |
| AttributesFactory af = new AttributesFactory(); |
| af.setScope(Scope.LOCAL); |
| af.setPoolName("pool1"); |
| Region region = createRootRegion(regionName, af.create()); |
| |
| getCache(); |
| } |
| }); |
| } |
| |
| protected Properties getClientProps() { |
| Properties p = new Properties(); |
| p.setProperty(MCAST_PORT, "0"); |
| p.setProperty(LOCATORS, ""); |
| return p; |
| } |
| |
| protected Properties getServerProperties(boolean disableQueryMonitorForMemory, int queryTimeout) { |
| Properties p = new Properties(); |
| p.setProperty(LOCATORS, "localhost[" + DistributedTestUtils.getDUnitLocatorPort() + "]"); |
| return p; |
| } |
| |
| private boolean isExceptionDueToLowMemory(QueryException e, int HEAP_USED) { |
| String message = e.getMessage(); |
| return (message.contains( |
| String.format( |
| "Query execution canceled due to memory threshold crossed in system, memory used: %s bytes.", |
| HEAP_USED)) |
| || message.contains( |
| "Query execution canceled due to low memory while gathering results from partitioned regions")); |
| } |
| |
| private boolean isExceptionDueToTimeout(QueryException e) { |
| String message = e.getMessage(); |
| // -1 needs to be matched due to client/server set up, BaseCommand uses the |
| // MAX_QUERY_EXECUTION_TIME and not the testMaxQueryExecutionTime |
| return (message.contains("The QueryMonitor thread may be sleeping longer than") |
| || message.contains("Query execution canceled after exceeding max execution time") |
| || message.contains( |
| String.format("Query execution canceled after exceeding max execution time %sms.", |
| -1))); |
| } |
| |
| private DefaultQuery.TestHook getPauseHook() { |
| return new PauseTestHook(); |
| } |
| |
| private DefaultQuery.TestHook getCancelDuringGatherHook() { |
| return new CancelDuringGatherHook(); |
| } |
| |
| private DefaultQuery.TestHook getCancelDuringAddResultsHook() { |
| return new CancelDuringAddResultsHook(); |
| } |
| |
| private static class PauseTestHook implements DefaultQuery.TestHook { |
| private CountDownLatch latch = new CountDownLatch(1); |
| public boolean rejectedObjects = false; |
| |
| @Override |
| public void doTestHook(final SPOTS spot, final DefaultQuery _ignored, |
| final ExecutionContext executionContext) { |
| switch (spot) { |
| case BEFORE_QUERY_EXECUTION: |
| try { |
| if (!latch.await(8, TimeUnit.SECONDS)) { |
| fail("query was never unlatched"); |
| } |
| } catch (InterruptedException e) { |
| e.printStackTrace(); |
| Thread.currentThread().interrupt(); |
| } |
| break; |
| case LOW_MEMORY_WHEN_DESERIALIZING_STREAMINGOPERATION: |
| rejectedObjects = true; |
| break; |
| } |
| } |
| |
| public void countDown() { |
| latch.countDown(); |
| } |
| } |
| |
| // non-static class because it needs to call getCache() |
| private class CancelDuringGatherHook implements DefaultQuery.TestHook { |
| public boolean rejectedObjects = false; |
| public boolean triggeredOOME = false; |
| private int count = 0; |
| private int numObjectsBeforeCancel = 5; |
| |
| @Override |
| public void doTestHook(final SPOTS spot, final DefaultQuery _ignored, |
| final ExecutionContext executionContext) { |
| switch (spot) { |
| case LOW_MEMORY_WHEN_DESERIALIZING_STREAMINGOPERATION: |
| rejectedObjects = true; |
| break; |
| case BEFORE_ADD_OR_UPDATE_MAPPING_OR_DESERIALIZING_NTH_STREAMINGOPERATION: |
| if (count++ == numObjectsBeforeCancel) { |
| InternalResourceManager resourceManager = |
| (InternalResourceManager) getCache().getResourceManager(); |
| resourceManager.getHeapMonitor().updateStateAndSendEvent(CRITICAL_HEAP_USED, "test"); |
| triggeredOOME = true; |
| } |
| break; |
| } |
| } |
| } |
| |
| // non-static class because it needs to call getCache() |
| private class CancelDuringAddResultsHook implements DefaultQuery.TestHook { |
| public boolean triggeredOOME = false; |
| public boolean rejectedObjects = false; |
| |
| @Override |
| public void doTestHook(final SPOTS spot, final DefaultQuery _ignored, |
| final ExecutionContext executionContext) { |
| switch (spot) { |
| case BEFORE_BUILD_CUMULATIVE_RESULT: |
| if (triggeredOOME == false) { |
| InternalResourceManager resourceManager = |
| (InternalResourceManager) getCache().getResourceManager(); |
| resourceManager.getHeapMonitor().updateStateAndSendEvent(CRITICAL_HEAP_USED, "test"); |
| triggeredOOME = true; |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| break; |
| case BEFORE_THROW_QUERY_CANCELED_EXCEPTION: |
| rejectedObjects = true; |
| break; |
| } |
| } |
| } |
| } |
