| /* |
| * 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.execute; |
| |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| 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 static org.junit.runners.Parameterized.UseParametersRunnerFactory; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Properties; |
| import java.util.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| import org.junit.runner.RunWith; |
| import org.junit.runners.Parameterized; |
| |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.CacheFactory; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionAttributes; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.client.Pool; |
| import org.apache.geode.cache.client.PoolManager; |
| import org.apache.geode.cache.execute.Execution; |
| import org.apache.geode.cache.execute.Function; |
| import org.apache.geode.cache.execute.FunctionAdapter; |
| import org.apache.geode.cache.execute.FunctionContext; |
| import org.apache.geode.cache.execute.FunctionService; |
| import org.apache.geode.cache.execute.ResultCollector; |
| import org.apache.geode.cache.execute.ResultSender; |
| import org.apache.geode.cache.server.CacheServer; |
| import org.apache.geode.distributed.ConfigurationProperties; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.DistributedSystem; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.internal.AvailablePort; |
| import org.apache.geode.internal.cache.PartitionAttributesImpl; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.PartitionedRegionTestHelper; |
| import org.apache.geode.internal.cache.execute.metrics.FunctionServiceStats; |
| import org.apache.geode.internal.cache.execute.metrics.FunctionStats; |
| import org.apache.geode.internal.cache.execute.metrics.FunctionStatsManager; |
| import org.apache.geode.internal.cache.functions.TestFunction; |
| import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| import org.apache.geode.test.dunit.Assert; |
| import org.apache.geode.test.dunit.Host; |
| import org.apache.geode.test.dunit.SerializableCallableIF; |
| import org.apache.geode.test.dunit.SerializableRunnableIF; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.Wait; |
| import org.apache.geode.test.junit.categories.FunctionServiceTest; |
| import org.apache.geode.test.junit.runners.CategoryWithParameterizedRunnerFactory; |
| |
| /** |
| * This is DUnite Test to test the Function Execution stats under various scenarios like |
| * Client-Server with Region/without Region, P2P with partitioned Region/Distributed Region,member |
| * Execution |
| */ |
| @Category({FunctionServiceTest.class}) |
| @RunWith(Parameterized.class) |
| @UseParametersRunnerFactory(CategoryWithParameterizedRunnerFactory.class) |
| public class FunctionServiceStatsDUnitTest extends PRClientServerTestBase { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| static InternalDistributedSystem ds = null; |
| |
| private static int noOfExecutionCalls_Aggregate = 0; |
| private static int noOfExecutionsCompleted_Aggregate = 0; |
| private static int resultReceived_Aggregate = 0; |
| |
| private static int noOfExecutionCalls_TESTFUNCTION1 = 0; |
| private static int noOfExecutionsCompleted_TESTFUNCTION1 = 0; |
| private static int resultReceived_TESTFUNCTION1 = 0; |
| |
| private static int noOfExecutionCalls_TESTFUNCTION2 = 0; |
| private static int noOfExecutionsCompleted_TESTFUNCTION2 = 0; |
| private static int resultReceived_TESTFUNCTION2 = 0; |
| |
| private static int noOfExecutionCalls_TESTFUNCTION3 = 0; |
| private static int noOfExecutionsCompleted_TESTFUNCTION3 = 0; |
| private static int resultReceived_TESTFUNCTION3 = 0; |
| |
| private static int noOfExecutionCalls_TESTFUNCTION5 = 0; |
| private static int noOfExecutionsCompleted_TESTFUNCTION5 = 0; |
| private static int resultReceived_TESTFUNCTION5 = 0; |
| |
| private static int noOfExecutionCalls_Inline = 0; |
| private static int noOfExecutionsCompleted_Inline = 0; |
| private static int resultReceived_Inline = 0; |
| |
| @Override |
| protected final void postSetUpPRClientServerTestBase() { |
| // Make sure stats to linger from a previous test |
| disconnectAllFromDS(); |
| } |
| |
| private final transient SerializableRunnableIF initializeStats = () -> { |
| noOfExecutionCalls_Aggregate = 0; |
| noOfExecutionsCompleted_Aggregate = 0; |
| resultReceived_Aggregate = 0; |
| |
| noOfExecutionCalls_TESTFUNCTION1 = 0; |
| noOfExecutionsCompleted_TESTFUNCTION1 = 0; |
| resultReceived_TESTFUNCTION1 = 0; |
| |
| noOfExecutionCalls_TESTFUNCTION2 = 0; |
| noOfExecutionsCompleted_TESTFUNCTION2 = 0; |
| resultReceived_TESTFUNCTION2 = 0; |
| |
| noOfExecutionCalls_TESTFUNCTION3 = 0; |
| noOfExecutionsCompleted_TESTFUNCTION3 = 0; |
| resultReceived_TESTFUNCTION3 = 0; |
| |
| noOfExecutionCalls_TESTFUNCTION5 = 0; |
| noOfExecutionsCompleted_TESTFUNCTION5 = 0; |
| resultReceived_TESTFUNCTION5 = 0; |
| |
| noOfExecutionCalls_Inline = 0; |
| noOfExecutionsCompleted_Inline = 0; |
| resultReceived_Inline = 0; |
| }; |
| |
| private final transient SerializableRunnableIF closeDistributedSystem = () -> { |
| if (getCache() != null && !getCache().isClosed()) { |
| getCache().close(); |
| getCache().getDistributedSystem().disconnect(); |
| } |
| }; |
| |
| /* |
| * This helper method prevents race conditions in local functions. Typically, when calling |
| * ResultCollector.getResult() one might expect the function to have completed. For local |
| * functions this is true, however, at this point the function stats may not have been updated yet |
| * thus any code which checks stats after calling getResult() may get wrong data. |
| */ |
| private void waitNoFunctionsRunning(FunctionServiceStats stats) { |
| int count = 100; |
| while (stats.getFunctionExecutionsRunning() > 0 && count > 0) { |
| count--; |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException ex) { |
| // Ignored |
| } |
| } |
| } |
| |
| /** |
| * 1-client 3-Servers Function : TEST_FUNCTION2 Function : TEST_FUNCTION3 Execution of the |
| * function on serverRegion with set multiple keys as the routing object and using the name of the |
| * function |
| * |
| * On server side, function execution calls should be equal to the no of function executions |
| * completed. |
| */ |
| @Test |
| public void testClientServerPartitonedRegionFunctionExecutionStats() { |
| createScenario(); |
| registerFunctionAtServer(new TestFunction(true, TestFunction.TEST_FUNCTION2)); |
| registerFunctionAtServer(new TestFunction(true, TestFunction.TEST_FUNCTION3)); |
| |
| client.invoke(initializeStats); |
| server1.invoke(initializeStats); |
| server2.invoke(initializeStats); |
| server3.invoke(initializeStats); |
| |
| client.invoke(() -> { |
| Region<String, Integer> region = cache.getRegion(PartitionedRegionName); |
| assertNotNull(region); |
| final HashSet<String> testKeysSet = new HashSet<>(); |
| for (int i = (totalNumBuckets * 2); i > 0; i--) { |
| testKeysSet.add("execKey-" + i); |
| } |
| DistributedSystem.setThreadsSocketPolicy(false); |
| Function function = new TestFunction(true, TestFunction.TEST_FUNCTION2); |
| if (shouldRegisterFunctionsOnClient()) { |
| FunctionService.registerFunction(function); |
| } |
| Execution dataSet = FunctionService.onRegion(region); |
| try { |
| int j = 0; |
| for (String s : testKeysSet) { |
| Integer val = j++; |
| region.put(s, val); |
| } |
| ResultCollector rc = |
| dataSet.withFilter(testKeysSet).setArguments(Boolean.TRUE).execute(function.getId()); |
| int resultSize = ((List) rc.getResult()).size(); |
| resultReceived_Aggregate += resultSize; |
| resultReceived_TESTFUNCTION2 += resultSize; |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionCalls_TESTFUNCTION2++; |
| noOfExecutionsCompleted_Aggregate++; |
| noOfExecutionsCompleted_TESTFUNCTION2++; |
| |
| rc = dataSet.withFilter(testKeysSet).setArguments(testKeysSet).execute(function.getId()); |
| resultSize = ((List) rc.getResult()).size(); |
| resultReceived_Aggregate += resultSize; |
| resultReceived_TESTFUNCTION2 += resultSize; |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionCalls_TESTFUNCTION2++; |
| noOfExecutionsCompleted_Aggregate++; |
| noOfExecutionsCompleted_TESTFUNCTION2++; |
| |
| function = new TestFunction(true, TestFunction.TEST_FUNCTION3); |
| FunctionService.registerFunction(function); |
| rc = dataSet.withFilter(testKeysSet).setArguments(Boolean.TRUE).execute(function.getId()); |
| resultSize = ((List) rc.getResult()).size(); |
| resultReceived_Aggregate += resultSize; |
| resultReceived_TESTFUNCTION3 += resultSize; |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionCalls_TESTFUNCTION3++; |
| noOfExecutionsCompleted_Aggregate++; |
| noOfExecutionsCompleted_TESTFUNCTION3++; |
| |
| } catch (Exception e) { |
| logger.info("Exception : " + e.getMessage()); |
| e.printStackTrace(); |
| fail("Test failed after the put operation"); |
| } |
| }); |
| |
| client.invoke(() -> { |
| // checks for the aggregate stats |
| InternalDistributedSystem iDS = (InternalDistributedSystem) cache.getDistributedSystem(); |
| FunctionServiceStats functionServiceStats = |
| iDS.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| assertTrue(functionServiceStats.getResultsReceived() >= resultReceived_Aggregate); |
| |
| logger.info("Calling FunctionStats for TEST_FUNCTION2 :"); |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION2, iDS); |
| logger.info("Called FunctionStats for TEST_FUNCTION2 :"); |
| assertEquals(noOfExecutionCalls_TESTFUNCTION2, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION2, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertTrue(functionStats.getResultsReceived() >= resultReceived_TESTFUNCTION2); |
| |
| functionStats = FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION3, iDS); |
| assertEquals(noOfExecutionCalls_TESTFUNCTION3, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION3, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertTrue(functionStats.getResultsReceived() >= resultReceived_TESTFUNCTION3); |
| }); |
| |
| SerializableRunnableIF checkStatsOnServer = () -> { |
| // checks for the aggregate stats |
| InternalDistributedSystem iDS = (InternalDistributedSystem) cache.getDistributedSystem(); |
| FunctionServiceStats functionServiceStats = |
| iDS.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| // functions are executed 3 times |
| noOfExecutionCalls_Aggregate += 3; |
| assertTrue( |
| functionServiceStats.getFunctionExecutionCalls() >= noOfExecutionCalls_Aggregate); |
| noOfExecutionsCompleted_Aggregate += 3; |
| assertTrue(functionServiceStats |
| .getFunctionExecutionsCompleted() >= noOfExecutionsCompleted_Aggregate); |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION2, iDS); |
| // TEST_FUNCTION2 is executed twice |
| noOfExecutionCalls_TESTFUNCTION2 += 2; |
| assertTrue(functionStats.getFunctionExecutionCalls() >= noOfExecutionCalls_TESTFUNCTION2); |
| noOfExecutionsCompleted_TESTFUNCTION2 += 2; |
| assertTrue(functionStats |
| .getFunctionExecutionsCompleted() >= noOfExecutionsCompleted_TESTFUNCTION2); |
| |
| functionStats = FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION3, iDS); |
| // TEST_FUNCTION3 is executed once |
| noOfExecutionCalls_TESTFUNCTION3 += 1; |
| assertTrue(functionStats.getFunctionExecutionCalls() >= noOfExecutionCalls_TESTFUNCTION3); |
| noOfExecutionsCompleted_TESTFUNCTION3 += 1; |
| assertTrue(functionStats |
| .getFunctionExecutionsCompleted() >= noOfExecutionsCompleted_TESTFUNCTION3); |
| }; |
| |
| server1.invoke(checkStatsOnServer); |
| server2.invoke(checkStatsOnServer); |
| server3.invoke(checkStatsOnServer); |
| } |
| |
| /** |
| * 1-client 3-Servers server1 : Replicate server2 : Replicate server3 : Replicate client : Empty |
| * Function : TEST_FUNCTION2 Execution of the function on serverRegion with set multiple keys as |
| * the routing object and using the name of the function |
| * |
| * On server side, function execution calls should be equal to the no of function executions |
| * completed. |
| */ |
| @Test |
| public void testClientServerDistributedRegionFunctionExecutionStats() { |
| |
| final String regionName = "FunctionServiceStatsDUnitTest"; |
| SerializableCallableIF<Integer> createCahenServer = () -> { |
| try { |
| Properties props = new Properties(); |
| DistributedSystem ds = getSystem(props); |
| assertNotNull(ds); |
| ds.disconnect(); |
| ds = getSystem(props); |
| cache = CacheFactory.create(ds); |
| logger.info("Created Cache on Server"); |
| assertNotNull(cache); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setDataPolicy(DataPolicy.REPLICATE); |
| assertNotNull(cache); |
| Region<String, Integer> region = cache.createRegion(regionName, factory.create()); |
| logger.info("Region Created :" + region); |
| assertNotNull(region); |
| for (int i = 1; i <= 200; i++) { |
| region.put("execKey-" + i, i); |
| } |
| CacheServer server = cache.addCacheServer(); |
| assertNotNull(server); |
| int port = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET); |
| server.setPort(port); |
| try { |
| server.start(); |
| } catch (IOException e) { |
| Assert.fail("Failed to start the Server", e); |
| } |
| assertTrue(server.isRunning()); |
| return server.getPort(); |
| } catch (Exception e) { |
| Assert.fail("FunctionServiceStatsDUnitTest#createCache() Failed while creating the cache", |
| e); |
| throw e; |
| } |
| }; |
| final Integer port1 = server1.invoke(createCahenServer); |
| final Integer port2 = server2.invoke(createCahenServer); |
| final Integer port3 = server3.invoke(createCahenServer); |
| |
| client.invoke(() -> { |
| try { |
| Properties props = new Properties(); |
| props.put(MCAST_PORT, "0"); |
| props.put(LOCATORS, ""); |
| DistributedSystem ds = getSystem(props); |
| assertNotNull(ds); |
| ds.disconnect(); |
| ds = getSystem(props); |
| cache = CacheFactory.create(ds); |
| logger.info("Created Cache on Client"); |
| assertNotNull(cache); |
| |
| |
| CacheServerTestUtil.disableShufflingOfEndpoints(); |
| Pool p; |
| try { |
| p = PoolManager.createFactory().addServer("localhost", port1) |
| .addServer("localhost", port2).addServer("localhost", port3) |
| .setPingInterval(250).setSubscriptionEnabled(false).setSubscriptionRedundancy(-1) |
| .setReadTimeout(2000).setSocketBufferSize(1000).setMinConnections(6) |
| .setMaxConnections(10).setRetryAttempts(3) |
| .create("FunctionServiceStatsDUnitTest_pool"); |
| } finally { |
| CacheServerTestUtil.enableShufflingOfEndpoints(); |
| } |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| factory.setDataPolicy(DataPolicy.EMPTY); |
| factory.setPoolName(p.getName()); |
| assertNotNull(cache); |
| Region<String, Integer> region = cache.createRegion(regionName, factory.create()); |
| logger.info("Client Region Created :" + region); |
| assertNotNull(region); |
| for (int i = 1; i <= 200; i++) { |
| region.put("execKey-" + i, i); |
| } |
| } catch (Exception e) { |
| Assert.fail("FunctionServiceStatsDUnitTest#createCache() Failed while creating the cache", |
| e); |
| throw e; |
| } |
| }); |
| |
| client.invoke(initializeStats); |
| server1.invoke(initializeStats); |
| server2.invoke(initializeStats); |
| server3.invoke(initializeStats); |
| |
| registerFunctionAtServer(new TestFunction(true, TestFunction.TEST_FUNCTION2)); |
| registerFunctionAtServer(new TestFunction(true, TestFunction.TEST_FUNCTION3)); |
| |
| client.invoke(() -> { |
| Function function2 = new TestFunction(true, TestFunction.TEST_FUNCTION2); |
| Function function3 = new TestFunction(true, TestFunction.TEST_FUNCTION3); |
| if (shouldRegisterFunctionsOnClient()) { |
| FunctionService.registerFunction(function2); |
| FunctionService.registerFunction(function3); |
| } |
| Region region = cache.getRegion(regionName); |
| Set<String> filter = new HashSet<>(); |
| for (int i = 100; i < 120; i++) { |
| filter.add("execKey-" + i); |
| } |
| |
| try { |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionCalls_TESTFUNCTION2++; |
| List list = (List) FunctionService.onRegion(region).withFilter(filter) |
| .execute(function2).getResult(); |
| noOfExecutionsCompleted_Aggregate++; |
| noOfExecutionsCompleted_TESTFUNCTION2++; |
| int size = list.size(); |
| resultReceived_Aggregate += size; |
| resultReceived_TESTFUNCTION2 += size; |
| |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionCalls_TESTFUNCTION2++; |
| list = (List) FunctionService.onRegion(region).withFilter(filter).execute(function2) |
| .getResult(); |
| noOfExecutionsCompleted_Aggregate++; |
| noOfExecutionsCompleted_TESTFUNCTION2++; |
| size = list.size(); |
| resultReceived_Aggregate += size; |
| resultReceived_TESTFUNCTION2 += size; |
| } catch (Exception e) { |
| e.printStackTrace(); |
| Assert.fail("test failed due to", e); |
| throw e; |
| } |
| |
| }); |
| |
| client.invoke(() -> { |
| // checks for the aggregate stats |
| InternalDistributedSystem iDS = (InternalDistributedSystem) cache.getDistributedSystem(); |
| FunctionServiceStats functionServiceStats = |
| iDS.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_Aggregate, functionServiceStats.getResultsReceived()); |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION2, iDS); |
| assertEquals(noOfExecutionCalls_TESTFUNCTION2, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION2, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_TESTFUNCTION2, functionStats.getResultsReceived()); |
| }); |
| } |
| |
| /** |
| * Execution of the function on server using the name of the function TEST_FUNCTION1 |
| * TEST_FUNCTION5 On client side, the no of result received should equal to the no of function |
| * execution calls. On server side, function execution calls should be equal to the no of function |
| * executions completed. |
| */ |
| @Test |
| public void testClientServerwithoutRegion() { |
| createClientServerScenarionWithoutRegion(); |
| registerFunctionAtServer(new TestFunction(true, TestFunction.TEST_FUNCTION1)); |
| registerFunctionAtServer(new TestFunction(true, TestFunction.TEST_FUNCTION5)); |
| |
| client.invoke(initializeStats); |
| server1.invoke(initializeStats); |
| server2.invoke(initializeStats); |
| server3.invoke(initializeStats); |
| |
| client.invoke(() -> { |
| DistributedSystem.setThreadsSocketPolicy(false); |
| Function function = new TestFunction(true, TestFunction.TEST_FUNCTION1); |
| if (shouldRegisterFunctionsOnClient()) { |
| FunctionService.registerFunction(function); |
| } |
| Execution member = FunctionService.onServers(pool); |
| |
| try { |
| ResultCollector rs = member.setArguments(Boolean.TRUE).execute(function.getId()); |
| int size = ((List) rs.getResult()).size(); |
| resultReceived_Aggregate += size; |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionsCompleted_Aggregate++; |
| resultReceived_TESTFUNCTION1 += size; |
| noOfExecutionCalls_TESTFUNCTION1++; |
| noOfExecutionsCompleted_TESTFUNCTION1++; |
| } catch (Exception ex) { |
| ex.printStackTrace(); |
| logger.info("Exception : ", ex); |
| fail("Test failed after the execute operation nn TRUE"); |
| } |
| function = new TestFunction(true, TestFunction.TEST_FUNCTION5); |
| if (shouldRegisterFunctionsOnClient()) { |
| FunctionService.registerFunction(function); |
| } |
| try { |
| ResultCollector rs = member.setArguments("Success").execute(function.getId()); |
| int size = ((List) rs.getResult()).size(); |
| resultReceived_Aggregate += size; |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionsCompleted_Aggregate++; |
| resultReceived_TESTFUNCTION5 += size; |
| noOfExecutionCalls_TESTFUNCTION5++; |
| noOfExecutionsCompleted_TESTFUNCTION5++; |
| } catch (Exception ex) { |
| ex.printStackTrace(); |
| logger.info("Exception : ", ex); |
| fail("Test failed after the execute operationssssss"); |
| } |
| }); |
| |
| |
| client.invoke(() -> { |
| // checks for the aggregate stats |
| InternalDistributedSystem iDS = (InternalDistributedSystem) cache.getDistributedSystem(); |
| FunctionServiceStats functionServiceStats = |
| iDS.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_Aggregate, functionServiceStats.getResultsReceived()); |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION1, iDS); |
| assertEquals(noOfExecutionCalls_TESTFUNCTION1, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION1, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_TESTFUNCTION1, functionStats.getResultsReceived()); |
| |
| functionStats = FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION5, iDS); |
| assertEquals(noOfExecutionCalls_TESTFUNCTION5, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION5, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_TESTFUNCTION5, functionStats.getResultsReceived()); |
| |
| }); |
| |
| SerializableRunnableIF checkStatsOnServer = () -> { |
| // checks for the aggregate stats |
| InternalDistributedSystem iDS = (InternalDistributedSystem) cache.getDistributedSystem(); |
| FunctionServiceStats functionServiceStats = |
| iDS.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| // functions are executed 2 times |
| noOfExecutionCalls_Aggregate += 2; |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| noOfExecutionsCompleted_Aggregate += 2; |
| // this check is time sensitive, so allow it to fail a few times |
| // before giving up |
| for (int i = 0; i < 10; i++) { |
| try { |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| } catch (RuntimeException r) { |
| if (i == 9) { |
| throw r; |
| } |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| throw r; |
| } |
| } |
| } |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION1, iDS); |
| // TEST_FUNCTION1 is executed once |
| noOfExecutionCalls_TESTFUNCTION1 += 1; |
| assertEquals(noOfExecutionCalls_TESTFUNCTION1, functionStats.getFunctionExecutionCalls()); |
| noOfExecutionsCompleted_TESTFUNCTION1 += 1; |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION1, |
| functionStats.getFunctionExecutionsCompleted()); |
| |
| functionStats = FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION5, iDS); |
| // TEST_FUNCTION5 is executed once |
| noOfExecutionCalls_TESTFUNCTION5 += 1; |
| assertEquals(noOfExecutionCalls_TESTFUNCTION5, functionStats.getFunctionExecutionCalls()); |
| noOfExecutionsCompleted_TESTFUNCTION5 += 1; |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION5, |
| functionStats.getFunctionExecutionsCompleted()); |
| |
| }; |
| |
| server1.invoke(checkStatsOnServer); |
| server2.invoke(checkStatsOnServer); |
| server3.invoke(checkStatsOnServer); |
| } |
| |
| @Test |
| public void testP2PDummyExecutionStats() { |
| Host host = Host.getHost(0); |
| final VM datastore0 = host.getVM(0); |
| final VM datastore1 = host.getVM(1); |
| final VM datastore2 = host.getVM(2); |
| final VM accessor = host.getVM(3); |
| accessor.invoke(closeDistributedSystem); |
| datastore0.invoke(closeDistributedSystem); |
| datastore1.invoke(closeDistributedSystem); |
| datastore2.invoke(closeDistributedSystem); |
| } |
| |
| /** |
| * Ensure that the execution is happening all the PR as a whole |
| * |
| * Function Execution will not take place on accessor, accessor will onlu receive the |
| * resultsReceived. On datastore, no of function execution calls should be equal to the no of |
| * function execution calls from the accessor. |
| */ |
| @Test |
| public void testP2PPartitionedRegionsFunctionExecutionStats() { |
| final String rName = getUniqueName(); |
| Host host = Host.getHost(0); |
| final VM datastore0 = host.getVM(0); |
| final VM datastore1 = host.getVM(1); |
| final VM datastore2 = host.getVM(2); |
| final VM accessor = host.getVM(3); |
| |
| datastore0.invoke(initializeStats); |
| datastore1.invoke(initializeStats); |
| datastore2.invoke(initializeStats); |
| accessor.invoke(initializeStats); |
| |
| accessor.invoke(() -> { |
| RegionAttributes ra = PartitionedRegionTestHelper.createRegionAttrsForPR(0, 0); |
| AttributesFactory raf = new AttributesFactory(ra); |
| PartitionAttributesImpl pa = new PartitionAttributesImpl(); |
| pa.setAll(ra.getPartitionAttributes()); |
| pa.setTotalNumBuckets(17); |
| raf.setPartitionAttributes(pa); |
| |
| getCache().createRegion(rName, raf.create()); |
| }); |
| |
| SerializableRunnableIF dataStoreCreate = () -> { |
| RegionAttributes ra = PartitionedRegionTestHelper.createRegionAttrsForPR(0, 10); |
| AttributesFactory raf = new AttributesFactory(ra); |
| PartitionAttributesImpl pa = new PartitionAttributesImpl(); |
| pa.setAll(ra.getPartitionAttributes()); |
| pa.setTotalNumBuckets(17); |
| raf.setPartitionAttributes(pa); |
| getCache().createRegion(rName, raf.create()); |
| Function function = new TestFunction(true, TestFunction.TEST_FUNCTION2); |
| FunctionService.registerFunction(function); |
| function = new TestFunction(true, TestFunction.TEST_FUNCTION3); |
| FunctionService.registerFunction(function); |
| }; |
| datastore0.invoke(dataStoreCreate); |
| datastore1.invoke(dataStoreCreate); |
| datastore2.invoke(dataStoreCreate); |
| |
| accessor.invoke(() -> { |
| PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(rName); |
| DistributedSystem.setThreadsSocketPolicy(false); |
| final HashSet<String> testKeys = new HashSet<>(); |
| for (int i = (pr.getTotalNumberOfBuckets() * 3); i > 0; i--) { |
| testKeys.add("execKey-" + i); |
| } |
| int j = 0; |
| for (Object testKey : testKeys) { |
| Integer val = j++; |
| pr.put(testKey, val); |
| } |
| Function function = new TestFunction(true, TestFunction.TEST_FUNCTION2); |
| FunctionService.registerFunction(function); |
| Execution dataSet = FunctionService.onRegion(pr); |
| ResultCollector rc1 = dataSet.setArguments(Boolean.TRUE).execute(function); |
| int size = ((List) rc1.getResult()).size(); |
| resultReceived_Aggregate += size; |
| resultReceived_TESTFUNCTION2 += size; |
| |
| rc1 = dataSet.setArguments(testKeys).execute(function); |
| size = ((List) rc1.getResult()).size(); |
| resultReceived_Aggregate += size; |
| resultReceived_TESTFUNCTION2 += size; |
| |
| function = new TestFunction(true, TestFunction.TEST_FUNCTION3); |
| FunctionService.registerFunction(function); |
| rc1 = dataSet.setArguments(Boolean.TRUE).execute(function); |
| size = ((List) rc1.getResult()).size(); |
| resultReceived_Aggregate += size; |
| resultReceived_TESTFUNCTION3 += size; |
| }); |
| |
| accessor.invoke(() -> { |
| InternalDistributedSystem iDS = |
| ((InternalDistributedSystem) getCache().getDistributedSystem()); |
| FunctionServiceStats functionServiceStats = |
| iDS.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_Aggregate, functionServiceStats.getResultsReceived()); |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION2, iDS); |
| assertEquals(noOfExecutionCalls_TESTFUNCTION2, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION2, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_TESTFUNCTION2, functionStats.getResultsReceived()); |
| |
| functionStats = FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION3, iDS); |
| assertEquals(noOfExecutionCalls_TESTFUNCTION3, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION3, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_TESTFUNCTION3, functionStats.getResultsReceived()); |
| }); |
| |
| SerializableRunnableIF checkFunctionExecutionStatsForDataStore = () -> { |
| InternalDistributedSystem iDS = |
| ((InternalDistributedSystem) getCache().getDistributedSystem()); |
| // 3 Function Executions took place |
| FunctionServiceStats functionServiceStats = |
| iDS.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| noOfExecutionCalls_Aggregate += 3; |
| noOfExecutionsCompleted_Aggregate += 3; |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION2, iDS); |
| // TEST_FUNCTION2 is executed twice |
| noOfExecutionCalls_TESTFUNCTION2 += 2; |
| assertEquals(noOfExecutionCalls_TESTFUNCTION2, |
| functionStats.getFunctionExecutionCalls()); |
| noOfExecutionsCompleted_TESTFUNCTION2 += 2; |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION2, |
| functionStats.getFunctionExecutionsCompleted()); |
| |
| functionStats = FunctionStatsManager.getFunctionStats(TestFunction.TEST_FUNCTION3, iDS); |
| // TEST_FUNCTION3 is executed once |
| noOfExecutionCalls_TESTFUNCTION3 += 1; |
| assertEquals(noOfExecutionCalls_TESTFUNCTION3, |
| functionStats.getFunctionExecutionCalls()); |
| noOfExecutionsCompleted_TESTFUNCTION3 += 1; |
| assertEquals(noOfExecutionsCompleted_TESTFUNCTION3, |
| functionStats.getFunctionExecutionsCompleted()); |
| }; |
| datastore0.invoke(checkFunctionExecutionStatsForDataStore); |
| datastore1.invoke(checkFunctionExecutionStatsForDataStore); |
| datastore2.invoke(checkFunctionExecutionStatsForDataStore); |
| |
| accessor.invoke(closeDistributedSystem); |
| datastore0.invoke(closeDistributedSystem); |
| datastore1.invoke(closeDistributedSystem); |
| datastore2.invoke(closeDistributedSystem); |
| } |
| |
| /** |
| * Test the function execution statistics in case of the distributed Region P2P DataStore0 is with |
| * Empty datapolicy |
| */ |
| @Test |
| public void testP2PDistributedRegionFunctionExecutionStats() { |
| final String rName = getUniqueName(); |
| Host host = Host.getHost(0); |
| final VM datastore0 = host.getVM(0); |
| final VM datastore1 = host.getVM(1); |
| final VM datastore2 = host.getVM(2); |
| final VM datastore3 = host.getVM(3); |
| |
| datastore0.invoke(initializeStats); |
| datastore1.invoke(initializeStats); |
| datastore2.invoke(initializeStats); |
| datastore3.invoke(initializeStats); |
| |
| datastore0.invoke(() -> { |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setDataPolicy(DataPolicy.EMPTY); |
| Region<String, Integer> region = getCache().createRegion(rName, factory.create()); |
| logger.info("Region Created :" + region); |
| assertNotNull(region); |
| FunctionService.registerFunction(new TestFunction(true, TestFunction.TEST_FUNCTION2)); |
| for (int i = 1; i <= 200; i++) { |
| region.put("execKey-" + i, i); |
| } |
| }); |
| |
| SerializableRunnableIF createAndPopulateRegionWithReplicate = () -> { |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setDataPolicy(DataPolicy.REPLICATE); |
| Region<String, Integer> region = getCache().createRegion(rName, factory.create()); |
| logger.info("Region Created :" + region); |
| assertNotNull(region); |
| FunctionService.registerFunction(new TestFunction(true, TestFunction.TEST_FUNCTION2)); |
| for (int i = 1; i <= 200; i++) { |
| region.put("execKey-" + i, i); |
| } |
| }; |
| |
| datastore1.invoke(createAndPopulateRegionWithReplicate); |
| datastore2.invoke(createAndPopulateRegionWithReplicate); |
| datastore3.invoke(createAndPopulateRegionWithReplicate); |
| |
| datastore0.invoke(() -> { |
| Region region = getCache().getRegion(rName); |
| try { |
| List list = (List) FunctionService.onRegion(region).setArguments(Boolean.TRUE) |
| .execute(TestFunction.TEST_FUNCTION2).getResult(); |
| // this is the Distributed Region with Empty Data policy. |
| // therefore no function execution takes place here. it only receives the results. |
| resultReceived_Aggregate += list.size(); |
| assertEquals(resultReceived_Aggregate, |
| ((InternalDistributedSystem) getCache().getDistributedSystem()) |
| .getFunctionStatsManager().getFunctionServiceStats().getResultsReceived()); |
| |
| resultReceived_TESTFUNCTION2 += list.size(); |
| assertEquals(resultReceived_TESTFUNCTION2, |
| ((InternalDistributedSystem) getCache().getDistributedSystem()) |
| .getFunctionStatsManager().getFunctionServiceStats().getResultsReceived()); |
| |
| } catch (Exception e) { |
| e.printStackTrace(); |
| Assert.fail("test failed due to", e); |
| } |
| }); |
| |
| // there is a replicated region on 3 nodes so we cannot predict on which |
| // node the function execution will take place |
| // so i have avoided that check. |
| |
| datastore0.invoke(closeDistributedSystem); |
| datastore1.invoke(closeDistributedSystem); |
| datastore2.invoke(closeDistributedSystem); |
| datastore3.invoke(closeDistributedSystem); |
| } |
| |
| |
| /** |
| * Test the execution of function on all memebers haveResults = true |
| * |
| * member1 calls for the function executions sp the results received on memeber 1 should be equal |
| * to the no of function execution calls. Function Execution should happen on all other members |
| * too. so the no of function execution calls and no of function executions completed should be |
| * equal tio the no of functions from member 1 |
| */ |
| @Test |
| public void testP2PMembersFunctionExecutionStats() { |
| Host host = Host.getHost(0); |
| VM member1 = host.getVM(0); |
| VM member2 = host.getVM(1); |
| VM member3 = host.getVM(2); |
| VM member4 = host.getVM(3); |
| |
| SerializableRunnableIF connectToDistributedSystem = () -> { |
| Properties props = new Properties(); |
| try { |
| ds = getSystem(props); |
| assertNotNull(ds); |
| } catch (Exception e) { |
| Assert.fail("Failed while creating the Distribued System", e); |
| } |
| }; |
| member1.invoke(connectToDistributedSystem); |
| member2.invoke(connectToDistributedSystem); |
| member3.invoke(connectToDistributedSystem); |
| member4.invoke(connectToDistributedSystem); |
| |
| member1.invoke(initializeStats); |
| member2.invoke(initializeStats); |
| member3.invoke(initializeStats); |
| member4.invoke(initializeStats); |
| |
| final Function inlineFunction = new FunctionAdapter() { |
| @Override |
| public void execute(FunctionContext context) { |
| @SuppressWarnings("unchecked") |
| final ResultSender<String> resultSender = context.getResultSender(); |
| if (context.getArguments() instanceof String) { |
| resultSender.lastResult("Success"); |
| } else { |
| resultSender.lastResult("Failure"); |
| } |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getName(); |
| } |
| |
| @Override |
| public boolean hasResult() { |
| return true; |
| } |
| }; |
| |
| member1.invoke(() -> { |
| |
| assertNotNull(ds); |
| DistributedMember localmember = ds.getDistributedMember(); |
| Execution memberExecution = FunctionService.onMember(localmember); |
| |
| memberExecution.setArguments("Key"); |
| try { |
| ResultCollector rc = memberExecution.execute(inlineFunction); |
| int size = ((List) rc.getResult()).size(); |
| resultReceived_Aggregate += size; |
| noOfExecutionCalls_Aggregate++; |
| noOfExecutionsCompleted_Aggregate++; |
| resultReceived_Inline += size; |
| noOfExecutionCalls_Inline++; |
| noOfExecutionsCompleted_Inline++; |
| |
| } catch (Exception e) { |
| logger.info("Exception Occurred : " + e.getMessage()); |
| e.printStackTrace(); |
| Assert.fail("Test failed", e); |
| } |
| }); |
| |
| member1.invoke(() -> { |
| FunctionServiceStats functionServiceStats = |
| ds.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_Aggregate, functionServiceStats.getResultsReceived()); |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(inlineFunction.getId(), ds); |
| assertEquals(noOfExecutionCalls_Inline, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Inline, |
| functionStats.getFunctionExecutionsCompleted()); |
| assertEquals(resultReceived_Inline, functionStats.getResultsReceived()); |
| }); |
| |
| SerializableRunnableIF checkFunctionExecutionStatsForOtherMember = () -> { |
| FunctionServiceStats functionServiceStats = |
| ds.getFunctionStatsManager().getFunctionServiceStats(); |
| waitNoFunctionsRunning(functionServiceStats); |
| |
| // One function Execution took place on there members |
| // noOfExecutionCalls_Aggregate++; |
| // noOfExecutionsCompleted_Aggregate++; |
| assertEquals(noOfExecutionCalls_Aggregate, |
| functionServiceStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Aggregate, |
| functionServiceStats.getFunctionExecutionsCompleted()); |
| |
| FunctionStats functionStats = |
| FunctionStatsManager.getFunctionStats(inlineFunction.getId(), ds); |
| // noOfExecutionCalls_Inline++; |
| // noOfExecutionsCompleted_Inline++; |
| assertEquals(noOfExecutionCalls_Inline, functionStats.getFunctionExecutionCalls()); |
| assertEquals(noOfExecutionsCompleted_Inline, |
| functionStats.getFunctionExecutionsCompleted()); |
| }; |
| member2.invoke(checkFunctionExecutionStatsForOtherMember); |
| member3.invoke(checkFunctionExecutionStatsForOtherMember); |
| member4.invoke(checkFunctionExecutionStatsForOtherMember); |
| |
| member1.invoke(closeDistributedSystem); |
| member2.invoke(closeDistributedSystem); |
| member3.invoke(closeDistributedSystem); |
| member4.invoke(closeDistributedSystem); |
| } |
| |
| @Override |
| public Properties getDistributedSystemProperties() { |
| Properties properties = super.getDistributedSystemProperties(); |
| properties.put(ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER, |
| "org.apache.geode.internal.cache.execute.MyFunctionExecutionException"); |
| return properties; |
| } |
| |
| /** |
| * Test the exception occurred while invoking the function execution on all members of DS |
| * |
| * Function throws the Exception, The check is added to for the no of function execution execption |
| * in datatostore1 |
| */ |
| @Test |
| public void testFunctionExecutionExceptionStatsOnAllNodesPRegion() { |
| final String rName = getUniqueName(); |
| Host host = Host.getHost(0); |
| final VM datastore0 = host.getVM(0); |
| final VM datastore1 = host.getVM(1); |
| final VM datastore2 = host.getVM(2); |
| final VM datastore3 = host.getVM(3); |
| |
| datastore0.invoke(initializeStats); |
| datastore1.invoke(initializeStats); |
| datastore2.invoke(initializeStats); |
| datastore3.invoke(initializeStats); |
| |
| SerializableRunnableIF dataStoreCreate = () -> { |
| RegionAttributes ra = PartitionedRegionTestHelper.createRegionAttrsForPR(0, 10); |
| AttributesFactory raf = new AttributesFactory(ra); |
| PartitionAttributesImpl pa = new PartitionAttributesImpl(); |
| pa.setAll(ra.getPartitionAttributes()); |
| pa.setTotalNumBuckets(17); |
| raf.setPartitionAttributes(pa); |
| getCache().createRegion(rName, raf.create()); |
| Function function = new TestFunction(true, "TestFunctionException"); |
| FunctionService.registerFunction(function); |
| }; |
| datastore0.invoke(dataStoreCreate); |
| datastore1.invoke(dataStoreCreate); |
| datastore2.invoke(dataStoreCreate); |
| datastore3.invoke(dataStoreCreate); |
| |
| datastore3.invoke(() -> { |
| PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(rName); |
| DistributedSystem.setThreadsSocketPolicy(false); |
| final HashSet<String> testKeys = new HashSet<>(); |
| for (int i = (pr.getTotalNumberOfBuckets() * 3); i > 0; i--) { |
| testKeys.add("execKey-" + i); |
| } |
| int j = 0; |
| for (Object testKey : testKeys) { |
| Integer key = j++; |
| pr.put(key, testKey); |
| } |
| try { |
| Function function = new TestFunction(true, "TestFunctionException"); |
| FunctionService.registerFunction(function); |
| Execution dataSet = FunctionService.onRegion(pr); |
| ResultCollector rc = dataSet.setArguments(Boolean.TRUE).execute(function.getId()); |
| // Wait Criterion is added to make sure that the function execution |
| // happens on all nodes and all nodes get the FunctionException so that the stats will get |
| // incremented, |
| Wait.pause(2000); |
| rc.getResult(); |
| fail("No exception Occurred"); |
| } catch (Exception ignored) { |
| } |
| }); |
| |
| datastore0.invoke(closeDistributedSystem); |
| datastore1.invoke(closeDistributedSystem); |
| datastore2.invoke(closeDistributedSystem); |
| datastore3.invoke(closeDistributedSystem); |
| } |
| |
| private void createScenario() { |
| ArrayList commonAttributes = |
| createCommonServerAttributes("TestPartitionedRegion", null, 0, null); |
| createClientServerScenarion(commonAttributes, 20, 20, 20); |
| } |
| } |
