You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ja...@apache.org on 2018/10/31 04:46:08 UTC
[geode] branch develop updated (8061497 -> 932f81e)
This is an automated email from the ASF dual-hosted git repository.
jasonhuynh pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git.
from 8061497 GEODE-5568: Rewrite QueryMonitor to use ScheduledThreadPoolExecutor (#2744)
new 913cec1 GEODE-5884: Non HA Function waits for all results if exception occurs
new 2410630 GEODE-5884: Refactored methods and added tests
new 932f81e GEODE-5884: Fix typo and renamed test exception class
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
...istributedRegionFunctionExecutionDUnitTest.java | 14 +-
...ientServerRegionFunctionExecutionDUnitTest.java | 2 +-
...RegionFunctionFunctionInvocationException.java} | 8 +-
.../sockets/command/ExecuteRegionFunction66.java | 337 +++++++++++++--------
.../command/ExecuteRegionFunction66Test.java | 196 ++++++++++++
5 files changed, 416 insertions(+), 141 deletions(-)
rename geode-core/src/distributedTest/java/org/apache/geode/internal/cache/functions/{DistribuedRegionFunctionFunctionInvocationException.java => DistributedRegionFunctionFunctionInvocationException.java} (82%)
create mode 100644 geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java
[geode] 03/03: GEODE-5884: Fix typo and renamed test exception class
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jasonhuynh pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 932f81e7d4615454d069c3a9cd2b3c190a9e8dc3
Author: Jason Huynh <hu...@gmail.com>
AuthorDate: Tue Oct 23 09:45:19 2018 -0700
GEODE-5884: Fix typo and renamed test exception class
---
.../execute/DistributedRegionFunctionExecutionDUnitTest.java | 12 ++++++------
...istributedRegionFunctionFunctionInvocationException.java} | 8 ++++----
2 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
index 78d62d9..2d7dc04 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
@@ -59,7 +59,7 @@ import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.LonerDistributionManager;
import org.apache.geode.internal.AvailablePort;
-import org.apache.geode.internal.cache.functions.DistribuedRegionFunctionFunctionInvocationException;
+import org.apache.geode.internal.cache.functions.DistributedRegionFunctionFunctionInvocationException;
import org.apache.geode.internal.cache.functions.DistributedRegionFunction;
import org.apache.geode.internal.cache.functions.TestFunction;
import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
@@ -1024,7 +1024,7 @@ public class DistributedRegionFunctionExecutionDUnitTest extends JUnit4Distribut
}
public static void registerFunction(Boolean isHA, Integer retryCount) {
- Function function = new DistribuedRegionFunctionFunctionInvocationException(isHA.booleanValue(),
+ Function function = new DistributedRegionFunctionFunctionInvocationException(isHA.booleanValue(),
retryCount.intValue());
FunctionService.registerFunction(function);
}
@@ -1131,7 +1131,7 @@ public class DistributedRegionFunctionExecutionDUnitTest extends JUnit4Distribut
public static void executeFunctionFunctionInvocationTargetException() {
try {
ResultCollector rc1 = FunctionService.onRegion(region).setArguments(Boolean.TRUE)
- .execute("DistribuedRegionFunctionFunctionInvocationException");
+ .execute("DistributedRegionFunctionFunctionInvocationException");
List list = (ArrayList) rc1.getResult();
assertEquals(5, list.get(0));
} catch (Exception e) {
@@ -1143,7 +1143,7 @@ public class DistributedRegionFunctionExecutionDUnitTest extends JUnit4Distribut
public static void executeFunctionFunctionInvocationTargetExceptionWithoutHA() {
try {
ResultCollector rc1 = FunctionService.onRegion(region).setArguments(Boolean.TRUE)
- .execute("DistribuedRegionFunctionFunctionInvocationException");
+ .execute("DistributedRegionFunctionFunctionInvocationException");
rc1.getResult();
fail("Function Invocation Target Exception should be thrown");
} catch (Exception e) {
@@ -1157,7 +1157,7 @@ public class DistributedRegionFunctionExecutionDUnitTest extends JUnit4Distribut
public static void executeFunctionFunctionInvocationTargetException_ClientServer() {
try {
List list = (ArrayList) FunctionService.onRegion(region).setArguments(Boolean.TRUE)
- .execute("DistribuedRegionFunctionFunctionInvocationException").getResult();
+ .execute("DistributedRegionFunctionFunctionInvocationException").getResult();
assertEquals(1, list.size());
assertEquals(5, list.get(0));
} catch (Exception e) {
@@ -1169,7 +1169,7 @@ public class DistributedRegionFunctionExecutionDUnitTest extends JUnit4Distribut
public static void executeFunctionFunctionInvocationTargetException_ClientServer_WithoutHA() {
try {
FunctionService.onRegion(region).setArguments(Boolean.TRUE)
- .execute("DistribuedRegionFunctionFunctionInvocationException").getResult();
+ .execute("DistributedRegionFunctionFunctionInvocationException").getResult();
fail("Function Invocation Target Exception should be thrown");
} catch (Exception e) {
e.printStackTrace();
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/functions/DistribuedRegionFunctionFunctionInvocationException.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/functions/DistributedRegionFunctionFunctionInvocationException.java
similarity index 82%
rename from geode-core/src/distributedTest/java/org/apache/geode/internal/cache/functions/DistribuedRegionFunctionFunctionInvocationException.java
rename to geode-core/src/distributedTest/java/org/apache/geode/internal/cache/functions/DistributedRegionFunctionFunctionInvocationException.java
index 811ed6f..b4608b4 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/functions/DistribuedRegionFunctionFunctionInvocationException.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/functions/DistributedRegionFunctionFunctionInvocationException.java
@@ -18,14 +18,14 @@ import org.apache.geode.cache.execute.FunctionAdapter;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionInvocationTargetException;
-public class DistribuedRegionFunctionFunctionInvocationException extends FunctionAdapter {
+public class DistributedRegionFunctionFunctionInvocationException extends FunctionAdapter {
private int retryCount;
private boolean isHA;
private int count;
- public DistribuedRegionFunctionFunctionInvocationException(boolean isHA, int retryCount) {
+ public DistributedRegionFunctionFunctionInvocationException(boolean isHA, int retryCount) {
this.isHA = isHA;
this.retryCount = retryCount;
this.count = 0;
@@ -37,12 +37,12 @@ public class DistribuedRegionFunctionFunctionInvocationException extends Functio
context.getResultSender().lastResult(new Integer(5));
} else {
throw new FunctionInvocationTargetException(
- "I have been thrown from DistribuedRegionFunctionFunctionInvocationException");
+ "I have been thrown from DistributedRegionFunctionFunctionInvocationException");
}
}
public String getId() {
- return "DistribuedRegionFunctionFunctionInvocationException";
+ return "DistributedRegionFunctionFunctionInvocationException";
}
public boolean isHA() {
[geode] 01/03: GEODE-5884: Non HA Function waits for all results if
exception occurs
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jasonhuynh pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 913cec14fd9918918a8f0ada451344fd990e7586
Author: Jason Huynh <hu...@gmail.com>
AuthorDate: Tue Oct 16 14:23:40 2018 -0700
GEODE-5884: Non HA Function waits for all results if exception occurs
* Non HA functions will attempt to return as many results as possible when an exception occurs
* This behavior will be more in-line with single-hop execution
---
...istributedRegionFunctionExecutionDUnitTest.java | 2 +-
...ientServerRegionFunctionExecutionDUnitTest.java | 2 +-
.../sockets/command/ExecuteRegionFunction66.java | 46 +++++----
.../command/ExecuteRegionFunction66Test.java | 113 +++++++++++++++++++++
4 files changed, 143 insertions(+), 20 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
index 0301ba8..78d62d9 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/DistributedRegionFunctionExecutionDUnitTest.java
@@ -1173,7 +1173,7 @@ public class DistributedRegionFunctionExecutionDUnitTest extends JUnit4Distribut
fail("Function Invocation Target Exception should be thrown");
} catch (Exception e) {
e.printStackTrace();
- if (!(e.getCause() instanceof FunctionInvocationTargetException)) {
+ if (!((e instanceof FunctionInvocationTargetException) || (e.getCause() instanceof FunctionInvocationTargetException))) {
fail("FunctionInvocationTargetException should be thrown");
}
}
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/PRClientServerRegionFunctionExecutionDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/PRClientServerRegionFunctionExecutionDUnitTest.java
index 07564ec..c1f0bfe 100755
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/PRClientServerRegionFunctionExecutionDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/execute/PRClientServerRegionFunctionExecutionDUnitTest.java
@@ -477,7 +477,7 @@ public class PRClientServerRegionFunctionExecutionDUnitTest extends PRClientServ
fail("Expected ServerConnectivityException not thrown!");
} catch (Exception ex) {
if (!(ex.getCause() instanceof ServerConnectivityException)
- && !(ex.getCause() instanceof FunctionInvocationTargetException)) {
+ && !((ex instanceof FunctionInvocationTargetException || ex.getCause() instanceof FunctionInvocationTargetException))) {
throw ex;
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
index 3a0c018..e953f00 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
@@ -246,24 +246,7 @@ public class ExecuteRegionFunction66 extends BaseCommand {
hasResult);
}
if (hasResult == 1) {
- if (function instanceof String) {
- switch (functionState) {
- case AbstractExecution.NO_HA_HASRESULT_NO_OPTIMIZEFORWRITE:
- execution.execute((String) function).getResult();
- break;
- case AbstractExecution.HA_HASRESULT_NO_OPTIMIZEFORWRITE:
- execution.execute((String) function).getResult();
- break;
- case AbstractExecution.HA_HASRESULT_OPTIMIZEFORWRITE:
- execution.execute((String) function).getResult();
- break;
- case AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE:
- execution.execute((String) function).getResult();
- break;
- }
- } else {
- execution.execute(functionObject).getResult();
- }
+ executeFunctionWithResult(function, functionState, functionObject, execution);
} else {
if (function instanceof String) {
switch (functionState) {
@@ -339,6 +322,33 @@ public class ExecuteRegionFunction66 extends BaseCommand {
}
}
+ void executeFunctionWithResult(Object function, byte functionState,
+ Function<?> functionObject, AbstractExecution execution) {
+ if (function instanceof String) {
+ switch (functionState) {
+ case AbstractExecution.NO_HA_HASRESULT_NO_OPTIMIZEFORWRITE:
+ execution.setWaitOnExceptionFlag(true);
+ execution.execute((String) function).getResult();
+ break;
+ case AbstractExecution.HA_HASRESULT_NO_OPTIMIZEFORWRITE:
+ execution.execute((String) function).getResult();
+ break;
+ case AbstractExecution.HA_HASRESULT_OPTIMIZEFORWRITE:
+ execution.execute((String) function).getResult();
+ break;
+ case AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE:
+ execution.setWaitOnExceptionFlag(true);
+ execution.execute((String) function).getResult();
+ break;
+ }
+ } else {
+ if (!functionObject.isHA()) {
+ execution.setWaitOnExceptionFlag(true);
+ }
+ execution.execute(functionObject).getResult();
+ }
+ }
+
private void sendException(byte hasResult, Message msg, String message,
ServerConnection serverConnection, Throwable e) throws IOException {
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java
new file mode 100644
index 0000000..b77a879
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java
@@ -0,0 +1,113 @@
+/*
+ * 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.tier.sockets.command;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+import org.junit.experimental.categories.Category;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import org.apache.geode.CancelCriterion;
+import org.apache.geode.cache.execute.Execution;
+import org.apache.geode.cache.execute.Function;
+import org.apache.geode.cache.execute.FunctionService;
+import org.apache.geode.cache.execute.ResultCollector;
+import org.apache.geode.cache.operations.ExecuteFunctionOperationContext;
+import org.apache.geode.cache.query.internal.ExecutionContext;
+import org.apache.geode.distributed.internal.DistributionConfig;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.internal.Version;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.control.HeapMemoryMonitor;
+import org.apache.geode.internal.cache.control.InternalResourceManager;
+import org.apache.geode.internal.cache.execute.AbstractExecution;
+import org.apache.geode.internal.cache.tier.CachedRegionHelper;
+import org.apache.geode.internal.cache.tier.ServerSideHandshake;
+import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
+import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage;
+import org.apache.geode.internal.cache.tier.sockets.Message;
+import org.apache.geode.internal.cache.tier.sockets.Part;
+import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
+import org.apache.geode.internal.security.AuthorizeRequest;
+import org.apache.geode.internal.security.SecurityService;
+import org.apache.geode.management.internal.security.ResourcePermissions;
+import org.apache.geode.security.NotAuthorizedException;
+import org.apache.geode.test.junit.categories.ClientServerTest;
+
+@Category({ClientServerTest.class})
+public class ExecuteRegionFunction66Test {
+ private static final String FUNCTION_ID = "function_id";
+
+ @Mock
+ private Function functionObject;
+
+ private ExecuteRegionFunction66 executeRegionFunction66;
+
+ @Rule
+ public RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
+
+ @Before
+ public void setUp() throws Exception {
+ this.executeRegionFunction66 = (ExecuteRegionFunction66) ExecuteRegionFunction66.getCommand();
+
+ this.functionObject = mock(Function.class);
+ when(this.functionObject.getId()).thenReturn(FUNCTION_ID);
+ doCallRealMethod().when(this.functionObject).getRequiredPermissions(any());
+ }
+
+ @Test
+ public void executingFunctionByStringWithNoHAShouldSetWaitOnException() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ String functionName = "functionName";
+ when (execution.execute(functionName)).thenReturn(mock(ResultCollector.class));
+ this.executeRegionFunction66.executeFunctionWithResult(functionName, AbstractExecution.NO_HA_HASRESULT_NO_OPTIMIZEFORWRITE, functionObject, execution);
+ verify(execution, times(1)).setWaitOnExceptionFlag(true);
+ }
+
+ @Test
+ public void executingFunctionByStringWithNoHAWithOptimizeForWriteShouldSetWaitOnException() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ String functionName = "functionName";
+ when (execution.execute(functionName)).thenReturn(mock(ResultCollector.class));
+ this.executeRegionFunction66.executeFunctionWithResult(functionName, AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE, functionObject, execution);
+ verify(execution, times(1)).setWaitOnExceptionFlag(true);
+ }
+
+ @Test
+ public void executeFunctionObjectShouldSetWaitOnException() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ when (execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+ this.executeRegionFunction66.executeFunctionWithResult(functionObject, AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE, functionObject, execution);
+ verify(execution, times(1)).setWaitOnExceptionFlag(true);
+ }
+
+
+
+}
[geode] 02/03: GEODE-5884: Refactored methods and added tests
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jasonhuynh pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 24106302e27004137a2e7459be24d93368321056
Author: Jason Huynh <hu...@gmail.com>
AuthorDate: Thu Oct 18 10:26:30 2018 -0700
GEODE-5884: Refactored methods and added tests
---
.../sockets/command/ExecuteRegionFunction66.java | 293 +++++++++++++--------
.../command/ExecuteRegionFunction66Test.java | 145 +++++++---
2 files changed, 295 insertions(+), 143 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
index e953f00..beb9128 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66.java
@@ -73,10 +73,8 @@ public class ExecuteRegionFunction66 extends BaseCommand {
final byte isReExecute;
Set<Object> filter = null;
byte hasResult = 0;
- int removedNodesSize = 0;
Set<Object> removedNodesSet = null;
- int filterSize = 0, partNumber = 0;
- CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
+ int partNumber = 0;
byte functionState = 0;
int functionTimeout = ConnectionImpl.DEFAULT_CLIENT_FUNCTION_TIMEOUT;
try {
@@ -99,12 +97,8 @@ public class ExecuteRegionFunction66 extends BaseCommand {
function = clientMessage.getPart(2).getStringOrObject();
args = clientMessage.getPart(3).getObject();
Part part = clientMessage.getPart(4);
- if (part != null) {
- Object obj = part.getObject();
- if (obj instanceof MemberMappedArgument) {
- memberMappedArg = (MemberMappedArgument) obj;
- }
- }
+ memberMappedArg = extractMemberMappedArgument(part);
+
byte[] flags = clientMessage.getPart(5).getSerializedForm();
if (serverConnection.getClientVersion().ordinal() > Version.GFE_81.ordinal()) {
isBucketsAsFilter = (flags[0] & ExecuteFunctionHelper.BUCKETS_AS_FILTER_MASK) != 0;
@@ -113,26 +107,14 @@ public class ExecuteRegionFunction66 extends BaseCommand {
isReExecute = flags[0];
isBucketsAsFilter = false;
}
- filterSize = clientMessage.getPart(6).getInt();
- if (filterSize != 0) {
- filter = new HashSet<Object>();
- partNumber = 7;
- for (int i = 0; i < filterSize; i++) {
- filter.add(clientMessage.getPart(partNumber + i).getStringOrObject());
- }
- }
+ int filterSize = clientMessage.getPart(6).getInt();
+ filter = populateFilters(clientMessage, filterSize);
partNumber = 7 + filterSize;
- removedNodesSize = clientMessage.getPart(partNumber).getInt();
- if (removedNodesSize != 0) {
- removedNodesSet = new HashSet<Object>();
- partNumber = partNumber + 1;
-
- for (int i = 0; i < removedNodesSize; i++) {
- removedNodesSet.add(clientMessage.getPart(partNumber + i).getStringOrObject());
- }
- }
+ int removedNodesSize = clientMessage.getPart(partNumber).getInt();
+ removedNodesSet =
+ populateRemovedNodes(clientMessage, removedNodesSize, partNumber);
} catch (ClassNotFoundException exception) {
logger.warn(String.format("Exception on server while executing function : %s",
@@ -147,22 +129,13 @@ public class ExecuteRegionFunction66 extends BaseCommand {
return;
}
if (function == null || regionName == null) {
- String message = null;
- if (function == null) {
- message =
- String.format("The input %s for the execute function request is null",
- "function");
- }
- if (regionName == null) {
- message =
- String.format("The input %s for the execute function request is null",
- "region");
- }
+ String message = generateNullArgumentMessage(regionName, function);
logger.warn("{}: {}", serverConnection.getName(), message);
sendError(hasResult, clientMessage, message, serverConnection);
return;
}
+ CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper();
Region region = crHelper.getRegion(regionName);
if (region == null) {
String message =
@@ -180,28 +153,10 @@ public class ExecuteRegionFunction66 extends BaseCommand {
try {
if (function instanceof String) {
functionObject = FunctionService.getFunction((String) function);
- if (functionObject == null) {
- String message =
- String.format("The function, %s, has not been registered",
- function);
- logger.warn("{}: {}", serverConnection.getName(), message);
- sendError(hasResult, clientMessage, message, serverConnection);
+ if (!validateFunctionObject(clientMessage, serverConnection, function, hasResult,
+ functionState,
+ functionObject)) {
return;
- } else {
- byte functionStateOnServerSide = AbstractExecution.getFunctionState(functionObject.isHA(),
- functionObject.hasResult(), functionObject.optimizeForWrite());
- if (logger.isDebugEnabled()) {
- logger.debug("Function State on server side: {} on client: {}",
- functionStateOnServerSide, functionState);
- }
- if (functionStateOnServerSide != functionState) {
- String message =
- String.format("Function attributes at client and server don't match: %s",
- function);
- logger.warn("{}: {}", serverConnection.getName(), message);
- sendError(hasResult, clientMessage, message, serverConnection);
- return;
- }
}
} else {
functionObject = (Function) function;
@@ -209,32 +164,26 @@ public class ExecuteRegionFunction66 extends BaseCommand {
// check if the caller is authorized to do this operation on server
functionObject.getRequiredPermissions(regionName).forEach(securityService::authorize);
- AuthorizeRequest authzRequest = serverConnection.getAuthzRequest();
- final String functionName = functionObject.getId();
- final String regionPath = region.getFullPath();
- ExecuteFunctionOperationContext executeContext = null;
- if (authzRequest != null) {
- executeContext = authzRequest.executeFunctionAuthorize(functionName, regionPath, filter,
- args, functionObject.optimizeForWrite());
- }
+ ExecuteFunctionOperationContext executeContext =
+ getAuthorizedExecuteFunctionOperationContext(args, filter,
+ functionObject.optimizeForWrite(), serverConnection.getAuthzRequest(),
+ functionObject.getId(), region.getFullPath());
+
- // Construct execution
- AbstractExecution execution = (AbstractExecution) FunctionService.onRegion(region);
ChunkedMessage m = serverConnection.getFunctionResponseMessage();
m.setTransactionId(clientMessage.getTransactionId());
resultSender =
new ServerToClientFunctionResultSender65(m, MessageType.EXECUTE_REGION_FUNCTION_RESULT,
serverConnection, functionObject, executeContext);
+ AbstractExecution execution =
+ createExecution(args, memberMappedArg, isBucketsAsFilter, filter,
+ removedNodesSet,
+ region, resultSender);
if (execution instanceof PartitionedRegionFunctionExecutor) {
if ((hasResult == 1) && filter != null && filter.size() == 1) {
ServerConnection.executeFunctionOnLocalNodeOnly((byte) 1);
}
- execution = new PartitionedRegionFunctionExecutor((PartitionedRegion) region, filter, args,
- memberMappedArg, resultSender, removedNodesSet, isBucketsAsFilter);
- } else {
- execution = new DistributedRegionFunctionExecutor((DistributedRegion) region, filter, args,
- memberMappedArg, resultSender);
}
if (isReExecute == 1) {
execution = execution.setIsReExecute();
@@ -248,18 +197,7 @@ public class ExecuteRegionFunction66 extends BaseCommand {
if (hasResult == 1) {
executeFunctionWithResult(function, functionState, functionObject, execution);
} else {
- if (function instanceof String) {
- switch (functionState) {
- case AbstractExecution.NO_HA_NO_HASRESULT_NO_OPTIMIZEFORWRITE:
- execution.execute((String) function);
- break;
- case AbstractExecution.NO_HA_NO_HASRESULT_OPTIMIZEFORWRITE:
- execution.execute((String) function);
- break;
- }
- } else {
- execution.execute(functionObject);
- }
+ executeFunctionNoResult(function, functionState, functionObject, execution);
writeReply(clientMessage, serverConnection);
}
} catch (IOException ioe) {
@@ -273,31 +211,7 @@ public class ExecuteRegionFunction66 extends BaseCommand {
Object cause = fe.getCause();
if (cause instanceof FunctionInvocationTargetException
|| cause instanceof QueryInvocationTargetException) {
- if (cause instanceof InternalFunctionInvocationTargetException) {
- // Fix for #44709: User should not be aware of
- // InternalFunctionInvocationTargetException. No instance of
- // InternalFunctionInvocationTargetException is giving useful
- // information to user to take any corrective action hence logging
- // this at fine level logging
- // 1> When bucket is moved
- // 2> Incase of HA FucntionInvocationTargetException thrown. Since
- // it is HA, fucntion will be reexecuted on right node
- // 3> Multiple target nodes found for single hop operation
- // 4> in case of HA member departed
- if (logger.isDebugEnabled()) {
- logger.debug(String.format("Exception on server while executing function: %s",
- new Object[] {function}),
- fe);
- }
- } else if (functionObject.isHA()) {
- logger.warn("Exception on server while executing function : {}",
- function + " :" + message);
- } else {
- logger.warn(String.format("Exception on server while executing function : %s",
- function),
- fe);
- }
-
+ logFunctionExceptionCause(function, functionObject, fe, message, cause);
resultSender.setException(fe);
} else {
if (setLastResultReceived(resultSender)) {
@@ -307,7 +221,6 @@ public class ExecuteRegionFunction66 extends BaseCommand {
sendException(hasResult, clientMessage, message, serverConnection, fe);
}
}
-
} catch (Exception e) {
if (setLastResultReceived(resultSender)) {
logger.warn(String.format("Exception on server while executing function : %s",
@@ -322,8 +235,164 @@ public class ExecuteRegionFunction66 extends BaseCommand {
}
}
+ void logFunctionExceptionCause(Object function, Function<?> functionObject,
+ FunctionException fe, String message, Object cause) {
+ if (cause instanceof InternalFunctionInvocationTargetException) {
+ // Fix for #44709: User should not be aware of
+ // InternalFunctionInvocationTargetException. No instance of
+ // InternalFunctionInvocationTargetException is giving useful
+ // information to user to take any corrective action hence logging
+ // this at fine level logging
+ // 1> When bucket is moved
+ // 2> Incase of HA FucntionInvocationTargetException thrown. Since
+ // it is HA, fucntion will be reexecuted on right node
+ // 3> Multiple target nodes found for single hop operation
+ // 4> in case of HA member departed
+ if (logger.isDebugEnabled()) {
+ logger.debug(String.format("Exception on server while executing function: %s",
+ new Object[] {function}),
+ fe);
+ }
+ } else if (functionObject.isHA()) {
+ logger.warn("Exception on server while executing function : {}",
+ function + " :" + message);
+ } else {
+ logger.warn(String.format("Exception on server while executing function : %s",
+ function),
+ fe);
+ }
+ }
+
+ AbstractExecution createExecution(Object args, MemberMappedArgument memberMappedArg,
+ boolean isBucketsAsFilter, Set<Object> filter,
+ Set<Object> removedNodesSet, Region region,
+ ServerToClientFunctionResultSender resultSender) {
+
+ AbstractExecution execution = (AbstractExecution) FunctionService.onRegion(region);
+ if (execution instanceof PartitionedRegionFunctionExecutor) {
+ execution = new PartitionedRegionFunctionExecutor((PartitionedRegion) region, filter, args,
+ memberMappedArg, resultSender, removedNodesSet, isBucketsAsFilter);
+ } else {
+ execution = new DistributedRegionFunctionExecutor((DistributedRegion) region, filter, args,
+ memberMappedArg, resultSender);
+ }
+ return execution;
+ }
+
+ boolean validateFunctionObject(Message clientMessage, ServerConnection serverConnection,
+ Object function, byte hasResult, byte functionState,
+ Function<?> functionObject) throws IOException {
+ if (functionObject == null) {
+ String message =
+ String.format("The function, %s, has not been registered",
+ function);
+ logger.warn("{}: {}", serverConnection.getName(), message);
+ sendError(hasResult, clientMessage, message, serverConnection);
+ return false;
+ } else {
+ byte functionStateOnServerSide = AbstractExecution.getFunctionState(functionObject.isHA(),
+ functionObject.hasResult(), functionObject.optimizeForWrite());
+ if (logger.isDebugEnabled()) {
+ logger.debug("Function State on server side: {} on client: {}",
+ functionStateOnServerSide, functionState);
+ }
+ if (functionStateOnServerSide != functionState) {
+ String message =
+ String.format("Function attributes at client and server don't match: %s",
+ function);
+ logger.warn("{}: {}", serverConnection.getName(), message);
+ sendError(hasResult, clientMessage, message, serverConnection);
+ return false;
+ }
+ }
+ return true;
+ }
+
+ String generateNullArgumentMessage(String regionName, Object function) {
+ String message = null;
+ if (regionName == null) {
+ message =
+ String.format("The input region for the execute function request is null");
+ } else if (function == null) {
+ message =
+ String.format("The input function for the execute function request is null");
+ }
+
+ return message;
+ }
+
+ MemberMappedArgument extractMemberMappedArgument(Part part)
+ throws IOException, ClassNotFoundException {
+ MemberMappedArgument memberMappedArg = null;
+ if (part != null) {
+ Object obj = part.getObject();
+ if (obj instanceof MemberMappedArgument) {
+ memberMappedArg = (MemberMappedArgument) obj;
+ }
+ }
+ return memberMappedArg;
+ }
+
+ Set<Object> populateRemovedNodes(Message clientMessage, int removedNodesSize, int partNumber)
+ throws IOException, ClassNotFoundException {
+ Set<Object> removedNodesSet = null;
+ if (removedNodesSize != 0) {
+ removedNodesSet = new HashSet<Object>();
+ partNumber = partNumber + 1;
+
+ for (int i = 0; i < removedNodesSize; i++) {
+ removedNodesSet.add(clientMessage.getPart(partNumber + i).getStringOrObject());
+ }
+ }
+ return removedNodesSet;
+ }
+
+ Set<Object> populateFilters(Message clientMessage, int filterSize)
+ throws IOException, ClassNotFoundException {
+ Set<Object> filter = null;
+ int partNumber;
+ if (filterSize != 0) {
+ filter = new HashSet<Object>();
+ partNumber = 7;
+ for (int i = 0; i < filterSize; i++) {
+ filter.add(clientMessage.getPart(partNumber + i).getStringOrObject());
+ }
+ }
+ return filter;
+ }
+
+ ExecuteFunctionOperationContext getAuthorizedExecuteFunctionOperationContext(Object args,
+ Set<Object> filter,
+ boolean optimizedForWrite,
+ AuthorizeRequest authzRequest,
+ String functionName,
+ String regionPath) {
+ ExecuteFunctionOperationContext executeContext = null;
+ if (authzRequest != null) {
+ executeContext = authzRequest.executeFunctionAuthorize(functionName, regionPath, filter,
+ args, optimizedForWrite);
+ }
+ return executeContext;
+ }
+
+ void executeFunctionNoResult(Object function, byte functionState,
+ Function<?> functionObject, AbstractExecution execution) {
+ if (function instanceof String) {
+ switch (functionState) {
+ case AbstractExecution.NO_HA_NO_HASRESULT_NO_OPTIMIZEFORWRITE:
+ execution.execute((String) function);
+ break;
+ case AbstractExecution.NO_HA_NO_HASRESULT_OPTIMIZEFORWRITE:
+ execution.execute((String) function);
+ break;
+ }
+ } else {
+ execution.execute(functionObject);
+ }
+ }
+
void executeFunctionWithResult(Object function, byte functionState,
- Function<?> functionObject, AbstractExecution execution) {
+ Function<?> functionObject, AbstractExecution execution) {
if (function instanceof String) {
switch (functionState) {
case AbstractExecution.NO_HA_HASRESULT_NO_OPTIMIZEFORWRITE:
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java
index b77a879..f7d369e 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteRegionFunction66Test.java
@@ -14,51 +14,35 @@
*/
package org.apache.geode.internal.cache.tier.sockets.command;
+import static junit.framework.TestCase.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.doCallRealMethod;
-import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.Set;
+
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.contrib.java.lang.system.RestoreSystemProperties;
import org.junit.experimental.categories.Category;
-import org.mockito.InjectMocks;
import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-import org.apache.geode.CancelCriterion;
-import org.apache.geode.cache.execute.Execution;
import org.apache.geode.cache.execute.Function;
-import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.operations.ExecuteFunctionOperationContext;
-import org.apache.geode.cache.query.internal.ExecutionContext;
-import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.distributed.internal.InternalDistributedSystem;
-import org.apache.geode.internal.Version;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
-import org.apache.geode.internal.cache.LocalRegion;
-import org.apache.geode.internal.cache.control.HeapMemoryMonitor;
-import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.execute.AbstractExecution;
-import org.apache.geode.internal.cache.tier.CachedRegionHelper;
-import org.apache.geode.internal.cache.tier.ServerSideHandshake;
-import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl;
-import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage;
import org.apache.geode.internal.cache.tier.sockets.Message;
import org.apache.geode.internal.cache.tier.sockets.Part;
-import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
import org.apache.geode.internal.security.AuthorizeRequest;
-import org.apache.geode.internal.security.SecurityService;
-import org.apache.geode.management.internal.security.ResourcePermissions;
-import org.apache.geode.security.NotAuthorizedException;
import org.apache.geode.test.junit.categories.ClientServerTest;
@Category({ClientServerTest.class})
@@ -86,28 +70,127 @@ public class ExecuteRegionFunction66Test {
public void executingFunctionByStringWithNoHAShouldSetWaitOnException() throws Exception {
AbstractExecution execution = mock(AbstractExecution.class);
String functionName = "functionName";
- when (execution.execute(functionName)).thenReturn(mock(ResultCollector.class));
- this.executeRegionFunction66.executeFunctionWithResult(functionName, AbstractExecution.NO_HA_HASRESULT_NO_OPTIMIZEFORWRITE, functionObject, execution);
+ when(execution.execute(functionName)).thenReturn(mock(ResultCollector.class));
+ this.executeRegionFunction66.executeFunctionWithResult(functionName,
+ AbstractExecution.NO_HA_HASRESULT_NO_OPTIMIZEFORWRITE, functionObject, execution);
verify(execution, times(1)).setWaitOnExceptionFlag(true);
}
@Test
- public void executingFunctionByStringWithNoHAWithOptimizeForWriteShouldSetWaitOnException() throws Exception {
+ public void executingFunctionByStringWithNoHAWithOptimizeForWriteShouldSetWaitOnException()
+ throws Exception {
AbstractExecution execution = mock(AbstractExecution.class);
String functionName = "functionName";
- when (execution.execute(functionName)).thenReturn(mock(ResultCollector.class));
- this.executeRegionFunction66.executeFunctionWithResult(functionName, AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE, functionObject, execution);
+ when(execution.execute(functionName)).thenReturn(mock(ResultCollector.class));
+ this.executeRegionFunction66.executeFunctionWithResult(functionName,
+ AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE, functionObject, execution);
verify(execution, times(1)).setWaitOnExceptionFlag(true);
}
@Test
public void executeFunctionObjectShouldSetWaitOnException() throws Exception {
AbstractExecution execution = mock(AbstractExecution.class);
- when (execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
- this.executeRegionFunction66.executeFunctionWithResult(functionObject, AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE, functionObject, execution);
+ when(execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+ this.executeRegionFunction66.executeFunctionWithResult(functionObject,
+ AbstractExecution.NO_HA_HASRESULT_OPTIMIZEFORWRITE, functionObject, execution);
verify(execution, times(1)).setWaitOnExceptionFlag(true);
}
+ @Test
+ public void generateNullArgumentMessageIfRegionIsNull() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ when(execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+ assertEquals("The input region for the execute function request is null",
+ this.executeRegionFunction66.generateNullArgumentMessage(null, null));
+ }
+
+ @Test
+ public void generateNullArgumentMessageIfFunctionIsNullAndRegionIsNotNull() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ when(execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+ assertEquals("The input function for the execute function request is null",
+ this.executeRegionFunction66.generateNullArgumentMessage("someRegion", null));
+ }
+
+ @Test
+ public void populateFiltersWillReturnFiltersReadFromClientMessage() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ when(execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+
+ Message clientMessage = mock(Message.class);
+ Part part1 = mock(Part.class);
+ Object object1 = new Object();
+ when(part1.getStringOrObject()).thenReturn(object1);
+ Part part2 = mock(Part.class);
+ Object object2 = new Object();
+ when(part2.getStringOrObject()).thenReturn(object2);
+ Part part3 = mock(Part.class);
+ Object object3 = new Object();
+ when(part3.getStringOrObject()).thenReturn(object3);
+
+ when(clientMessage.getPart(7)).thenReturn(part1);
+ when(clientMessage.getPart(8)).thenReturn(part2);
+ when(clientMessage.getPart(9)).thenReturn(part3);
+ int filterSize = 3;
+ Set filter = this.executeRegionFunction66.populateFilters(clientMessage, filterSize);
+ assertSame(filterSize, filter.size());
+ assertTrue(filter.contains(object1));
+ assertTrue(filter.contains(object2));
+ assertTrue(filter.contains(object3));
+ }
+
+ @Test
+ public void populateRemovedNodexWillReturnNodesReadFromClient() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ when(execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+
+ Message clientMessage = mock(Message.class);
+ Part part1 = mock(Part.class);
+ Object object1 = new Object();
+ when(part1.getStringOrObject()).thenReturn(object1);
+ Part part2 = mock(Part.class);
+ Object object2 = new Object();
+ when(part2.getStringOrObject()).thenReturn(object2);
+ Part part3 = mock(Part.class);
+ Object object3 = new Object();
+ when(part3.getStringOrObject()).thenReturn(object3);
+
+ when(clientMessage.getPart(7)).thenReturn(part1);
+ when(clientMessage.getPart(8)).thenReturn(part2);
+ when(clientMessage.getPart(9)).thenReturn(part3);
+ Set nodes = this.executeRegionFunction66.populateRemovedNodes(clientMessage, 3, 6);
+ assertTrue(nodes.contains(object1));
+ assertTrue(nodes.contains(object2));
+ assertTrue(nodes.contains(object3));
+ }
+ @Test
+ public void getAuthorizedExecuteFunctionReturnsNullIfAuthorizationIsNull() throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ when(execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+ String functionName = "functionName";
+ String regionPath = "regionPath";
+ ExecuteFunctionOperationContext context =
+ executeRegionFunction66.getAuthorizedExecuteFunctionOperationContext(null, null, true, null,
+ functionName, regionPath);
+ assertNull(context);
+ }
+
+ @Test
+ public void getAuthorizedExecuteFunctionReturnsExecutionContextIfAuthorizeRequestIsNotNull()
+ throws Exception {
+ AbstractExecution execution = mock(AbstractExecution.class);
+ when(execution.execute(functionObject)).thenReturn(mock(ResultCollector.class));
+ String functionName = "functionName";
+ String regionPath = "regionPath";
+ AuthorizeRequest request = mock(AuthorizeRequest.class);
+ when(request.executeFunctionAuthorize(any(), any(), any(), any(), anyBoolean()))
+ .thenReturn(mock(ExecuteFunctionOperationContext.class));
+
+ ExecuteFunctionOperationContext context =
+ executeRegionFunction66.getAuthorizedExecuteFunctionOperationContext(null, null, true,
+ request, functionName, regionPath);
+ assertNotNull(context);
+ }
}