You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by zh...@apache.org on 2019/10/03 23:28:45 UTC
[geode] 01/01: GEODE-7258: retry number should be based on current
live members
This is an automated email from the ASF dual-hosted git repository.
zhouxj pushed a commit to branch feature/GEODE-7258
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 44fe3af55f93ede5e29aae83691488d2e134a969
Author: zhouxh <gz...@pivotal.io>
AuthorDate: Thu Oct 3 16:26:59 2019 -0700
GEODE-7258: retry number should be based on current live members
---
.../client/internal/ExecuteRegionFunctionOp.java | 13 ++++++-
.../internal/ExecuteRegionFunctionSingleHopOp.java | 6 +++-
.../client/internal/SingleHopClientExecutor.java | 8 ++++-
.../LuceneSearchWithRollingUpgradeDUnit.java | 42 ++++++++++++++--------
...tResultsAfterClientAndServersAreRolledOver.java | 5 +--
...ntAndServersAreRolledOverAllBucketsCreated.java | 5 ++-
6 files changed, 57 insertions(+), 22 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java
index eba38a6..e265e60 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionOp.java
@@ -78,6 +78,7 @@ public class ExecuteRegionFunctionOp {
ExecuteRegionFunctionOpImpl op, boolean isReexecute,
Set<String> failedNodes) {
+ final boolean isDebugEnabled = logger.isDebugEnabled();
if (!isHA) {
maxRetryAttempts = 0;
}
@@ -111,7 +112,17 @@ public class ExecuteRegionFunctionOp {
// If the retryAttempt is set to default(-1). Try it on all servers once.
// Calculating number of servers when function is re-executed as it involves
// messaging locator.
- maxRetryAttempts = ((PoolImpl) pool).getConnectionSource().getAllServers().size() - 1;
+ // Since it's in exception handling, getAllServers() will return
+ // current available servers, excluded the failed one.
+ maxRetryAttempts = ((PoolImpl) pool).getConnectionSource().getAllServers().size();
+ if (isDebugEnabled) {
+ logger.debug("maxRetryAttempts was using pool's default, change to " + maxRetryAttempts
+ + " based on current live server number");
+ }
+ } else {
+ if (isDebugEnabled) {
+ logger.debug("There are still " + maxRetryAttempts + " retries");
+ }
}
if ((maxRetryAttempts--) < 1) {
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
index eb162829..92c182d 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ExecuteRegionFunctionSingleHopOp.java
@@ -102,8 +102,12 @@ public class ExecuteRegionFunctionSingleHopOp {
(ExecuteRegionFunctionOp.ExecuteRegionFunctionOpImpl) executeRegionFunctionOpSupplier
.get();
+ if (isDebugEnabled) {
+ logger.debug(
+ "Singlehop failed, there are " + retryAttempts + " retries left on non-singlehop mode");
+ }
ExecuteRegionFunctionOp.execute(pool,
- resultCollector, retryAttempts - 1,
+ resultCollector, retryAttempts,
isHA,
executeRegionFunctionOp, true, failedNodes);
}
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/SingleHopClientExecutor.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/SingleHopClientExecutor.java
index 40a128c..f9492ff 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/client/internal/SingleHopClientExecutor.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/client/internal/SingleHopClientExecutor.java
@@ -128,7 +128,13 @@ public class SingleHopClientExecutor {
// If the retryAttempt is set to default(-1). Try it on all servers once.
// Calculating number of servers when function is re-executed as it involves
// messaging locator.
- maxRetryAttempts = pool.getConnectionSource().getAllServers().size() - 1;
+ // Since it's in exception handling, getAllServers() will return
+ // current available servers, excluded the failed one.
+ maxRetryAttempts = pool.getConnectionSource().getAllServers().size();
+ if (isDebugEnabled) {
+ logger.debug("In singlehop, maxRetryAttempts was using pool's default, change to "
+ + maxRetryAttempts + " based on current live server number");
+ }
}
if (ee.getCause() instanceof InternalFunctionInvocationTargetException) {
diff --git a/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java b/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java
index 2a8cf9f..69d9573 100644
--- a/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java
+++ b/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/LuceneSearchWithRollingUpgradeDUnit.java
@@ -68,14 +68,16 @@ import org.apache.geode.test.version.VersionManager;
public abstract class LuceneSearchWithRollingUpgradeDUnit extends JUnit4DistributedTestCase {
- @Parameterized.Parameters(name = "from_v{0}, with reindex={1}")
+ @Parameterized.Parameters(name = "from_v{0}, with reindex={1}, singleHopEnabled={2}")
public static Collection<Object[]> data() {
Collection<String> luceneVersions = getLuceneVersions();
Collection<Object[]> rval = new ArrayList<>();
luceneVersions.forEach(v -> {
- rval.add(new Object[] {v, true});
- rval.add(new Object[] {v, false});
+ rval.add(new Object[] {v, true, true});
+ rval.add(new Object[] {v, false, true});
});
+ rval.add(new Object[] {VersionManager.CURRENT_VERSION, true, true});
+ rval.add(new Object[] {VersionManager.CURRENT_VERSION, true, false});
return rval;
}
@@ -108,6 +110,9 @@ public abstract class LuceneSearchWithRollingUpgradeDUnit extends JUnit4Distribu
@Parameterized.Parameter(1)
public Boolean reindex;
+ @Parameterized.Parameter(2)
+ public Boolean singleHopEnabled;
+
private void deleteVMFiles() {
System.out.println("deleting files in vm" + VM.getCurrentVMNum());
File pwd = new File(".");
@@ -145,20 +150,24 @@ public abstract class LuceneSearchWithRollingUpgradeDUnit extends JUnit4Distribu
return props;
}
- VM rollClientToCurrentAndCreateRegion(VM oldClient, ClientRegionShortcut shortcut,
- String regionName, String[] hostNames, int[] locatorPorts, boolean subscriptionEnabled) {
- VM rollClient = rollClientToCurrent(oldClient, hostNames, locatorPorts, subscriptionEnabled);
+ VM rollClientToCurrentAndCreateRegion(VM oldClient,
+ ClientRegionShortcut shortcut,
+ String regionName, String[] hostNames, int[] locatorPorts,
+ boolean subscriptionEnabled, boolean singleHopEnabled) {
+ VM rollClient = rollClientToCurrent(oldClient, hostNames, locatorPorts, subscriptionEnabled,
+ singleHopEnabled);
// recreate region on "rolled" client
invokeRunnableInVMs(invokeCreateClientRegion(regionName, shortcut), rollClient);
return rollClient;
}
- private VM rollClientToCurrent(VM oldClient, String[] hostNames, int[] locatorPorts,
- boolean subscriptionEnabled) {
+ private VM rollClientToCurrent(VM oldClient, String[] hostNames,
+ int[] locatorPorts,
+ boolean subscriptionEnabled, boolean singleHopEnabled) {
oldClient.invoke(invokeCloseCache());
VM rollClient = Host.getHost(0).getVM(VersionManager.CURRENT_VERSION, oldClient.getId());
rollClient.invoke(invokeCreateClientCache(getClientSystemProperties(), hostNames, locatorPorts,
- subscriptionEnabled));
+ subscriptionEnabled, singleHopEnabled));
rollClient.invoke(invokeAssertVersion(Version.CURRENT_ORDINAL));
return rollClient;
}
@@ -203,14 +212,17 @@ public abstract class LuceneSearchWithRollingUpgradeDUnit extends JUnit4Distribu
cacheServer.start();
}
- CacheSerializableRunnable invokeCreateClientCache(final Properties systemProperties,
- final String[] hosts, final int[] ports, boolean subscriptionEnabled) {
+ CacheSerializableRunnable invokeCreateClientCache(
+ final Properties systemProperties,
+ final String[] hosts, final int[] ports, boolean subscriptionEnabled,
+ boolean singleHopEnabled) {
return new CacheSerializableRunnable("execute: createClientCache") {
@Override
public void run2() {
try {
LuceneSearchWithRollingUpgradeDUnit.cache =
- createClientCache(systemProperties, hosts, ports, subscriptionEnabled);
+ createClientCache(systemProperties, hosts, ports, subscriptionEnabled,
+ singleHopEnabled);
} catch (Exception e) {
fail("Error creating client cache", e);
}
@@ -225,13 +237,15 @@ public abstract class LuceneSearchWithRollingUpgradeDUnit extends JUnit4Distribu
}
- private static ClientCache createClientCache(Properties systemProperties, String[] hosts,
- int[] ports, boolean subscriptionEnabled) {
+ private static ClientCache createClientCache(Properties systemProperties,
+ String[] hosts,
+ int[] ports, boolean subscriptionEnabled, boolean singleHopEnabled) {
ClientCacheFactory cf = new ClientCacheFactory(systemProperties);
if (subscriptionEnabled) {
cf.setPoolSubscriptionEnabled(true);
cf.setPoolSubscriptionRedundancy(-1);
}
+ cf.setPoolPRSingleHopEnabled(singleHopEnabled);
int hostsLength = hosts.length;
for (int i = 0; i < hostsLength; i++) {
cf.addPoolLocator(hosts[i], ports[i]);
diff --git a/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOver.java b/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOver.java
index 95c0498..9ef5ca8 100644
--- a/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOver.java
+++ b/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOver.java
@@ -70,7 +70,8 @@ public class RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRol
invokeRunnableInVMs(invokeStartCacheServer(csPorts[1]), server3);
invokeRunnableInVMs(
- invokeCreateClientCache(getClientSystemProperties(), hostNames, locatorPorts, false),
+ invokeCreateClientCache(getClientSystemProperties(), hostNames, locatorPorts, false,
+ singleHopEnabled),
client);
server2.invoke(() -> createLuceneIndex(cache, regionName, INDEX_NAME));
server3.invoke(() -> createLuceneIndex(cache, regionName, INDEX_NAME));
@@ -107,7 +108,7 @@ public class RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRol
60, server3);
client = rollClientToCurrentAndCreateRegion(client, ClientRegionShortcut.PROXY, regionName,
- hostNames, locatorPorts, false);
+ hostNames, locatorPorts, false, singleHopEnabled);
expectedRegionSize += 10;
putSerializableObjectAndVerifyLuceneQueryResult(client, regionName, expectedRegionSize, 60,
70, server2, server3);
diff --git a/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOverAllBucketsCreated.java b/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOverAllBucketsCreated.java
index 3acba6c..1d7670c 100644
--- a/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOverAllBucketsCreated.java
+++ b/geode-lucene/src/upgradeTest/java/org/apache/geode/cache/lucene/RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOverAllBucketsCreated.java
@@ -17,7 +17,6 @@ package org.apache.geode.cache.lucene;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.junit.Assert.assertTrue;
-import org.junit.Ignore;
import org.junit.Test;
import org.apache.geode.cache.RegionShortcut;
@@ -32,7 +31,6 @@ import org.apache.geode.test.dunit.VM;
public class RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRolledOverAllBucketsCreated
extends LuceneSearchWithRollingUpgradeDUnit {
- @Ignore("Disabled until GEODE-7258 is fixed")
@Test
public void test()
throws Exception {
@@ -85,7 +83,8 @@ public class RollingUpgradeQueryReturnsCorrectResultsAfterClientAndServersAreRol
invokeRunnableInVMs(invokeStartCacheServer(csPorts[0]), server1);
invokeRunnableInVMs(invokeStartCacheServer(csPorts[1]), server2);
invokeRunnableInVMs(
- invokeCreateClientCache(getClientSystemProperties(), hostNames, locatorPorts, false),
+ invokeCreateClientCache(getClientSystemProperties(), hostNames, locatorPorts, false,
+ singleHopEnabled),
client);
// Create the index on the servers