| /* |
| * 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 java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static java.util.stream.Collectors.toList; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.distributed.ConfigurationProperties.NAME; |
| import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTHENTICATOR; |
| import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT; |
| import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout; |
| import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException; |
| import static org.apache.geode.test.dunit.VM.getController; |
| import static org.apache.geode.test.dunit.VM.getVM; |
| import static org.apache.geode.test.dunit.VM.toArray; |
| import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX; |
| import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast; |
| import static org.assertj.core.api.Assertions.assertThat; |
| import static org.assertj.core.api.Assertions.catchThrowable; |
| |
| import java.io.IOException; |
| import java.io.Serializable; |
| import java.util.Comparator; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.stream.IntStream; |
| |
| import org.junit.After; |
| import org.junit.Before; |
| import org.junit.Rule; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.cache.CacheClosedException; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.client.ClientRegionShortcut; |
| import org.apache.geode.cache.client.Pool; |
| import org.apache.geode.cache.client.PoolFactory; |
| import org.apache.geode.cache.client.PoolManager; |
| import org.apache.geode.cache.client.internal.InternalClientCache; |
| import org.apache.geode.cache.execute.Function; |
| import org.apache.geode.cache.execute.FunctionContext; |
| import org.apache.geode.cache.execute.FunctionException; |
| import org.apache.geode.cache.execute.FunctionInvocationTargetException; |
| import org.apache.geode.cache.execute.FunctionService; |
| import org.apache.geode.cache.execute.RegionFunctionContext; |
| import org.apache.geode.cache.execute.ResultCollector; |
| import org.apache.geode.cache.server.CacheServer; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.execute.util.TypedFunctionService; |
| import org.apache.geode.security.templates.DummyAuthenticator; |
| import org.apache.geode.security.templates.UserPasswordAuthInit; |
| import org.apache.geode.test.dunit.AsyncInvocation; |
| import org.apache.geode.test.dunit.IgnoredException; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.rules.CacheRule; |
| import org.apache.geode.test.dunit.rules.ClientCacheRule; |
| import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties; |
| import org.apache.geode.test.dunit.rules.DistributedRule; |
| import org.apache.geode.test.junit.categories.FunctionServiceTest; |
| |
| @Category(FunctionServiceTest.class) |
| @SuppressWarnings("serial") |
| public class DistributedRegionFunctionExecutionDUnitTest implements Serializable { |
| |
| private static final AtomicReference<Region<?, ?>> REGION = new AtomicReference<>(); |
| private static final AtomicReference<CountDownLatch> LATCH = new AtomicReference<>(); |
| |
| private final Set<String> filter = new HashSet<>(); |
| |
| private String regionName; |
| private String poolName; |
| |
| private VM empty; |
| private VM normal; |
| private VM replicate1; |
| private VM replicate2; |
| private VM replicate3; |
| |
| @Rule |
| public CacheRule cacheRule = new CacheRule(); |
| |
| @Rule |
| public ClientCacheRule clientCacheRule = new ClientCacheRule(); |
| |
| @Rule |
| public DistributedRestoreSystemProperties restoreSystemProperties = |
| new DistributedRestoreSystemProperties(); |
| |
| @Before |
| public void setUp() throws Exception { |
| empty = getController(); |
| replicate1 = getVM(0); |
| replicate2 = getVM(1); |
| replicate3 = getVM(2); |
| normal = getVM(3); |
| |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| // zero is no-op unless test replaces the reference |
| vm.invoke(() -> LATCH.set(new CountDownLatch(0))); |
| } |
| |
| regionName = getClass().getSimpleName() + "_region"; |
| poolName = getClass().getSimpleName() + "_pool"; |
| |
| for (int i = 100; i < 120; i++) { |
| filter.add("execKey-" + i); |
| } |
| } |
| |
| @After |
| public void tearDown() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> LATCH.get().countDown()); |
| } |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> { |
| populateRegion(200); |
| |
| executeDistributedRegionFunction(); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_SendException() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> { |
| populateRegion(200); |
| |
| executeResultWithExceptionFunction(); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_NoLastResult() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> { |
| populateRegion(200); |
| |
| executeNoLastResultFunction(); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyNormal() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| normal.invoke(() -> { |
| Throwable thrown = catchThrowable(() -> executeDistributedRegionFunction()); |
| |
| assertThat(thrown) |
| .isInstanceOf(FunctionException.class) |
| .hasMessage("Function execution on region with DataPolicy.NORMAL is not supported"); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| replicate1.invoke(() -> executeDistributedRegionFunction()); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicateNotTimedOut() |
| throws Exception { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> { |
| createCache(getDistributedSystemProperties()); |
| |
| LATCH.set(new CountDownLatch(1)); |
| |
| FunctionService.registerFunction(new DistributedRegionFunction()); |
| FunctionService.registerFunction(new LongRunningFunction(LATCH.get())); |
| }); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| AsyncInvocation executeFunctionInReplicate1 = replicate1.invokeAsync(() -> { |
| ResultCollector<String, List<String>> resultCollector = TypedFunctionService |
| .<Void, String, List<String>>onRegion(getRegion()) |
| .withFilter(filter) |
| .execute(LongRunningFunction.class.getSimpleName(), getTimeout().toMillis(), |
| MILLISECONDS); |
| |
| assertThat(resultCollector.getResult().get(0)) |
| .isEqualTo("LongRunningFunction completed"); |
| }); |
| |
| // how long LongRunningFunction runs for is now controlled here: |
| Thread.sleep(2000); |
| |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> { |
| LATCH.get().countDown(); |
| }); |
| } |
| |
| executeFunctionInReplicate1.await(); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicateTimedOut() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> { |
| createCache(getDistributedSystemProperties()); |
| |
| LATCH.set(new CountDownLatch(1)); |
| |
| FunctionService.registerFunction(new DistributedRegionFunction()); |
| FunctionService.registerFunction(new LongRunningFunction(LATCH.get())); |
| }); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| replicate1.invoke(() -> { |
| Throwable thrown = catchThrowable(() -> { |
| TypedFunctionService |
| .<Void, String, List<String>>onRegion(getRegion()) |
| .withFilter(filter) |
| .execute(LongRunningFunction.class.getSimpleName(), 1000, MILLISECONDS); |
| }); |
| |
| assertThat(thrown) |
| .hasCauseInstanceOf(FunctionException.class) |
| .hasMessageContaining("All results not received in time provided"); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_SendException() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| replicate1.invoke(() -> executeResultWithExceptionFunction()); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_NoLastResult() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| replicate1.invoke(() -> executeNoLastResultFunction()); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetException() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| replicate1.invoke(() -> { |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetException(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetException_WithoutHA() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| replicate1.invoke(() -> { |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetExceptionWithoutHA(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetExceptionForEmptyDataPolicy() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| empty.invoke(() -> { |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetException(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetExceptionForEmptyDataPolicy_WithoutHA() { |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| for (VM vm : toArray(replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0)); |
| } |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| empty.invoke(() -> { |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetExceptionWithoutHA(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionHACacheClosedException() { |
| for (VM vm : toArray(empty, replicate1)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| replicate1.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| replicate2.invoke(() -> { |
| createCache(getDistributedSystemProperties()); |
| createRegion(DataPolicy.REPLICATE); |
| }); |
| |
| empty.invoke(() -> { |
| List<Boolean> result = executeDistributedRegionFunction(); |
| |
| assertThat(result) |
| .hasSize(5001) |
| .containsOnly(true); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionHANodeFailure() throws Exception { |
| for (VM vm : toArray(empty, replicate1)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| replicate1.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| |
| empty.invoke(() -> populateRegion(200)); |
| |
| AsyncInvocation<List<Boolean>> executeFunctionHaInEmptyVM = |
| empty.invokeAsync(() -> executeDistributedRegionFunction()); |
| |
| replicate2.invoke(() -> { |
| createCache(getDistributedSystemProperties()); |
| createRegion(DataPolicy.REPLICATE); |
| }); |
| |
| replicate1.invoke(() -> getCache().close()); |
| |
| List<Boolean> result = executeFunctionHaInEmptyVM.get(); |
| |
| assertThat(result) |
| .hasSize(5001) |
| .containsOnly(true); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer() { |
| VM empty1 = replicate3; |
| VM empty2 = normal; |
| VM client = empty; |
| |
| for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| client.invoke(() -> createClientCache()); |
| |
| int port1 = empty1.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| int port2 = empty2.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| |
| for (VM vm : toArray(replicate1, replicate2)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| executeDistributedRegionFunction(); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_SendException() { |
| VM empty1 = replicate3; |
| VM empty2 = normal; |
| VM client = empty; |
| |
| for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| client.invoke(() -> createClientCache()); |
| |
| int port1 = empty1.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| int port2 = empty2.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| |
| for (VM vm : toArray(replicate1, replicate2)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| executeResultWithExceptionFunction(); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_NoLastResult() { |
| VM empty1 = replicate3; |
| VM empty2 = normal; |
| VM client = empty; |
| |
| for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| client.invoke(() -> createClientCache()); |
| |
| int port1 = empty1.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| int port2 = empty2.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| |
| for (VM vm : toArray(replicate1, replicate2)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| try (IgnoredException ie1 = addIgnoredException("did not send last result", empty1); |
| IgnoredException ie2 = addIgnoredException("did not send last result", empty2)) { |
| executeNoLastResultFunction(); |
| } |
| }); |
| } |
| |
| /** |
| * If one server goes down while executing a function, that function should failover to other |
| * available server. |
| */ |
| @Test |
| public void testServerFailoverWithTwoServerAliveHA() throws InterruptedException { |
| VM emptyServer1 = replicate1; |
| VM client = normal; |
| |
| for (VM vm : toArray(emptyServer1, replicate2, replicate3)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| int port1 = emptyServer1.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| int port2 = replicate2.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| replicate3.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| |
| client.invoke(() -> { |
| LATCH.set(new CountDownLatch(1)); |
| createClientCache(); |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| }); |
| |
| replicate2.invoke(() -> stopServerHA()); |
| |
| AsyncInvocation<List<Boolean>> executeFunctionHaInClientVm = |
| client.invokeAsync(() -> executeDistributedRegionFunction()); |
| |
| replicate2.invoke(() -> { |
| startServerHA(); |
| client.invoke(() -> LATCH.get().countDown()); |
| }); |
| |
| client.invoke(() -> { |
| LATCH.get().await(getTimeout().toMillis(), MILLISECONDS); |
| }); |
| |
| emptyServer1.invoke(() -> closeCacheHA()); |
| |
| List<Boolean> result = executeFunctionHaInClientVm.get(); |
| |
| assertThat(result) |
| .hasSize(5001) |
| .containsOnly(true); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyNormal_ClientServer() { |
| VM client = empty; |
| VM normal1 = normal; |
| VM normal2 = replicate3; |
| VM empty = replicate2; |
| |
| for (VM vm : toArray(normal1, replicate1, empty, normal2)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| client.invoke(() -> createClientCache()); |
| |
| int port1 = normal1.invoke(() -> { |
| createRegion(DataPolicy.NORMAL); |
| return createCacheServer(); |
| }); |
| int port2 = normal2.invoke(() -> { |
| createRegion(DataPolicy.NORMAL); |
| return createCacheServer(); |
| }); |
| |
| for (VM vm : toArray(empty, replicate1)) { |
| vm.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| // add expected exception |
| try (IgnoredException ie = addIgnoredException(FunctionException.class)) { |
| Throwable thrown = catchThrowable(() -> executeDistributedRegionFunction()); |
| |
| assertThat(thrown) |
| .isInstanceOf(FunctionException.class) |
| .hasMessageContaining( |
| "Function execution on region with DataPolicy.NORMAL is not supported"); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer() { |
| VM client = empty; |
| VM empty = replicate3; |
| |
| for (VM vm : toArray(normal, replicate1, replicate2, empty)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| client.invoke(() -> createClientCache()); |
| |
| int port1 = replicate1.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| int port2 = replicate2.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| executeUnregisteredFunction(); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer_WithoutRegister() { |
| VM client = empty; |
| VM empty = replicate3; |
| |
| for (VM vm : toArray(normal, replicate1, replicate2, empty)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| client.invoke(() -> createClientCache()); |
| |
| int port1 = replicate1.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| int port2 = replicate2.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| executeDistributedRegionFunction(); |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer_FunctionInvocationTargetException() { |
| VM client = empty; |
| VM empty = replicate3; |
| |
| for (VM vm : toArray(normal, replicate1, replicate2, empty)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| int port1 = replicate1.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| int port2 = replicate2.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| client.invoke(() -> createClientCache()); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| |
| for (VM vm : toArray(client, replicate1, replicate2, normal, empty)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| // add expected exception to avoid suspect strings |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetException_ClientServer(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer_FunctionInvocationTargetException_WithoutHA() { |
| VM client = empty; |
| VM empty = replicate3; |
| |
| for (VM vm : toArray(normal, replicate1, replicate2, empty)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| int port1 = replicate1.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| int port2 = replicate2.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| client.invoke(() -> createClientCache()); |
| normal.invoke(() -> createRegion(DataPolicy.NORMAL)); |
| empty.invoke(() -> createRegion(DataPolicy.EMPTY)); |
| |
| for (VM vm : toArray(client, replicate1, replicate2, normal, empty)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| // add expected exception to avoid suspect strings |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetException_ClientServer_WithoutHA(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_FunctionInvocationTargetException() { |
| VM client = empty; |
| VM empty1 = replicate3; |
| VM empty2 = normal; |
| |
| for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| int port1 = empty1.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| int port2 = empty2.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| |
| client.invoke(() -> createClientCache()); |
| for (VM vm : toArray(replicate1, replicate2)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| }); |
| |
| for (VM vm : toArray(client, empty1, empty2, replicate1, replicate2)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5)); |
| } |
| |
| client.invoke(() -> { |
| // add expected exception to avoid suspect strings |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetException_ClientServer(); |
| } |
| }); |
| } |
| |
| @Test |
| public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_FunctionInvocationTargetException_WithoutHA() { |
| VM client = empty; |
| VM empty1 = replicate3; |
| VM empty2 = normal; |
| |
| for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| |
| client.invoke(() -> createClientCache()); |
| |
| int port1 = empty1.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| int port2 = empty2.invoke(() -> { |
| createRegion(DataPolicy.EMPTY); |
| return createCacheServer(); |
| }); |
| |
| for (VM vm : toArray(replicate1, replicate2)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| }); |
| |
| for (VM vm : toArray(client, empty1, empty2, replicate1, replicate2)) { |
| vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0)); |
| } |
| |
| client.invoke(() -> { |
| // add expected exception to avoid suspect strings |
| try (IgnoredException ie = addIgnoredException("I have been thrown")) { |
| executeFunctionFunctionInvocationTargetException_ClientServer_WithoutHA(); |
| } |
| }); |
| } |
| |
| @Test |
| public void inlineFunctionIsUsedOnClientInsteadOfLookingUpFunctionById() { |
| VM client = empty; |
| VM empty = replicate3; |
| |
| for (VM vm : toArray(normal, empty, replicate1, replicate2)) { |
| vm.invoke(() -> createCache(getDistributedSystemProperties())); |
| } |
| client.invoke(() -> createClientCache(getClientProperties())); |
| |
| int port1 = replicate1.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| int port2 = replicate2.invoke(() -> { |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| for (VM vm : toArray(normal, empty)) { |
| vm.invoke(() -> createRegion(DataPolicy.REPLICATE)); |
| } |
| client.invoke(() -> { |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| }); |
| |
| for (VM vm : toArray(normal, empty, replicate1, replicate2, client)) { |
| vm.invoke(() -> FunctionService.registerFunction(inlineFunction("Failure", false))); |
| } |
| |
| client.invoke(() -> { |
| ResultCollector<Boolean, List<Boolean>> resultCollector = TypedFunctionService |
| .<Boolean, Boolean, List<Boolean>>onRegion(getRegion()) |
| .setArguments(true) |
| .execute(inlineFunction("Success", true)); |
| |
| assertThat(resultCollector.getResult()) |
| .hasSize(1) |
| .containsOnly(true); |
| }); |
| } |
| |
| /** |
| * Verify that AuthenticationRequiredException is not thrown when security-* properties are NOT |
| * provided. We have to grep for this exception in logs for any occurrence. |
| */ |
| @Test |
| public void authenticationRequiredExceptionIsNotThrownWhenSecurityIsNotConfigured() { |
| VM client = replicate1; |
| VM server = replicate2; |
| |
| int port = server.invoke(() -> { |
| createCacheWithSecurity(); |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| client.invoke(() -> { |
| createClientCacheWithSecurity(); |
| createClientRegion(port); |
| |
| executeNoResultFunction(); |
| doPut(); |
| }); |
| } |
| |
| @Test |
| public void testFunctionWithNoResultThrowsException() { |
| VM client = empty; |
| |
| int port1 = replicate1.invoke(() -> { |
| createCache(getDistributedSystemProperties()); |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| int port2 = replicate2.invoke(() -> { |
| createCache(getDistributedSystemProperties()); |
| createRegion(DataPolicy.REPLICATE); |
| return createCacheServer(); |
| }); |
| |
| client.invoke(() -> { |
| createClientCache(); |
| createClientRegion(port1, port2); |
| populateClientRegion(200); |
| |
| try (IgnoredException ie = addIgnoredException(RuntimeException.class)) { |
| executeThrowsRuntimeExceptionFunction(); |
| } |
| }); |
| } |
| |
| private Properties getDistributedSystemProperties() { |
| Properties props = DistributedRule.getDistributedSystemProperties(); |
| props.setProperty(SERIALIZABLE_OBJECT_FILTER, |
| "org.apache.geode.internal.cache.functions.**;org.apache.geode.internal.cache.execute.**;org.apache.geode.test.dunit.**"); |
| return props; |
| } |
| |
| private Properties getClientProperties() { |
| Properties props = new Properties(); |
| props.setProperty(SERIALIZABLE_OBJECT_FILTER, |
| "org.apache.geode.internal.cache.functions.**;org.apache.geode.internal.cache.execute.**;org.apache.geode.test.dunit.**"); |
| return props; |
| } |
| |
| private void createCache(Properties props) { |
| cacheRule.createCache(props); |
| } |
| |
| private void createClientCache(Properties props) { |
| clientCacheRule.createClientCache(props); |
| } |
| |
| private InternalCache getCache() { |
| return cacheRule.getCache(); |
| } |
| |
| private InternalClientCache getClientCache() { |
| return clientCacheRule.getClientCache(); |
| } |
| |
| private void createCacheWithSecurity() { |
| Properties props = getDistributedSystemProperties(); |
| props.setProperty(NAME, "SecurityServer"); |
| props.setProperty(SECURITY_CLIENT_AUTHENTICATOR, |
| DummyAuthenticator.class.getName() + ".create"); |
| |
| createCache(props); |
| } |
| |
| private void createClientCache() { |
| Properties props = getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| |
| createClientCache(props); |
| } |
| |
| private void createClientCacheWithSecurity() { |
| Properties props = getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| props.setProperty(NAME, "SecurityClient"); |
| props.setProperty(SECURITY_CLIENT_AUTH_INIT, UserPasswordAuthInit.class.getName() + ".create"); |
| props.setProperty("security-username", "reader1"); |
| props.setProperty("security-password", "reader1"); |
| |
| createClientCache(props); |
| } |
| |
| private void createClientRegion(int... ports) { |
| System.setProperty(GEMFIRE_PREFIX + "bridge.disableShufflingOfEndpoints", "true"); |
| |
| PoolFactory poolFactory = PoolManager.createFactory(); |
| for (int port : ports) { |
| poolFactory.addServer("localhost", port); |
| } |
| Pool pool = poolFactory |
| .setMaxConnections(10) |
| .setMinConnections(6) |
| .setPingInterval(3000) |
| .setReadTimeout(2000) |
| .setRetryAttempts(2) |
| .setSocketBufferSize(1000) |
| .setSubscriptionEnabled(false) |
| .setSubscriptionRedundancy(-1) |
| .create(poolName); |
| |
| Region<?, ?> region = getClientCache() |
| .createClientRegionFactory(ClientRegionShortcut.PROXY) |
| .setPoolName(pool.getName()) |
| .create(regionName); |
| |
| setRegion(region); |
| } |
| |
| private void createRegion(DataPolicy dataPolicy) { |
| Region<?, ?> region = getCache().createRegionFactory() |
| .setDataPolicy(dataPolicy) |
| .setScope(Scope.DISTRIBUTED_ACK) |
| .create(regionName); |
| |
| setRegion(region); |
| } |
| |
| private int createCacheServer() throws IOException { |
| CacheServer cacheServer = getCache().addCacheServer(); |
| cacheServer.setPort(0); |
| cacheServer.start(); |
| return cacheServer.getPort(); |
| } |
| |
| private void doPut() { |
| Region<String, String> region = getRegion(); |
| region.put("K1", "B1"); |
| } |
| |
| private void populateRegion(int count) { |
| Region<String, Integer> region = getRegion(); |
| for (int i = 1; i <= count; i++) { |
| region.put("execKey-" + i, i); |
| } |
| } |
| |
| private void populateClientRegion(int count) { |
| Region<String, Integer> region = getRegion(); |
| for (int i = 1; i <= count; i++) { |
| region.put("execKey-" + i, i); |
| } |
| } |
| |
| private void startServerHA() throws IOException { |
| for (CacheServer cacheServer : getCache().getCacheServers()) { |
| cacheServer.start(); |
| } |
| } |
| |
| private void stopServerHA() { |
| for (CacheServer cacheServer : getCache().getCacheServers()) { |
| cacheServer.stop(); |
| } |
| } |
| |
| private void closeCacheHA() { |
| for (CacheServer cacheServer : getCache().getCacheServers()) { |
| cacheServer.stop(); |
| } |
| |
| getCache().close(); |
| } |
| |
| private void registerThrowsFunctionInvocationTargetExceptionFunction(boolean isHA, |
| int retryCount) { |
| FunctionService.registerFunction( |
| new ThrowsFunctionInvocationTargetExceptionFunction(isHA, retryCount)); |
| } |
| |
| private void executeNoResultFunction() { |
| TypedFunctionService |
| .onRegion(getRegion()) |
| .execute(new NoResultFunction()); |
| } |
| |
| private List<Boolean> executeDistributedRegionFunction() { |
| return TypedFunctionService |
| .<Boolean, Boolean, List<Boolean>>onRegion(getRegion()) |
| .withFilter(filter) |
| .setArguments(false) |
| .execute(new DistributedRegionFunction()) |
| .getResult(); |
| } |
| |
| private void executeThrowsRuntimeExceptionFunction() { |
| TypedFunctionService |
| .<Void, Void, Void>onRegion(getRegion()) |
| .withFilter(filter) |
| .execute(new ThrowsRuntimeExceptionFunction()); |
| } |
| |
| private void executeResultWithExceptionFunction() { |
| Function function = new ResultWithExceptionFunction(); |
| |
| Set<String> filter = new HashSet<>(); |
| for (int i = 0; i <= 19; i++) { |
| filter.add("execKey-" + 100 + i); |
| } |
| |
| ResultCollector<Object, List<Object>> resultCollector = TypedFunctionService |
| .<Boolean, Object, List<Object>>onRegion(getRegion()) |
| .withFilter(filter) |
| .setArguments(true) |
| .execute(function); |
| |
| List<Object> result = resultCollector.getResult(); |
| result.sort(new NumericComparator()); |
| |
| assertThat(result.get(0)) |
| .as("First element of " + resultCollector.getResult()) |
| .isInstanceOf(CustomRuntimeException.class); |
| |
| resultCollector = TypedFunctionService |
| .<Set<String>, Object, List<Object>>onRegion(getRegion()) |
| .withFilter(filter) |
| .setArguments(filter) |
| .execute(function); |
| |
| result = resultCollector.getResult(); |
| result.sort(new NumericComparator()); |
| |
| assertThat(result) |
| .hasSize(filter.size() + 1) |
| .containsAll(IntStream.rangeClosed(0, 19).boxed().collect(toList())); |
| |
| assertThat(result.get(result.size() - 1)) |
| .as("Last element of " + result) |
| .isInstanceOf(CustomRuntimeException.class); |
| } |
| |
| private void executeNoLastResultFunction() { |
| Throwable thrown = catchThrowable(() -> { |
| TypedFunctionService |
| .onRegion(getRegion()) |
| .withFilter(filter) |
| .execute(new NoLastResultFunction()) |
| .getResult(); |
| }); |
| |
| assertThat(thrown) |
| .hasMessageContaining("did not send last result"); |
| } |
| |
| private void executeUnregisteredFunction() { |
| FunctionService.unregisterFunction(new DistributedRegionFunction().getId()); |
| |
| TypedFunctionService |
| .<Void, Boolean, List<Boolean>>onRegion(getRegion()) |
| .withFilter(filter) |
| .execute(new DistributedRegionFunction()) |
| .getResult(); |
| } |
| |
| private void executeFunctionFunctionInvocationTargetException() { |
| ResultCollector<Integer, List<Integer>> resultCollector = TypedFunctionService |
| .<Boolean, Integer, List<Integer>>onRegion(getRegion()) |
| .setArguments(true) |
| .execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName()); |
| |
| assertThat(resultCollector.getResult()) |
| .containsOnly(5); |
| } |
| |
| private void executeFunctionFunctionInvocationTargetExceptionWithoutHA() { |
| Throwable thrown = catchThrowable(() -> { |
| TypedFunctionService |
| .<Boolean, Integer, List<Integer>>onRegion(getRegion()) |
| .setArguments(true) |
| .execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName()) |
| .getResult(); |
| }); |
| |
| assertThat(thrown) |
| .isInstanceOf(FunctionException.class); |
| |
| assertThat(thrown.getCause()) |
| .isInstanceOf(FunctionInvocationTargetException.class); |
| } |
| |
| private void executeFunctionFunctionInvocationTargetException_ClientServer() { |
| ResultCollector<Integer, List<Integer>> resultCollector = TypedFunctionService |
| .<Boolean, Integer, List<Integer>>onRegion(getRegion()) |
| .setArguments(true) |
| .execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName()); |
| |
| assertThat(resultCollector.getResult()) |
| .containsOnly(5); |
| } |
| |
| private void executeFunctionFunctionInvocationTargetException_ClientServer_WithoutHA() { |
| Throwable thrown = catchThrowable(() -> { |
| TypedFunctionService |
| .<Boolean, Integer, List<Integer>>onRegion(getRegion()) |
| .setArguments(true) |
| .execute(ThrowsFunctionInvocationTargetExceptionFunction.class.getSimpleName()) |
| .getResult(); |
| }); |
| |
| assertThat(thrown) |
| .isInstanceOf(FunctionException.class); |
| |
| assertThat(thrown.getCause()) |
| .isInstanceOf(FunctionInvocationTargetException.class); |
| } |
| |
| private static <K, V> Region<K, V> getRegion() { |
| return uncheckedCast(REGION.get()); |
| } |
| |
| private static void setRegion(Region<?, ?> region) { |
| REGION.set(region); |
| } |
| |
| private static Function<Object> inlineFunction(String stringResult, boolean booleanResult) { |
| return new Function<Object>() { |
| @Override |
| public void execute(FunctionContext<Object> context) { |
| if (context.getArguments() instanceof String) { |
| context.getResultSender().lastResult(stringResult); |
| } else if (context.getArguments() instanceof Boolean) { |
| context.getResultSender().lastResult(booleanResult); |
| } |
| } |
| |
| @Override |
| public String getId() { |
| return "Function"; |
| } |
| }; |
| } |
| |
| /** |
| * Return Integers provided as Arguments and a last result of CustomRuntimeException. |
| */ |
| private static class ResultWithExceptionFunction implements Function<Object> { |
| |
| @Override |
| public void execute(FunctionContext<Object> context) { |
| if (context.getArguments() instanceof Set) { |
| Set<Integer> arguments = |
| uncheckedCast(context.getArguments()); |
| for (int i = 0; i < arguments.size(); i++) { |
| context.getResultSender().sendResult(i); |
| } |
| } |
| context.getResultSender().sendException( |
| new CustomRuntimeException("I have been thrown from TestFunction with set")); |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getName(); |
| } |
| |
| @Override |
| public boolean isHA() { |
| return false; |
| } |
| } |
| |
| /** |
| * Throw RuntimeException without sending any results. |
| */ |
| private static class ThrowsRuntimeExceptionFunction implements Function<Void> { |
| |
| @Override |
| public void execute(FunctionContext<Void> context) { |
| throw new RuntimeException("failure"); |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getName(); |
| } |
| |
| @Override |
| public boolean hasResult() { |
| return false; |
| } |
| |
| @Override |
| public boolean isHA() { |
| return false; |
| } |
| } |
| |
| /** |
| * Return the Integer 5 or throw FunctionInvocationTargetException depending on object state. |
| */ |
| private static class ThrowsFunctionInvocationTargetExceptionFunction implements Function<Void> { |
| |
| private final AtomicInteger count = new AtomicInteger(); |
| private final int retryCount; |
| private final boolean isHA; |
| |
| private ThrowsFunctionInvocationTargetExceptionFunction(boolean isHA, int retryCount) { |
| this.isHA = isHA; |
| this.retryCount = retryCount; |
| } |
| |
| @Override |
| public void execute(FunctionContext<Void> context) { |
| count.incrementAndGet(); |
| if (retryCount != 0 && count.get() >= retryCount) { |
| context.getResultSender().lastResult(5); |
| } else { |
| throw new FunctionInvocationTargetException("I have been thrown from " + getId()); |
| } |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getSimpleName(); |
| } |
| |
| @Override |
| public boolean isHA() { |
| return isHA; |
| } |
| } |
| |
| /** |
| * Validate DataPolicy and Filter state, perform Region put operations, and then return 5000 |
| * boolean true values and a last result of boolean false. |
| */ |
| private static class DistributedRegionFunction implements Function<Boolean> { |
| |
| @Override |
| public void execute(FunctionContext<Boolean> context) { |
| RegionFunctionContext regionFunctionContext = (RegionFunctionContext) context; |
| Region<Object, Object> region = regionFunctionContext.getDataSet(); |
| InternalDistributedSystem sys = InternalDistributedSystem.getConnectedInstance(); |
| |
| assertThat(region.getAttributes().getDataPolicy().withStorage()).isTrue(); |
| assertThat(region.getAttributes().getDataPolicy()).isNotEqualTo(DataPolicy.NORMAL); |
| assertThat(regionFunctionContext.getFilter()).hasSize(20); |
| |
| // argument true indicates that CacheClose has to be done from the body itself |
| if (context.getArguments() != null && context.getArguments()) { |
| // do not close cache in retry |
| if (!regionFunctionContext.isPossibleDuplicate()) { |
| sys.disconnect(); |
| throw new CacheClosedException( |
| "Throwing CacheClosedException to simulate failover during function exception"); |
| } |
| } |
| |
| // intentionally doing region operation to cause cacheClosedException |
| region.put("execKey-201", 201); |
| |
| if (regionFunctionContext.isPossibleDuplicate()) { |
| // Below operation is done when the function is reexecuted |
| region.put("execKey-202", 202); |
| region.put("execKey-203", 203); |
| } |
| |
| for (int i = 0; i < 5000; i++) { |
| context.getResultSender().sendResult(true); |
| } |
| context.getResultSender().lastResult(true); |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getSimpleName(); |
| } |
| } |
| |
| /** |
| * Execute until the provided CountDownLatch counts down to zero. |
| */ |
| private static class LongRunningFunction implements Function<Void> { |
| |
| private final CountDownLatch latch; |
| |
| private LongRunningFunction(CountDownLatch latch) { |
| this.latch = latch; |
| } |
| |
| @Override |
| public void execute(FunctionContext<Void> context) { |
| try { |
| latch.await(getTimeout().toMillis(), MILLISECONDS); |
| } catch (InterruptedException e) { |
| throw new RuntimeException(e); |
| } |
| |
| context.getResultSender().lastResult("LongRunningFunction completed"); |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getSimpleName(); |
| } |
| } |
| |
| /** |
| * Complete immediately with no results. |
| */ |
| private static class NoResultFunction implements Function<Void> { |
| |
| @Override |
| public void execute(FunctionContext<Void> context) { |
| // no result |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getSimpleName(); |
| } |
| |
| @Override |
| public boolean hasResult() { |
| return false; |
| } |
| |
| @Override |
| public boolean isHA() { |
| return false; |
| } |
| } |
| |
| /** |
| * Return Arguments as results without a last result. |
| */ |
| private static class NoLastResultFunction implements Function<Object> { |
| |
| @Override |
| public void execute(FunctionContext<Object> context) { |
| context.getResultSender().sendResult(context.getArguments()); |
| } |
| |
| @Override |
| public String getId() { |
| return getClass().getSimpleName(); |
| } |
| } |
| |
| /** |
| * Custom RuntimeException. |
| */ |
| private static class CustomRuntimeException extends RuntimeException { |
| |
| private CustomRuntimeException(String message) { |
| super(message); |
| } |
| } |
| |
| /** |
| * Use natural ordering for Integers and moves optional Exception to end of Collection. |
| */ |
| private static class NumericComparator implements Comparator<Object>, Serializable { |
| |
| @Override |
| public int compare(Object o1, Object o2) { |
| if (o1 == o2) { |
| return 0; |
| } |
| if (!(o1 instanceof Integer)) { |
| return 1; |
| } |
| if (!(o2 instanceof Integer)) { |
| return -1; |
| } |
| if ((int) o1 > (int) o2) { |
| return 1; |
| } |
| return -1; |
| } |
| } |
| } |
