You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by kh...@apache.org on 2017/07/18 14:15:27 UTC
[1/9] geode git commit: GEODE-2601: Made banner and configs log only
once and added a test to verify changes
Repository: geode
Updated Branches:
refs/heads/develop 9e7696a64 -> 47c857848
GEODE-2601: Made banner and configs log only once and added a test to verify changes
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/c5520f95
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/c5520f95
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/c5520f95
Branch: refs/heads/develop
Commit: c5520f95635ebc3493aa4246e975f1f7e4aac828
Parents: 0bce1ea
Author: YehEmily <em...@gmail.com>
Authored: Tue Jun 27 12:56:40 2017 -0700
Committer: YehEmily <em...@gmail.com>
Committed: Mon Jul 17 11:31:17 2017 -0700
----------------------------------------------------------------------
.../cli/commands/GfshStartLocatorLogTest.java | 61 ++++++++++++++++++++
.../distributed/internal/InternalLocator.java | 6 --
.../apache/geode/internal/AbstractConfig.java | 18 +++---
.../java/org/apache/geode/internal/Banner.java | 8 +--
.../geode/internal/i18n/LocalizedStrings.java | 2 +-
.../internal/logging/LogWriterFactory.java | 2 +-
.../logging/log4j/LogWriterAppenders.java | 8 +--
.../internal/logging/TestLogWriterFactory.java | 4 +-
.../geode/test/dunit/SerializableCallable.java | 2 +-
9 files changed, 85 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-assembly/src/test/java/org/apache/geode/management/internal/cli/commands/GfshStartLocatorLogTest.java
----------------------------------------------------------------------
diff --git a/geode-assembly/src/test/java/org/apache/geode/management/internal/cli/commands/GfshStartLocatorLogTest.java b/geode-assembly/src/test/java/org/apache/geode/management/internal/cli/commands/GfshStartLocatorLogTest.java
new file mode 100644
index 0000000..eadbea8
--- /dev/null
+++ b/geode-assembly/src/test/java/org/apache/geode/management/internal/cli/commands/GfshStartLocatorLogTest.java
@@ -0,0 +1,61 @@
+/*
+ * 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.management.internal.cli.commands;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+
+import com.google.common.io.Files;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.internal.AbstractConfig;
+import org.apache.geode.internal.Banner;
+import org.apache.geode.internal.i18n.LocalizedStrings;
+import org.apache.geode.internal.logging.log4j.LocalizedMessage;
+import org.apache.geode.test.dunit.rules.gfsh.GfshExecution;
+import org.apache.geode.test.dunit.rules.gfsh.GfshRule;
+import org.apache.geode.test.dunit.rules.gfsh.GfshScript;
+import org.apache.geode.test.junit.categories.DistributedTest;
+
+@Category(DistributedTest.class)
+public class GfshStartLocatorLogTest {
+ @Rule
+ public GfshRule gfshRule = new GfshRule();
+
+ @Test
+ public void bannerOnlyLogsOnce() throws Exception {
+ final String banner = "Licensed to the Apache Software Foundation (ASF)";
+ String lines = getExecutionLogs();
+ assertThat(lines.indexOf(banner)).isEqualTo(lines.lastIndexOf(banner));
+ }
+
+ @Test
+ public void startupConfigsOnlyLogsOnce() throws Exception {
+ final String startupConfigs = AbstractConfig.GEM_FIRE_PROPERTIES_USING_DEFAULT_VALUES;
+ String lines = getExecutionLogs();
+ assertThat(lines.indexOf(startupConfigs)).isEqualTo(lines.lastIndexOf(startupConfigs));
+ }
+
+ private String getExecutionLogs() throws Exception {
+ GfshExecution gfshExecution = GfshScript.of("start locator").execute(gfshRule);
+ File[] files = gfshExecution.getWorkingDir().listFiles();
+ String logName = files[0].getAbsolutePath() + "/" + files[0].getName() + ".log";
+ return Files.readLines(new File(logName), StandardCharsets.UTF_8).toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
index 6eaaec2..ffe3be4 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
@@ -675,7 +675,6 @@ public class InternalLocator extends Locator implements ConnectListener {
System.setProperty(propName, locatorsProp);
}
}
-
// No longer default mcast-port to zero. See 46277.
}
@@ -686,11 +685,6 @@ public class InternalLocator extends Locator implements ConnectListener {
logger.info(
LocalizedMessage.create(LocalizedStrings.InternalLocator_STARTING_DISTRIBUTED_SYSTEM));
- // LOG:CONFIG: changed from config to info
- logger.info(LogMarker.CONFIG,
- LocalizedMessage.create(
- LocalizedStrings.InternalDistributedSystem_STARTUP_CONFIGURATIONN_0,
- this.config.toLoggerString()));
this.myDs = (InternalDistributedSystem) DistributedSystem.connect(connectEnv);
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/main/java/org/apache/geode/internal/AbstractConfig.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/AbstractConfig.java b/geode-core/src/main/java/org/apache/geode/internal/AbstractConfig.java
index 7bb2de9..9b9c0da 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/AbstractConfig.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/AbstractConfig.java
@@ -49,6 +49,12 @@ import org.apache.geode.internal.security.SecurableCommunicationChannel;
*/
public abstract class AbstractConfig implements Config {
+ public static final String GEM_FIRE_PROPERTIES_USING_DEFAULT_VALUES =
+ "### GemFire Properties using default values ###";
+ public static final String GEM_FIRE_PROPERTIES_DEFINED_WITH_PREFIX =
+ "### GemFire Properties defined with ";
+ public static final String GEM_FIRE_PROPERTIES_DEFINED_WITH_SUFFIX = " ###";
+
/**
* Returns the string to use as the exception message when an attempt is made to set an
* unmodifiable attribute.
@@ -173,9 +179,10 @@ public abstract class AbstractConfig implements Config {
if (!sourceFound) {
sourceFound = true;
if (source == null) {
- pw.println("### GemFire Properties using default values ###");
+ pw.println(GEM_FIRE_PROPERTIES_USING_DEFAULT_VALUES);
} else {
- pw.println("### GemFire Properties defined with " + source.getDescription() + " ###");
+ pw.println(GEM_FIRE_PROPERTIES_DEFINED_WITH_PREFIX + source.getDescription()
+ + GEM_FIRE_PROPERTIES_DEFINED_WITH_SUFFIX);
}
}
// hide the shiro-init configuration for now. Remove after we can allow customer to specify
@@ -205,10 +212,7 @@ public abstract class AbstractConfig implements Config {
if (attName.startsWith(DistributionConfig.SYS_PROP_NAME)) {
return false;
}
- if (attName.toLowerCase().contains("password") /* bug 45381 */) {
- return false;
- }
- return true;
+ return !attName.toLowerCase().contains("password");
}
/**
@@ -388,7 +392,7 @@ public abstract class AbstractConfig implements Config {
String trimAttName = trimAttributeName(attName);
throw new UnmodifiableException(
LocalizedStrings.AbstractConfig_THE_0_CONFIGURATION_ATTRIBUTE_CAN_NOT_BE_SET_FROM_THE_COMMAND_LINE_SET_1_FOR_EACH_INDIVIDUAL_PARAMETER_INSTEAD
- .toLocalizedString(new Object[] {attName, trimAttName}));
+ .toLocalizedString(attName, trimAttName));
}
} else if (valueType.equals(FlowControlParams.class)) {
String values[] = attValue.split(",");
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/main/java/org/apache/geode/internal/Banner.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/Banner.java b/geode-core/src/main/java/org/apache/geode/internal/Banner.java
index a218a5b..38b3a31 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/Banner.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/Banner.java
@@ -59,7 +59,7 @@ public class Banner {
static void print(PrintWriter out, String args[]) {
Map sp = new TreeMap((Properties) System.getProperties().clone()); // fix for 46822
int processId = -1;
- final String SEPERATOR =
+ final String SEPARATOR =
"---------------------------------------------------------------------------";
try {
processId = OSProcess.getId();
@@ -80,7 +80,7 @@ public class Banner {
final String productName = GemFireVersion.getProductName();
- out.println(SEPERATOR);
+ out.println(SEPARATOR);
out.println(" ");
out.println(" Licensed to the Apache Software Foundation (ASF) under one or more");
@@ -100,7 +100,7 @@ public class Banner {
out.println(" under the License.");
out.println(" ");
- out.println(SEPERATOR);
+ out.println(SEPARATOR);
GemFireVersion.print(out);
@@ -153,7 +153,7 @@ public class Banner {
out.println("Log4J 2 Configuration:");
out.println(" " + LogService.getConfigInformation());
}
- out.println(SEPERATOR);
+ out.println(SEPARATOR);
}
/**
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java b/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
index baad039..3a32db8 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/i18n/LocalizedStrings.java
@@ -1112,7 +1112,7 @@ public class LocalizedStrings {
new StringId(1690, "Running in local mode since no locators were specified.");
public static final StringId InternalDistributedSystem_SHUTDOWNLISTENER__0__THREW =
new StringId(1691, "ShutdownListener < {0} > threw...");
- public static final StringId InternalDistributedSystem_STARTUP_CONFIGURATIONN_0 =
+ public static final StringId InternalDistributedSystem_STARTUP_CONFIGURATION_0 =
new StringId(1692, "Startup Configuration:\n {0}");
public static final StringId InternalInstantiator_CLASS_0_DOES_NOT_HAVE_A_TWOARGUMENT_CLASS_INT_CONSTRUCTOR =
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/main/java/org/apache/geode/internal/logging/LogWriterFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/logging/LogWriterFactory.java b/geode-core/src/main/java/org/apache/geode/internal/logging/LogWriterFactory.java
index 4202afd..e8c283c 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/logging/LogWriterFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/logging/LogWriterFactory.java
@@ -95,7 +95,7 @@ public class LogWriterFactory {
// LOG:CONFIG: changed from config to info
logger.info(LogMarker.CONFIG,
LocalizedMessage.create(
- LocalizedStrings.InternalDistributedSystem_STARTUP_CONFIGURATIONN_0,
+ LocalizedStrings.InternalDistributedSystem_STARTUP_CONFIGURATION_0,
config.toLoggerString()));
}
return logger;
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/main/java/org/apache/geode/internal/logging/log4j/LogWriterAppenders.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/logging/log4j/LogWriterAppenders.java b/geode-core/src/main/java/org/apache/geode/internal/logging/log4j/LogWriterAppenders.java
index 310deb2..a500654 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/logging/log4j/LogWriterAppenders.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/logging/log4j/LogWriterAppenders.java
@@ -40,14 +40,14 @@ public class LogWriterAppenders {
MAIN(false), SECURITY(true);
private final boolean isSecure;
- private Identifier(final boolean isSecure) {
+ Identifier(final boolean isSecure) {
this.isSecure = isSecure;
}
public boolean isSecure() {
return this.isSecure;
}
- };
+ }
private static Map<Identifier, LogWriterAppender> appenders =
new HashMap<Identifier, LogWriterAppender>();
@@ -151,7 +151,7 @@ public class LogWriterAppenders {
} else {
firstMsgWarning = true;
firstMsg = LocalizedStrings.InternalDistributedSystem_COULD_NOT_RENAME_0_TO_1
- .toLocalizedString(new Object[] {logFile, oldMain});
+ .toLocalizedString(logFile, oldMain);
}
}
}
@@ -220,7 +220,7 @@ public class LogWriterAppenders {
if (logConfig) {
if (!isLoner) {
// LOG:CONFIG: changed from config to info
- logWriter.info(LocalizedStrings.InternalDistributedSystem_STARTUP_CONFIGURATIONN_0,
+ logWriter.info(LocalizedStrings.InternalDistributedSystem_STARTUP_CONFIGURATION_0,
config.toLoggerString());
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/test/java/org/apache/geode/internal/logging/TestLogWriterFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/logging/TestLogWriterFactory.java b/geode-core/src/test/java/org/apache/geode/internal/logging/TestLogWriterFactory.java
index 87f370c..17fdd05 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/logging/TestLogWriterFactory.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/logging/TestLogWriterFactory.java
@@ -71,7 +71,7 @@ public class TestLogWriterFactory extends Assert {
} else {
firstMsgWarning = true;
firstMsg = LocalizedStrings.InternalDistributedSystem_COULD_NOT_RENAME_0_TO_1
- .toLocalizedString(new Object[] {logFile, oldMain});
+ .toLocalizedString(logFile, oldMain);
}
}
}
@@ -114,7 +114,7 @@ public class TestLogWriterFactory extends Assert {
}
if (logConfig && logger.configEnabled()) {
logger.convertToLogWriterI18n().config(
- LocalizedStrings.InternalDistributedSystem_STARTUP_CONFIGURATIONN_0,
+ LocalizedStrings.InternalDistributedSystem_STARTUP_CONFIGURATION_0,
config.toLoggerString());
}
http://git-wip-us.apache.org/repos/asf/geode/blob/c5520f95/geode-core/src/test/java/org/apache/geode/test/dunit/SerializableCallable.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/SerializableCallable.java b/geode-core/src/test/java/org/apache/geode/test/dunit/SerializableCallable.java
index 68d5229..29b083a 100644
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/SerializableCallable.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/SerializableCallable.java
@@ -22,7 +22,7 @@ import java.util.concurrent.Callable;
* conjunction with {@link VM#invoke(SerializableCallableIF)}.
*
* <PRE>
- * public void testRepilcatedRegionPut() {
+ * public void testReplicatedRegionPut() {
* final Host host = Host.getHost(0);
* VM vm0 = host.getVM(0);
* VM vm1 = host.getVM(1);
[9/9] geode git commit: Merge branch 'feature/GEODE-2601' into develop
Posted by kh...@apache.org.
Merge branch 'feature/GEODE-2601' into develop
This closes #606
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/47c85784
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/47c85784
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/47c85784
Branch: refs/heads/develop
Commit: 47c857848a29370205158d6efef13bce8c0c63f4
Parents: 9e7696a 3f9be9a
Author: Ken Howe <kh...@pivotal.io>
Authored: Tue Jul 18 07:13:32 2017 -0700
Committer: Ken Howe <kh...@pivotal.io>
Committed: Tue Jul 18 07:13:32 2017 -0700
----------------------------------------------------------------------
.../cli/commands/GfshStartLocatorLogTest.java | 61 ++++++++++++++++++++
.../distributed/internal/InternalLocator.java | 6 --
.../apache/geode/internal/AbstractConfig.java | 18 +++---
.../java/org/apache/geode/internal/Banner.java | 8 +--
.../geode/internal/i18n/LocalizedStrings.java | 2 +-
.../internal/logging/LogWriterFactory.java | 2 +-
.../logging/log4j/LogWriterAppenders.java | 8 +--
.../internal/logging/TestLogWriterFactory.java | 4 +-
.../geode/test/dunit/SerializableCallable.java | 2 +-
9 files changed, 85 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
[6/9] geode git commit: GEODE-3166: use the 3 param getCredential
interface.
Posted by kh...@apache.org.
GEODE-3166: use the 3 param getCredential interface.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/d02bf864
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/d02bf864
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/d02bf864
Branch: refs/heads/develop
Commit: d02bf86422e456d483cc5eb3fd3364b03cb2f0a0
Parents: 87b56a9
Author: Jinmei Liao <ji...@pivotal.io>
Authored: Wed Jul 5 08:59:59 2017 -0700
Committer: Ken Howe <kh...@pivotal.io>
Committed: Tue Jul 18 07:11:33 2017 -0700
----------------------------------------------------------------------
.../org/apache/geode/security/AuthInitialize.java | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/d02bf864/geode-core/src/main/java/org/apache/geode/security/AuthInitialize.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/security/AuthInitialize.java b/geode-core/src/main/java/org/apache/geode/security/AuthInitialize.java
index 6b0675a..0efd958 100644
--- a/geode-core/src/main/java/org/apache/geode/security/AuthInitialize.java
+++ b/geode-core/src/main/java/org/apache/geode/security/AuthInitialize.java
@@ -82,19 +82,19 @@ public interface AuthInitialize extends CacheCallback {
*
* @return the credentials to be used for the given <code>server</code>
*
- * @deprecated since Geode 1.0, use getCredentials(Properties). When using Integrated security,
- * all members, peer/client will use the same credentials.
+ * When using Integrated security, all members, peer/client will use the same credentials.
+ * but we still need to use these params to support the old authenticator
*/
- @Deprecated
- public Properties getCredentials(Properties securityProps, DistributedMember server,
- boolean isPeer) throws AuthenticationFailedException;
+ Properties getCredentials(Properties securityProps, DistributedMember server, boolean isPeer)
+ throws AuthenticationFailedException;
/**
- * Implement this since Geode1.0
- *
+ *
* @param securityProps
* @return the credentials to be used. It needs to contain "security-username" and
* "security-password"
+ * @deprecated As of Geode 1.3, please implement getCredentials(Properties, DistributedMember,
+ * boolean)
*/
default Properties getCredentials(Properties securityProps) {
return getCredentials(securityProps, null, true);
[8/9] geode git commit: GEODE-3207 Swagger library updates: update
user guide
Posted by kh...@apache.org.
GEODE-3207 Swagger library updates: update user guide
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/87b56a91
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/87b56a91
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/87b56a91
Branch: refs/heads/develop
Commit: 87b56a911b21f80d9175e2f1343214699be28948
Parents: e1840b1
Author: Dave Barnes <db...@pivotal.io>
Authored: Fri Jul 14 12:51:52 2017 -0700
Committer: Ken Howe <kh...@pivotal.io>
Committed: Tue Jul 18 07:11:33 2017 -0700
----------------------------------------------------------------------
geode-docs/rest_apps/setup_config.html.md.erb | 9 +++++----
geode-docs/rest_apps/using_swagger.html.md.erb | 4 ++--
2 files changed, 7 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/87b56a91/geode-docs/rest_apps/setup_config.html.md.erb
----------------------------------------------------------------------
diff --git a/geode-docs/rest_apps/setup_config.html.md.erb b/geode-docs/rest_apps/setup_config.html.md.erb
index c557f4f..7ba24ea 100644
--- a/geode-docs/rest_apps/setup_config.html.md.erb
+++ b/geode-docs/rest_apps/setup_config.html.md.erb
@@ -36,8 +36,8 @@ where _install-dir_ is the server installation directory and _n.n.n_ is a versio
The REST API service for application development runs only on servers; you cannot run the service on a locator.
-To enable the Developer REST API service on a given server, set the `start-dev-rest-api` property
-to `true` for the server, using either the `gfsh start server` command or the ServerLauncher API.
+To enable the Developer REST API service on a given server, use the `gfsh start server` command with the `--start-rest-api` option,
+or set the `start-dev-rest-api` property to `true` for the server via the ServerLauncher API.
This starts an embedded Jetty server and deploys the Developer REST API WAR file on that server.
## Enabling the REST API on Multiple Servers
@@ -103,8 +103,9 @@ To configure PDX in your cluster, perform the following steps:
## Start the REST API Service on One or More Servers
-As described above, you can start the REST API service on a server by setting the Geode property `start-dev-rest-api` set to `true`.
-If you wish to start the service on multiple servers, use a `http-service-bind-address` and `http-service-port` to
+As described above, you can start the REST API service on a server by using `gfsh start server --start-rest-api`,
+or by setting the Geode property `start-dev-rest-api` to `true`.
+If you wish to start the service on multiple servers, use `http-service-bind-address` and `http-service-port` to
identify the cache server and specific port that will host REST services. If you do not specify
the `http-service-port`, the default port is 7070, which may collide with other locators and servers.
If you do not specify `http-service-bind-address`, the HTTP service will bind to all local addresses by default.
http://git-wip-us.apache.org/repos/asf/geode/blob/87b56a91/geode-docs/rest_apps/using_swagger.html.md.erb
----------------------------------------------------------------------
diff --git a/geode-docs/rest_apps/using_swagger.html.md.erb b/geode-docs/rest_apps/using_swagger.html.md.erb
index ab0d84b..995f33a 100644
--- a/geode-docs/rest_apps/using_swagger.html.md.erb
+++ b/geode-docs/rest_apps/using_swagger.html.md.erb
@@ -32,8 +32,8 @@ Specify an `http-service-port` for the developer REST service, as the default po
gfsh>start locator --name=locator1
Starting a Geode Locator in /Users/admin/apache-geode-1.2.0/locator1...
....
- gfsh>start server --name=server1 --J=-Dgemfire.start-dev-rest-api=true \
- --J=-Dgemfire.http-service-bind-address=localhost --J=-Dgemfire.http-service-port=8080
+ gfsh>start server --name=server1 --start-rest-api=true \
+ --http-service-bind-address=localhost --J=-Dgemfire.http-service-port=8080
```
2. To access Swagger, open a browser and enter the following URL. For example:
[3/9] geode git commit: GEODE-2860: Refactor use of EventTracker
Posted by kh...@apache.org.
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventSequenceNumberHolder.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventSequenceNumberHolder.java b/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventSequenceNumberHolder.java
new file mode 100644
index 0000000..9bb450d
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventSequenceNumberHolder.java
@@ -0,0 +1,124 @@
+/*
+ * 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.event;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.geode.DataSerializable;
+import org.apache.geode.DataSerializer;
+import org.apache.geode.internal.cache.versions.VersionTag;
+
+/**
+ * A sequence number tracker to keep events from clients from being re-applied to the cache if
+ * they've already been seen.
+ *
+ * @since GemFire 5.5
+ */
+public class EventSequenceNumberHolder implements DataSerializable {
+ private static final long serialVersionUID = 8137262960763308046L;
+
+ /**
+ * event sequence number. These
+ */
+ private long lastSequenceNumber = -1;
+
+ /**
+ * millisecond timestamp
+ */
+ private transient long endOfLifeTimestamp;
+
+ /**
+ * whether this entry is being removed
+ */
+ private transient boolean removed;
+
+ /**
+ * version tag, if any, for the operation
+ */
+ private VersionTag versionTag;
+
+ // for debugging
+ // transient Exception context;
+
+ EventSequenceNumberHolder(long id, VersionTag versionTag) {
+ this.lastSequenceNumber = id;
+ this.versionTag = versionTag;
+ }
+
+ public EventSequenceNumberHolder() {}
+
+ public long getLastSequenceNumber() {
+ return lastSequenceNumber;
+ }
+
+ public VersionTag getVersionTag() {
+ return versionTag;
+ }
+
+ public boolean isRemoved() {
+ return removed;
+ }
+
+ void setRemoved(boolean removed) {
+ this.removed = removed;
+ }
+
+ void setEndOfLifeTimestamp(long endOfLifeTimestamp) {
+ this.endOfLifeTimestamp = endOfLifeTimestamp;
+ }
+
+ void setVersionTag(VersionTag versionTag) {
+ this.versionTag = versionTag;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder result = new StringBuilder();
+ result.append("seqNo").append(this.lastSequenceNumber);
+ if (this.versionTag != null) {
+ result.append(",").append(this.versionTag);
+ }
+ return result.toString();
+ }
+
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ lastSequenceNumber = in.readLong();
+ versionTag = (VersionTag) DataSerializer.readObject(in);
+ }
+
+ public void toData(DataOutput out) throws IOException {
+ out.writeLong(lastSequenceNumber);
+ DataSerializer.writeObject(versionTag, out);
+ }
+
+ public synchronized boolean expire(long now, long expirationTime) {
+ if (endOfLifeTimestamp == 0) {
+ endOfLifeTimestamp = now; // a new holder - start the timer
+ }
+ boolean expire = false;
+ if (endOfLifeTimestamp <= expirationTime) {
+ removed = true;
+ lastSequenceNumber = -1;
+ expire = true;
+ }
+ return expire;
+ }
+
+ public void setLastSequenceNumber(long lastSequenceNumber) {
+ this.lastSequenceNumber = lastSequenceNumber;
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTracker.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTracker.java b/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTracker.java
new file mode 100644
index 0000000..43a0458
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTracker.java
@@ -0,0 +1,136 @@
+/*
+ * 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.event;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+import org.apache.geode.internal.cache.versions.VersionTag;
+
+/**
+ * EventTracker tracks the last sequence number for a particular memberID:threadID. It is used to
+ * avoid replaying events in client/server and partitioned-region configurations.
+ *
+ * @since GemFire 6.0
+ */
+public interface EventTracker {
+ /** start this event tracker */
+ void start();
+
+ /** stop this event tracker */
+ void stop();
+
+ /**
+ * retrieve a deep copy of the state of the event tracker. Synchronization is not used while
+ * copying the tracker's state.
+ */
+ Map<ThreadIdentifier, EventSequenceNumberHolder> getState();
+
+ /**
+ * record the given state in the tracker.
+ *
+ * @param provider the member that provided this state
+ * @param state a Map obtained from getState();
+ */
+ void recordState(InternalDistributedMember provider,
+ Map<ThreadIdentifier, EventSequenceNumberHolder> state);
+
+ /**
+ * Use this method to ensure that the tracker is put in an initialized state
+ */
+ void setInitialized();
+
+ /**
+ * Wait for the tracker to finish being initialized
+ */
+ void waitOnInitialization() throws InterruptedException;
+
+ /** record the event's threadid/sequenceid to prevent replay */
+ void recordEvent(InternalCacheEvent event);
+
+ /**
+ * Determines if an event has already been seen by the tracker
+ *
+ * @param event The event to determine if it has been seen by the tracker already
+ * @return if the event provided has already been seen
+ */
+ boolean hasSeenEvent(InternalCacheEvent event);
+
+ /**
+ * Determines if an event has already been seen by the tracker
+ *
+ * @param eventID The id of the event to determine if it has been seen by the tracker already
+ * @return if the event provided has already been seen
+ */
+ boolean hasSeenEvent(EventID eventID);
+
+ /**
+ * Determines if an event has already been seen by the tracker
+ *
+ * @param eventID The id of the event to determine if it has been seen by the tracker already
+ * @param tagHolder Event to update version tag with that of eventID, if event was seen before
+ * @return if the event provided has already been seen
+ */
+ boolean hasSeenEvent(EventID eventID, InternalCacheEvent tagHolder);
+
+ VersionTag findVersionTagForSequence(EventID eventID);
+
+ VersionTag findVersionTagForBulkOp(EventID eventID);
+
+ /**
+ * The name of the event tracker for logging purposes
+ *
+ * @return the name of the tracker
+ */
+ String getName();
+
+ ConcurrentMap<ThreadIdentifier, BulkOperationHolder> getRecordedBulkOpVersionTags();
+
+ ConcurrentMap<ThreadIdentifier, EventSequenceNumberHolder> getRecordedEvents();
+
+ /**
+ * A routine to provide synchronization running based on <memberShipID, threadID> of the
+ * requesting client
+ *
+ * @param r - a Runnable to wrap the processing of the bulk op
+ * @param eventID - the base event ID of the bulk op
+ *
+ * @since GemFire 5.7
+ */
+ void syncBulkOp(Runnable r, EventID eventID, boolean partOfTransaction);
+
+ /**
+ * Called when a new bulkOp is started on the local region. Used to clear event tracker state from
+ * the last bulkOp.
+ */
+ void recordBulkOpStart(EventID eventID, ThreadIdentifier tid);
+
+ /**
+ * @return the initialization state of the tracker
+ */
+ boolean isInitialized();
+
+ /**
+ * @param mbr the member in question
+ * @return true if the given member provided the initial image event state for this tracker
+ */
+ boolean isInitialImageProvider(DistributedMember mbr);
+
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTask.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTask.java b/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTask.java
new file mode 100644
index 0000000..dff80df
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTask.java
@@ -0,0 +1,97 @@
+/*
+ * 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.event;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.geode.internal.SystemTimer;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+
+public class EventTrackerExpiryTask extends SystemTimer.SystemTimerTask {
+
+ private final long lifetimeInMillis;
+ private final List<EventTracker> trackers = new LinkedList<>();
+ private final boolean traceEnabled = logger.isTraceEnabled();
+
+ public EventTrackerExpiryTask(long lifetimeInMillis) {
+ this.lifetimeInMillis = lifetimeInMillis;
+ }
+
+ void addTracker(EventTracker tracker) {
+ synchronized (trackers) {
+ trackers.add(tracker);
+ }
+ }
+
+ void removeTracker(EventTracker tracker) {
+ synchronized (trackers) {
+ trackers.remove(tracker);
+ }
+ }
+
+ int getNumberOfTrackers() {
+ return trackers.size();
+ }
+
+ @Override
+ public void run2() {
+ long now = System.currentTimeMillis();
+ long expirationTime = now - lifetimeInMillis;
+ synchronized (trackers) {
+ for (EventTracker tracker : trackers) {
+ if (traceEnabled) {
+ logger.trace("{} sweeper: starting", tracker.getName());
+ }
+ removeExpiredSequenceTracker(tracker, now, expirationTime);
+ removeExpiredBulkOperations(tracker, now, expirationTime);
+
+ if (traceEnabled) {
+ logger.trace("{} sweeper: done", tracker.getName());
+ }
+ }
+ }
+ }
+
+ private void removeExpiredSequenceTracker(EventTracker tracker, long now, long expirationTime) {
+ for (Iterator<Map.Entry<ThreadIdentifier, EventSequenceNumberHolder>> entryIterator =
+ tracker.getRecordedEvents().entrySet().iterator(); entryIterator.hasNext();) {
+ Map.Entry<ThreadIdentifier, EventSequenceNumberHolder> entry = entryIterator.next();
+ EventSequenceNumberHolder evh = entry.getValue();
+ if (evh.expire(now, expirationTime)) {
+ if (traceEnabled) {
+ logger.trace("{} sweeper: removing {}", tracker.getName(), entry.getKey());
+ }
+ entryIterator.remove();
+ }
+ }
+ }
+
+ private void removeExpiredBulkOperations(EventTracker tracker, long now, long expirationTime) {
+ for (Iterator<Map.Entry<ThreadIdentifier, BulkOperationHolder>> entryIterator =
+ tracker.getRecordedBulkOpVersionTags().entrySet().iterator(); entryIterator.hasNext();) {
+ Map.Entry<ThreadIdentifier, BulkOperationHolder> entry = entryIterator.next();
+ BulkOperationHolder evh = entry.getValue();
+ if (evh.expire(now, expirationTime)) {
+ if (traceEnabled) {
+ logger.trace("{} sweeper: removing bulkOp {}", tracker.getName(), entry.getKey());
+ }
+ entryIterator.remove();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/event/NonDistributedEventTracker.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/event/NonDistributedEventTracker.java b/geode-core/src/main/java/org/apache/geode/internal/cache/event/NonDistributedEventTracker.java
new file mode 100644
index 0000000..f5b1831
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/event/NonDistributedEventTracker.java
@@ -0,0 +1,135 @@
+/*
+ * 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.event;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+import org.apache.geode.internal.cache.versions.VersionTag;
+
+public class NonDistributedEventTracker implements EventTracker {
+ private static final NonDistributedEventTracker INSTANCE = new NonDistributedEventTracker();
+
+ static final String NAME = "The NonDistributedEventTracker";
+
+ public static NonDistributedEventTracker getInstance() {
+ return INSTANCE;
+ }
+
+ private NonDistributedEventTracker() {
+ // private no arg constructor to enforce singleton pattern
+ }
+
+ @Override
+ public void start() {
+
+ }
+
+ @Override
+ public void stop() {
+
+ }
+
+ @Override
+ public Map<ThreadIdentifier, EventSequenceNumberHolder> getState() {
+ return null;
+ }
+
+ @Override
+ public void recordState(InternalDistributedMember provider, Map state) {
+
+ }
+
+ @Override
+ public void recordEvent(InternalCacheEvent event) {
+
+ }
+
+ @Override
+ public boolean hasSeenEvent(InternalCacheEvent event) {
+ return false;
+ }
+
+ @Override
+ public void waitOnInitialization() throws InterruptedException {
+
+ }
+
+ @Override
+ public VersionTag findVersionTagForSequence(EventID eventId) {
+ return null;
+ }
+
+ @Override
+ public VersionTag findVersionTagForBulkOp(EventID eventId) {
+ return null;
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public boolean hasSeenEvent(EventID eventID) {
+ return false;
+ }
+
+ @Override
+ public boolean hasSeenEvent(EventID eventID, InternalCacheEvent tagHolder) {
+ return false;
+ }
+
+ @Override
+ public void recordBulkOpStart(EventID eventID, ThreadIdentifier membershipID) {
+
+ }
+
+ @Override
+ public void syncBulkOp(Runnable task, EventID eventId, boolean partOfTransaction) {
+ task.run();
+ }
+
+ @Override
+ public boolean isInitialized() {
+ return true;
+ }
+
+ @Override
+ public void setInitialized() {
+
+ }
+
+ @Override
+ public boolean isInitialImageProvider(DistributedMember mbr) {
+ return false;
+ }
+
+ @Override
+ public ConcurrentMap<ThreadIdentifier, BulkOperationHolder> getRecordedBulkOpVersionTags() {
+ return null;
+ }
+
+ @Override
+ public ConcurrentMap<ThreadIdentifier, EventSequenceNumberHolder> getRecordedEvents() {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java
index cc78cca..1f47897 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java
@@ -211,9 +211,9 @@ public abstract class BaseCommand implements Command {
LocalRegion r = clientEvent.getRegion();
VersionTag tag;
if (clientEvent.getVersionTag() != null && clientEvent.getVersionTag().isGatewayTag()) {
- tag = r.findVersionTagForGatewayEvent(clientEvent.getEventId());
+ tag = r.findVersionTagForEvent(clientEvent.getEventId());
} else {
- tag = r.findVersionTagForClientEvent(clientEvent.getEventId());
+ tag = r.findVersionTagForEvent(clientEvent.getEventId());
}
if (tag == null) {
if (r instanceof DistributedRegion || r instanceof PartitionedRegion) {
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
index 6fa17a4..9bad91a 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
@@ -52,11 +52,13 @@ import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.CachedDeserializable;
import org.apache.geode.internal.cache.Conflatable;
+import org.apache.geode.internal.cache.event.EventTracker;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.event.NonDistributedEventTracker;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
@@ -1188,7 +1190,9 @@ public class SerialGatewaySenderQueue implements RegionQueue {
// @override event tracker not needed for this type of region
@Override
- public void createEventTracker() {}
+ public EventTracker createEventTracker() {
+ return NonDistributedEventTracker.getInstance();
+ }
@Override
protected boolean shouldNotifyBridgeClients() {
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
index 61f91a0..db5f7ca 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
@@ -88,7 +88,7 @@ import org.apache.geode.internal.cache.DiskStoreFactoryImpl;
import org.apache.geode.internal.cache.DiskStoreImpl;
import org.apache.geode.internal.cache.DiskStoreMonitor;
import org.apache.geode.internal.cache.DistributedRegion;
-import org.apache.geode.internal.cache.EventTracker.ExpiryTask;
+import org.apache.geode.internal.cache.event.EventTrackerExpiryTask;
import org.apache.geode.internal.cache.ExpirationScheduler;
import org.apache.geode.internal.cache.FilterProfile;
import org.apache.geode.internal.cache.GemFireCacheImpl;
@@ -1670,7 +1670,7 @@ public class CacheCreation implements InternalCache {
}
@Override
- public ExpiryTask getEventTrackerTask() {
+ public EventTrackerExpiryTask getEventTrackerTask() {
throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRegionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRegionJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRegionJUnitTest.java
index d76ff1b..eaaa3aa 100755
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRegionJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRegionJUnitTest.java
@@ -20,6 +20,7 @@ import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.*;
+import java.io.IOException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -49,6 +50,9 @@ public class BucketRegionJUnitTest extends DistributedRegionJUnitTest {
protected DistributedRegion createAndDefineRegion(boolean isConcurrencyChecksEnabled,
RegionAttributes ra, InternalRegionArguments ira, GemFireCacheImpl cache) {
BucketRegion br = new BucketRegion("testRegion", ra, null, cache, ira);
+ // it is necessary to set the event tracker to initialized, since initialize() in not being
+ // called on the instantiated region
+ br.getEventTracker().setInitialized();
// since br is a real bucket region object, we need to tell mockito to monitor it
br = spy(br);
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedRegionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedRegionJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedRegionJUnitTest.java
index ce21c67..78cdd84 100755
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedRegionJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedRegionJUnitTest.java
@@ -27,7 +27,8 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.internal.cache.EventTracker.BulkOpHolder;
+import org.apache.geode.internal.cache.event.BulkOperationHolder;
+import org.apache.geode.internal.cache.event.EventTracker;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.versions.VersionTag;
@@ -112,7 +113,6 @@ public class DistributedRegionJUnitTest extends AbstractDistributedRegionJUnitTe
DistributedRegion region = prepare(true, true);
DistributedMember member = mock(DistributedMember.class);
ClientProxyMembershipID memberId = mock(ClientProxyMembershipID.class);
- doReturn(false).when(region).isUsedForPartitionedRegionBucket();
byte[] memId = {1, 2, 3};
long threadId = 1;
@@ -124,8 +124,9 @@ public class DistributedRegionJUnitTest extends AbstractDistributedRegionJUnitTe
recordPutAllEvents(region, memId, threadId, skipCallbacks, member, memberId, size);
EventTracker eventTracker = region.getEventTracker();
- ConcurrentMap<ThreadIdentifier, BulkOpHolder> map = eventTracker.getRecordedBulkOpVersionTags();
- BulkOpHolder holder = map.get(tid);
+ ConcurrentMap<ThreadIdentifier, BulkOperationHolder> map =
+ eventTracker.getRecordedBulkOpVersionTags();
+ BulkOperationHolder holder = map.get(tid);
EntryEventImpl retryEvent = EntryEventImpl.create(region, Operation.PUTALL_CREATE, "key1",
"value1", null, false, member, !skipCallbacks, retryEventID);
@@ -133,7 +134,7 @@ public class DistributedRegionJUnitTest extends AbstractDistributedRegionJUnitTe
retryEvent.setPutAllOperation(mock(DistributedPutAllOperation.class));
region.hasSeenEvent(retryEvent);
- assertTrue(retryEvent.getVersionTag().equals(holder.entryVersionTags.get(retryEventID)));
+ assertTrue(retryEvent.getVersionTag().equals(holder.getEntryVersionTags().get(retryEventID)));
}
protected void recordPutAllEvents(DistributedRegion region, byte[] memId, long threadId,
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerDUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerDUnitTest.java
deleted file mode 100755
index 77c0998..0000000
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerDUnitTest.java
+++ /dev/null
@@ -1,486 +0,0 @@
-/*
- * 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;
-
-import static org.junit.Assert.*;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
-
-import org.awaitility.Awaitility;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-import org.apache.geode.cache.AttributesFactory;
-import org.apache.geode.cache.Cache;
-import org.apache.geode.cache.CacheException;
-import org.apache.geode.cache.PartitionAttributesFactory;
-import org.apache.geode.cache.Region;
-import org.apache.geode.cache.RegionFactory;
-import org.apache.geode.cache.RegionShortcut;
-import org.apache.geode.cache.Scope;
-import org.apache.geode.cache.server.CacheServer;
-import org.apache.geode.cache30.CacheSerializableRunnable;
-import org.apache.geode.cache30.ClientServerTestCase;
-import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.internal.cache.EventTracker.BulkOpHolder;
-import org.apache.geode.internal.cache.ha.ThreadIdentifier;
-import org.apache.geode.test.dunit.Assert;
-import org.apache.geode.test.dunit.Host;
-import org.apache.geode.test.dunit.NetworkUtils;
-import org.apache.geode.test.dunit.SerializableRunnable;
-import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.dunit.Wait;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
-
-/**
- * Tests <code>EventTracker</code> management.
- *
- * @since GemFire 6.5
- */
-@Category(DistributedTest.class)
-public class EventTrackerDUnitTest extends JUnit4CacheTestCase {
-
- /** The port on which the <code>CacheServer</code> was started in this VM */
- private static int cacheServerPort;
-
- /** The <code>Cache</code>'s <code>ExpiryTask</code>'s ping interval */
- private static final String MESSAGE_TRACKING_TIMEOUT = "5000";
-
- @Override
- public final void postTearDownCacheTestCase() throws Exception {
- disconnectAllFromDS();
- }
-
- /**
- * Tests <code>EventTracker</code> is created and destroyed when a <code>Region</code> is created
- * and destroyed.
- */
- @Test
- public void testEventTrackerCreateDestroy() throws CacheException {
- // Verify the Cache's ExpiryTask contains no EventTrackers
- GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
- EventTracker.ExpiryTask expiryTask = cache.getEventTrackerTask();
- assertNotNull(expiryTask);
-
- // We start with 3 event trackers:
- // one for the PDX registry region
- // one for ManagementConstants.MONITORING_REGION
- // one for ManagementConstants.NOTIFICATION_REGION
- final int EXPECTED_TRACKERS = 3;
- assertEquals(EXPECTED_TRACKERS, expiryTask.getNumberOfTrackers());
-
- // Create a distributed Region
- AttributesFactory factory = new AttributesFactory();
- factory.setScope(Scope.DISTRIBUTED_ACK);
- LocalRegion region = (LocalRegion) createRegion(getName(), factory.create());
-
- // Verify an EventTracker is created and is empty
- EventTracker eventTracker = region.getEventTracker();
- assertNotNull(eventTracker);
- Map eventState = region.getEventState();
- assertNotNull(eventState);
- assertEquals(0, eventState.size());
-
- // Verify it and the root region's EventTracker are added to the Cache's ExpiryTask's trackers
- assertEquals(EXPECTED_TRACKERS + 2, expiryTask.getNumberOfTrackers());
-
- // Destroy the Region
- region.destroyRegion();
-
- // Verify the EventTracker is removed from the Cache's ExpiryTask's trackers
- assertEquals(EXPECTED_TRACKERS + 1, expiryTask.getNumberOfTrackers());
- }
-
- /**
- * Tests adding threads to an <code>EventTracker</code>.
- */
- @Test
- public void testEventTrackerAddThreadIdentifier() throws CacheException {
- Host host = Host.getHost(0);
- VM serverVM = host.getVM(0);
- VM clientVM = host.getVM(1);
- final String regionName = getName();
-
- // Create Region in the server and verify tracker is created
- serverVM.invoke(new CacheSerializableRunnable("Create server") {
- public void run2() throws CacheException {
- // Create a distributed Region
- AttributesFactory factory = new AttributesFactory();
- factory.setScope(Scope.DISTRIBUTED_ACK);
- LocalRegion region = (LocalRegion) createRegion(regionName, factory.create());
-
- // Verify an EventTracker is created
- EventTracker eventTracker = region.getEventTracker();
- assertNotNull(eventTracker);
- try {
- startCacheServer();
- } catch (Exception ex) {
- Assert.fail("While starting CacheServer", ex);
- }
- }
- });
-
- // Verify tracker in server contains no entries
- serverVM.invoke(new CacheSerializableRunnable("Do puts") {
- public void run2() throws CacheException {
- LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
- Map eventState = region.getEventState();
- assertEquals(0, eventState.size());
- }
- });
-
- // Create Create Region in the client
- final int port = serverVM.invoke(() -> EventTrackerDUnitTest.getCacheServerPort());
- final String hostName = NetworkUtils.getServerHostName(host);
- clientVM.invoke(new CacheSerializableRunnable("Create client") {
- public void run2() throws CacheException {
- getCache();
- AttributesFactory factory = new AttributesFactory();
- factory.setScope(Scope.LOCAL);
- ClientServerTestCase.configureConnectionPool(factory, hostName, port, -1, false, -1, -1,
- null);
- createRegion(regionName, factory.create());
- }
- });
-
- // Do puts in the client
- clientVM.invoke(new CacheSerializableRunnable("Do puts") {
- public void run2() throws CacheException {
- Region region = getRootRegion().getSubregion(regionName);
- for (int i = 0; i < 10; i++) {
- region.put(i, i);
- }
- }
- });
-
- // Verify tracker in server contains an entry for client thread
- serverVM.invoke(new CacheSerializableRunnable("Do puts") {
- public void run2() throws CacheException {
- LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
- Map eventState = region.getEventState();
- assertEquals(1, eventState.size());
- }
- });
- }
-
- /**
- * Tests adding events to and removing events from an <code>EventTracker</code>.
- */
- @Test
- public void testEventTrackerAddRemoveThreadIdentifier() throws CacheException {
- Host host = Host.getHost(0);
- VM serverVM = host.getVM(0);
- VM clientVM = host.getVM(1);
- final String regionName = getName();
-
- // Create Region in the server and verify tracker is created
- serverVM.invoke(new CacheSerializableRunnable("Create server") {
- public void run2() throws CacheException {
- // Set the message tracking timeout
- System.setProperty(DistributionConfig.GEMFIRE_PREFIX + "messageTrackingTimeout",
- MESSAGE_TRACKING_TIMEOUT);
-
- // Create a distributed Region
- AttributesFactory factory = new AttributesFactory();
- factory.setScope(Scope.DISTRIBUTED_ACK);
- LocalRegion region = (LocalRegion) createRegion(regionName, factory.create());
-
- // Verify an EventTracker is created
- EventTracker eventTracker = region.getEventTracker();
- assertNotNull(eventTracker);
- try {
- startCacheServer();
- } catch (Exception ex) {
- Assert.fail("While starting CacheServer", ex);
- }
- }
- });
-
- // Verify tracker in server contains no entries
- serverVM.invoke(new CacheSerializableRunnable("Do puts") {
- public void run2() throws CacheException {
- LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
- Map eventState = region.getEventState();
- assertEquals(0, eventState.size());
- }
- });
-
- // Create Create Region in the client
- final int port = serverVM.invoke(() -> EventTrackerDUnitTest.getCacheServerPort());
- final String hostName = NetworkUtils.getServerHostName(host);
- clientVM.invoke(new CacheSerializableRunnable("Create client") {
- public void run2() throws CacheException {
- getCache();
- AttributesFactory factory = new AttributesFactory();
- factory.setScope(Scope.LOCAL);
- ClientServerTestCase.configureConnectionPool(factory, hostName, port, -1, false, -1, -1,
- null);
- createRegion(regionName, factory.create());
- }
- });
-
- // Do puts in the client
- clientVM.invoke(new CacheSerializableRunnable("Do puts") {
- public void run2() throws CacheException {
- Region region = getRootRegion().getSubregion(regionName);
- for (int i = 0; i < 10; i++) {
- region.put(i, i);
- }
- }
- });
-
- // Verify tracker in server
- serverVM.invoke(new CacheSerializableRunnable("Do puts") {
- public void run2() throws CacheException {
- // First verify it contains an entry
- LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
- Map eventState = region.getEventState();
- assertEquals(1, eventState.size());
-
- // Pause for the message tracking timeout
- int waitTime = Integer.parseInt(MESSAGE_TRACKING_TIMEOUT) * 3;
- Wait.pause(waitTime);
-
- // Verify the server no longer contains an entry
- eventState = region.getEventState();
- assertEquals(0, eventState.size());
- }
- });
- }
-
- /**
- * Test to make sure we don't leak put all events in the event tracker after multiple putAlls
- */
- @Test
- public void testPutAllHoldersInEventTracker() {
- Host host = Host.getHost(0);
- VM vm0 = host.getVM(0);
- VM vm1 = host.getVM(1);
- VM vm2 = host.getVM(2);
-
- SerializableRunnable createRegion = new SerializableRunnable("createRegion") {
-
- public void run() {
- Cache cache = getCache();
- RegionFactory<Object, Object> rf =
- cache.createRegionFactory(RegionShortcut.PARTITION_REDUNDANT);
- PartitionAttributesFactory paf = new PartitionAttributesFactory();
- paf.setRedundantCopies(1);
- paf.setTotalNumBuckets(3);
- rf.setPartitionAttributes(paf.create());
- rf.setConcurrencyChecksEnabled(true);
- rf.create("partitioned");
-
- rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
- rf.setConcurrencyChecksEnabled(true);
- rf.create("replicate");
- try {
- startCacheServer();
- } catch (Exception ex) {
- Assert.fail("While starting CacheServer", ex);
- }
- }
- };
-
- vm0.invoke(createRegion);
- vm1.invoke(createRegion);
-
- // Create Create Region in the client
- final int port = vm0.invoke(() -> EventTrackerDUnitTest.getCacheServerPort());
- final String hostName = NetworkUtils.getServerHostName(host);
- vm2.invoke(new CacheSerializableRunnable("Create client") {
- public void run2() throws CacheException {
- getCache();
- AttributesFactory factory = new AttributesFactory();
- factory.setScope(Scope.LOCAL);
- ClientServerTestCase.configureConnectionPool(factory, hostName, port, -1, false, -1, -1,
- null);
- createRootRegion("partitioned", factory.create());
- createRootRegion("replicate", factory.create());
- }
- });
-
- doTwoPutAlls(vm2, "partitioned");
- doTwoPutAlls(vm2, "replicate");
-
- // Make sure that the event tracker for each bucket only records the last
- // event.
- checkBucketEventTracker(vm0, 0, 3);
- checkBucketEventTracker(vm1, 0, 3);
- checkBucketEventTracker(vm0, 1, 3);
- checkBucketEventTracker(vm1, 1, 3);
- checkBucketEventTracker(vm0, 2, 3);
- checkBucketEventTracker(vm1, 2, 3);
-
- checkReplicateEventTracker(vm0, 9);
- checkReplicateEventTracker(vm1, 9);
- }
-
- private void doTwoPutAlls(VM vm, final String regionName) {
- SerializableRunnable createData = new SerializableRunnable("putAlls") {
-
- public void run() {
- Cache cache = getCache();
- Region region = cache.getRegion(regionName);
-
- Map putAllMap = new HashMap();
- for (int i = 0; i < 9; i++) {
- putAllMap.put(i, i);
- }
- region.putAll(putAllMap);
-
- putAllMap.clear();
- for (int i = 10; i < 19; i++) {
- putAllMap.put(i, i);
- }
- region.putAll(putAllMap);
- }
- };
-
- vm.invoke(createData);
- }
-
- private SerializableRunnable checkReplicateEventTracker(VM vm, final int expectedEntryCount) {
- SerializableRunnable checkEventTracker = new SerializableRunnable("checkEventTracker") {
-
- public void run() {
- Cache cache = getCache();
- DistributedRegion region = (DistributedRegion) cache.getRegion("replicate");
- checkEventTracker(region, expectedEntryCount);
- }
- };
- vm.invoke(checkEventTracker);
- return checkEventTracker;
- }
-
- private SerializableRunnable checkBucketEventTracker(VM vm, final int bucketNumber,
- final int expectedEntryCount) {
- SerializableRunnable checkEventTracker = new SerializableRunnable("checkEventTracker") {
-
- public void run() {
- Cache cache = getCache();
- PartitionedRegion region = (PartitionedRegion) cache.getRegion("partitioned");
- BucketRegion br = region.getBucketRegion(bucketNumber);
-
- checkEventTracker(br, expectedEntryCount);
- }
- };
- vm.invoke(checkEventTracker);
- return checkEventTracker;
- }
-
- private void checkEventTracker(LocalRegion region, int numberOfEvents) {
- EventTracker tracker = region.getEventTracker();
- ConcurrentMap<ThreadIdentifier, BulkOpHolder> memberToTags =
- tracker.getRecordedBulkOpVersionTags();
- assertEquals("memberToTags=" + memberToTags, 1, memberToTags.size());
- BulkOpHolder holder = memberToTags.values().iterator().next();
- // We expect the holder to retain only the last putAll that was performed.
- assertEquals("entryToVersionTags=" + holder.entryVersionTags, numberOfEvents,
- holder.entryVersionTags.size());
- }
-
- protected void startCacheServer() throws IOException {
- CacheServer cacheServer = getCache().addCacheServer();
- cacheServer.setPort(0);
- cacheServer.start();
- cacheServerPort = cacheServer.getPort();
- }
-
- protected static int getCacheServerPort() {
- return cacheServerPort;
- }
-
- /**
- * Tests event track is initialized after gii
- */
- @Test
- public void testEventTrackerIsInitalized() throws CacheException {
- Host host = Host.getHost(0);
- VM vm0 = host.getVM(0);
- VM vm1 = host.getVM(1);
- VM vm2 = host.getVM(2);
-
- createPRInVMs(vm0, vm1, vm2);
-
- createPR();
-
- doPutsInVMs(vm0, vm1, vm2);
-
- doPuts();
-
- verifyEventTrackerContent();
-
- // close the region
- getCache().getRegion(getName()).close();
-
- // create the region again.
- createPR();
-
- for (int i = 0; i < 12; i++) {
- waitEntryIsLocal(i);
- }
-
- // verify event track initialized after create region
- verifyEventTrackerContent();
-
- }
-
- private void waitEntryIsLocal(int i) {
- Awaitility.await().pollInterval(10, TimeUnit.MILLISECONDS).pollDelay(10, TimeUnit.MILLISECONDS)
- .atMost(30, TimeUnit.SECONDS)
- .until(() -> getCache().getRegion(getName()).getEntry(i) != null);
- }
-
- private void verifyEventTrackerContent() {
- PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(getName());
- BucketRegion br = pr.getDataStore().getLocalBucketById(0);
- Map<?, ?> eventStates = br.getEventState();
- assertTrue(eventStates.size() == 4);
- }
-
- public void createPRInVMs(VM... vms) {
- for (VM vm : vms) {
- vm.invoke(() -> createPR());
- }
- }
-
- private void createPR() {
- PartitionAttributesFactory paf =
- new PartitionAttributesFactory().setRedundantCopies(3).setTotalNumBuckets(4);
- RegionFactory fact = getCache().createRegionFactory(RegionShortcut.PARTITION)
- .setPartitionAttributes(paf.create());
- fact.create(getName());
- }
-
- public void doPutsInVMs(VM... vms) {
- for (VM vm : vms) {
- vm.invoke(() -> doPuts());
- }
- }
-
- private void doPuts() {
- Region region = getCache().getRegion(getName());
- for (int i = 0; i < 12; i++) {
- region.put(i, i);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerTest.java
deleted file mode 100644
index d74b3d5..0000000
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/EventTrackerTest.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * 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;
-
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
-
-import java.util.concurrent.ConcurrentMap;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.apache.geode.CancelCriterion;
-import org.apache.geode.cache.DataPolicy;
-import org.apache.geode.cache.Operation;
-import org.apache.geode.cache.RegionAttributes;
-import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.internal.cache.EventTracker.BulkOpHolder;
-import org.apache.geode.internal.cache.ha.ThreadIdentifier;
-import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
-import org.apache.geode.internal.cache.versions.VersionTag;
-import org.apache.geode.test.junit.categories.UnitTest;
-
-@Category(UnitTest.class)
-public class EventTrackerTest {
- LocalRegion lr;
- RegionAttributes<?, ?> ra;
- EntryEventImpl[] events;
- EventTracker eventTracker;
- ClientProxyMembershipID memberId;
- DistributedMember member;
-
- @Before
- public void setUp() {
- lr = mock(LocalRegion.class);
- ra = mock(RegionAttributes.class);
- when(lr.createStopper()).thenCallRealMethod();
- CancelCriterion stopper = lr.createStopper();
- when(lr.getStopper()).thenReturn(stopper);
- memberId = mock(ClientProxyMembershipID.class);
- when(lr.getAttributes()).thenReturn(ra);
- when(ra.getDataPolicy()).thenReturn(mock(DataPolicy.class));
- when(lr.getConcurrencyChecksEnabled()).thenReturn(true);
-
- member = mock(DistributedMember.class);
- }
-
- @Test
- public void retriedBulkOpDoesNotRemoveRecordedBulkOpVersionTags() {
- byte[] memId = {1, 2, 3};
- long threadId = 1;
- long retrySeqId = 1;
- ThreadIdentifier tid = new ThreadIdentifier(memId, threadId);
- EventID retryEventID = new EventID(memId, threadId, retrySeqId);
- boolean skipCallbacks = true;
- int size = 5;
- recordPutAllEvents(memId, threadId, skipCallbacks, size);
-
- ConcurrentMap<ThreadIdentifier, BulkOpHolder> map = eventTracker.getRecordedBulkOpVersionTags();
- BulkOpHolder holder = map.get(tid);
- int beforeSize = holder.entryVersionTags.size();
-
- eventTracker.recordBulkOpStart(tid, retryEventID);
- map = eventTracker.getRecordedBulkOpVersionTags();
- holder = map.get(tid);
- // Retried bulk op should not remove exiting BulkOpVersionTags
- assertTrue(holder.entryVersionTags.size() == beforeSize);
- }
-
- private void recordPutAllEvents(byte[] memId, long threadId, boolean skipCallbacks, int size) {
- events = new EntryEventImpl[size];
- eventTracker = new EventTracker(lr);
- for (int i = 0; i < size; i++) {
- events[i] = EntryEventImpl.create(lr, Operation.PUTALL_CREATE, "key" + i, "value" + i, null,
- false, member, !skipCallbacks, new EventID(memId, threadId, i + 1));
- events[i].setContext(memberId);
- events[i].setVersionTag(mock(VersionTag.class));
- eventTracker.recordEvent(events[i]);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/IteratorDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/IteratorDUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/IteratorDUnitTest.java
index 6ca11b2..a1922cf 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/IteratorDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/IteratorDUnitTest.java
@@ -58,7 +58,7 @@ public class IteratorDUnitTest extends JUnit4CacheTestCase {
r.put("key3", "value3");
LocalRegion lr = (LocalRegion) r;
// simulate a removed key
- // lr.getRegionMap().getEntry("key")._setValue(Token.REMOVED_PHASE1);
+ // region.getRegionMap().getEntry("key")._setValue(Token.REMOVED_PHASE1);
lr.getRegionMap().getEntry("key").setValue(lr, Token.REMOVED_PHASE1);
Iterator it = r.keySet().iterator();
int numKeys = 0;
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionDUnitTestCase.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionDUnitTestCase.java b/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionDUnitTestCase.java
index 54263b9..c3aff4c 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionDUnitTestCase.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionDUnitTestCase.java
@@ -192,9 +192,9 @@ public class PartitionedRegionDUnitTestCase extends JUnit4CacheTestCase {
final long recoveryDelay) {
return new CacheSerializableRunnable("getCreateMultiplePRregion") {
public void run2() throws CacheException {
- // final Random ra = new Random();
+ // final Random regionAttributes = new Random();
for (int i = 0; i < maxIndex; i++) {
- // final int rind = ra.nextInt(maxIndex);
+ // final int rind = regionAttributes.nextInt(maxIndex);
try {
getCache().createRegion(prPrefix + i, PartitionedRegionTestHelper
.createRegionAttrsForPR(redundancy, localmaxMemory, recoveryDelay));
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionSingleNodeOperationsJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionSingleNodeOperationsJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionSingleNodeOperationsJUnitTest.java
index e69aa95..cd8408a 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionSingleNodeOperationsJUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionSingleNodeOperationsJUnitTest.java
@@ -581,11 +581,12 @@ public class PartitionedRegionSingleNodeOperationsJUnitTest {
*
* Region distRegion = null;
*
- * AttributesFactory af = new AttributesFactory(); RegionAttributes ra; // setting property
- * af.setScope(Scope.DISTRIBUTED_ACK); // creating region attributes ra = af.create(); try {
- * distRegion = PartitionedRegionTestHelper.createCache().createRegion( diRegion, ra); } catch
- * (RegionExistsException rex) { distRegion = PartitionedRegionTestHelper.createCache()
- * .getRegion(diRegion); } // Closing the regions distRegion.close(); pr.close();
+ * AttributesFactory af = new AttributesFactory(); RegionAttributes regionAttributes; // setting
+ * property af.setScope(Scope.DISTRIBUTED_ACK); // creating region attributes regionAttributes =
+ * af.create(); try { distRegion = PartitionedRegionTestHelper.createCache().createRegion(
+ * diRegion, regionAttributes); } catch (RegionExistsException rex) { distRegion =
+ * PartitionedRegionTestHelper.createCache() .getRegion(diRegion); } // Closing the regions
+ * distRegion.close(); pr.close();
*
* if (!pr.getCache().equals(distRegion.getCache())) {
* fail("testValidateCloseFunction: getCache is not matching. "); } else { if
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/event/DistributedEventTrackerTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/event/DistributedEventTrackerTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/event/DistributedEventTrackerTest.java
new file mode 100644
index 0000000..fd69f98
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/event/DistributedEventTrackerTest.java
@@ -0,0 +1,328 @@
+/*
+ * 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.event;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.apache.geode.CancelCriterion;
+import org.apache.geode.cache.DataPolicy;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.RegionAttributes;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
+import org.apache.geode.internal.cache.versions.VersionTag;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class DistributedEventTrackerTest {
+ LocalRegion region;
+ RegionAttributes<?, ?> regionAttributes;
+ DistributedEventTracker eventTracker;
+ ClientProxyMembershipID memberId;
+ DistributedMember member;
+
+ @Before
+ public void setup() {
+ region = mock(LocalRegion.class);
+ regionAttributes = mock(RegionAttributes.class);
+ when(region.createStopper()).thenCallRealMethod();
+ memberId = mock(ClientProxyMembershipID.class);
+ when(region.getAttributes()).thenReturn(regionAttributes);
+ when(regionAttributes.getDataPolicy()).thenReturn(mock(DataPolicy.class));
+ when(region.getConcurrencyChecksEnabled()).thenReturn(true);
+
+ member = mock(DistributedMember.class);
+ eventTracker = new DistributedEventTracker(region.getCache(), mock(CancelCriterion.class),
+ region.getName());
+ }
+
+ @Test
+ public void retriedBulkOpDoesNotRemoveRecordedBulkOpVersionTags() {
+ byte[] memId = {1, 2, 3};
+ long threadId = 1;
+ long retrySeqId = 1;
+ ThreadIdentifier tid = new ThreadIdentifier(memId, threadId);
+ EventID retryEventID = new EventID(memId, threadId, retrySeqId);
+ boolean skipCallbacks = true;
+ int size = 5;
+ recordPutAllEvents(memId, threadId, skipCallbacks, size);
+
+ ConcurrentMap<ThreadIdentifier, BulkOperationHolder> map =
+ eventTracker.getRecordedBulkOpVersionTags();
+ BulkOperationHolder holder = map.get(tid);
+ int beforeSize = holder.getEntryVersionTags().size();
+
+ eventTracker.recordBulkOpStart(retryEventID, tid);
+ map = eventTracker.getRecordedBulkOpVersionTags();
+ holder = map.get(tid);
+ // Retried bulk op should not remove exiting BulkOpVersionTags
+ assertTrue(holder.getEntryVersionTags().size() == beforeSize);
+ }
+
+ private void recordPutAllEvents(byte[] memId, long threadId, boolean skipCallbacks, int size) {
+ for (int i = 0; i < size; i++) {
+ putEvent("key" + i, "value" + i, memId, threadId, skipCallbacks, i + 1);
+ EntryEventImpl event = EntryEventImpl.create(region, Operation.PUTALL_CREATE, "key" + i,
+ "value" + i, null, false, member, !skipCallbacks, new EventID(memId, threadId, i + 1));
+ event.setContext(memberId);
+ event.setVersionTag(mock(VersionTag.class));
+ eventTracker.recordEvent(event);
+ }
+ }
+
+ private void putEvent(String key, String value, byte[] memId, long threadId,
+ boolean skipCallbacks, int sequenceId) {
+ EntryEventImpl event = EntryEventImpl.create(region, Operation.PUTALL_CREATE, key, value, null,
+ false, member, !skipCallbacks, new EventID(memId, threadId, sequenceId));
+ event.setContext(memberId);
+ event.setVersionTag(mock(VersionTag.class));
+ eventTracker.recordEvent(event);
+ }
+
+ private void putEvent(String key, String value, byte[] memId, long threadId,
+ boolean skipCallbacks, int sequenceId, VersionTag tag) {
+ EntryEventImpl event = EntryEventImpl.create(region, Operation.PUTALL_CREATE, key, value, null,
+ false, member, !skipCallbacks, new EventID(memId, threadId, sequenceId));
+ event.setContext(memberId);
+ event.setVersionTag(tag);
+ eventTracker.recordEvent(event);
+ }
+
+ @Test
+ public void returnsCorrectNameOfCache() {
+ String testName = "testing";
+ when(region.getName()).thenReturn(testName);
+ eventTracker = new DistributedEventTracker(region.getCache(), mock(CancelCriterion.class),
+ region.getName());
+ assertEquals("Event Tracker for " + testName, eventTracker.getName());
+ }
+
+ @Test
+ public void initializationCorrectlyReadiesTheTracker() throws InterruptedException {
+ assertFalse(eventTracker.isInitialized());
+ eventTracker.setInitialized();
+ assertTrue(eventTracker.isInitialized());
+ eventTracker.waitOnInitialization();
+ }
+
+ @Test
+ public void startAndStopAddAndRemoveTrackerFromExpiryTask() {
+ EventTrackerExpiryTask task = mock(EventTrackerExpiryTask.class);
+ InternalCache cache = mock(InternalCache.class);
+ when(region.getCache()).thenReturn(cache);
+ when(cache.getEventTrackerTask()).thenReturn(task);
+ eventTracker = new DistributedEventTracker(region.getCache(), mock(CancelCriterion.class),
+ region.getName());
+ eventTracker.start();
+ verify(task, times(1)).addTracker(eventTracker);
+ eventTracker.stop();
+ verify(task, times(1)).removeTracker(eventTracker);
+ }
+
+ @Test
+ public void returnsEmptyMapIfRecordedEventsAreEmpty() {
+ assertEquals(0, eventTracker.getState().size());
+ }
+
+ @Test
+ public void returnsMapContainingSequenceIdHoldersCurrentlyPresent() {
+ EventSequenceNumberHolder sequenceIdHolder = new EventSequenceNumberHolder(0L, null);
+ ThreadIdentifier threadId = new ThreadIdentifier(new byte[0], 0L);
+ eventTracker.recordSequenceNumber(threadId, sequenceIdHolder);
+ Map<ThreadIdentifier, EventSequenceNumberHolder> state = eventTracker.getState();
+ assertEquals(1, state.size());
+ EventSequenceNumberHolder returnedHolder = state.get(threadId);
+ assertNotNull(returnedHolder);
+ // the version tag is stripped out on purpose, so passed in object and returned one are not
+ // equal to each other
+ assertNull(returnedHolder.getVersionTag());
+ assertEquals(sequenceIdHolder.getLastSequenceNumber(), returnedHolder.getLastSequenceNumber());
+ }
+
+ @Test
+ public void setToInitializedWhenStateRecorded() {
+ eventTracker.recordState(null, Collections.emptyMap());
+ assertTrue(eventTracker.isInitialized());
+ }
+
+ @Test
+ public void setsInitialImageProvidedWhenStateRecorded() {
+ InternalDistributedMember distributedMember = mock(InternalDistributedMember.class);
+ eventTracker.recordState(distributedMember, Collections.emptyMap());
+ assertTrue(eventTracker.isInitialImageProvider(distributedMember));
+ }
+
+ @Test
+ public void entryInRecordedStateStoredWhenNotInCurrentState() {
+ EventSequenceNumberHolder sequenceIdHolder = new EventSequenceNumberHolder(0L, null);
+ ThreadIdentifier threadId = new ThreadIdentifier(new byte[0], 0L);
+ Map<ThreadIdentifier, EventSequenceNumberHolder> state =
+ Collections.singletonMap(threadId, sequenceIdHolder);
+ eventTracker.recordState(null, state);
+ Map<ThreadIdentifier, EventSequenceNumberHolder> storedState = eventTracker.getState();
+ assertEquals(storedState.get(threadId).getLastSequenceNumber(),
+ sequenceIdHolder.getLastSequenceNumber());
+ }
+
+ @Test
+ public void entryInRecordedStateNotStoredIfAlreadyInCurrentState() {
+ EventSequenceNumberHolder originalSequenceIdHolder = new EventSequenceNumberHolder(0L, null);
+ ThreadIdentifier threadId = new ThreadIdentifier(new byte[0], 0L);
+ Map<ThreadIdentifier, EventSequenceNumberHolder> state =
+ Collections.singletonMap(threadId, originalSequenceIdHolder);
+ eventTracker.recordState(null, state);
+
+ EventSequenceNumberHolder newSequenceIdHolder = new EventSequenceNumberHolder(1L, null);
+ Map<ThreadIdentifier, EventSequenceNumberHolder> newState =
+ Collections.singletonMap(threadId, newSequenceIdHolder);
+ eventTracker.recordState(null, newState);
+
+ Map<ThreadIdentifier, EventSequenceNumberHolder> storedState = eventTracker.getState();
+ assertEquals(storedState.get(threadId).getLastSequenceNumber(),
+ originalSequenceIdHolder.getLastSequenceNumber());
+ }
+
+ @Test
+ public void hasSeenEventReturnsFalseForEventWithNoID() {
+ InternalCacheEvent event = mock(InternalCacheEvent.class);
+ when(event.getEventId()).thenReturn(null);
+ assertFalse(eventTracker.hasSeenEvent(event));
+ }
+
+ @Test
+ public void hasSeenEventReturnsFalseForNullEventID() {
+ assertFalse(eventTracker.hasSeenEvent((EventID) null));
+ assertFalse(eventTracker.hasSeenEvent(null, null));
+ }
+
+ @Test
+ public void hasNotSeenEventIDThatIsNotInRecordedEvents() {
+ EventID eventID = new EventID(new byte[0], 0L, 0L);
+ assertFalse(eventTracker.hasSeenEvent(eventID));
+ }
+
+ @Test
+ public void hasSeenEventIDThatIsInRecordedEvents() {
+ EventID eventID = new EventID(new byte[0], 0L, 0L);
+ recordSequence(eventID);
+ assertTrue(eventTracker.hasSeenEvent(eventID));
+ }
+
+ @Test
+ public void hasNotSeenEventIDWhosSequenceIDIsMarkedRemoved() {
+ EventID eventID = new EventID(new byte[0], 0L, 0L);
+ EventSequenceNumberHolder sequenceIdHolder =
+ new EventSequenceNumberHolder(eventID.getSequenceID(), null);
+ sequenceIdHolder.setRemoved(true);
+ ThreadIdentifier threadId = new ThreadIdentifier(new byte[0], 0L);
+ eventTracker.recordSequenceNumber(threadId, sequenceIdHolder);
+
+ assertFalse(eventTracker.hasSeenEvent(eventID));
+ }
+
+ @Test
+ public void hasNotSeeEventIDWhosSequenceIDIsLargerThanSeen() {
+ EventID eventID = new EventID(new byte[0], 0L, 0L);
+ recordSequence(eventID);
+
+ EventID higherSequenceID = new EventID(new byte[0], 0L, 1);
+ assertFalse(eventTracker.hasSeenEvent(higherSequenceID));
+ }
+
+ @Test
+ public void returnsNoTagIfNoSequenceForEvent() {
+ EventID eventID = new EventID(new byte[0], 0L, 1L);
+ assertNull(eventTracker.findVersionTagForSequence(eventID));
+ }
+
+ @Test
+ public void returnsNoTagIfSequencesDoNotMatchForEvent() {
+ EventID eventID = new EventID(new byte[0], 0L, 1);
+ recordSequence(eventID);
+ assertNull(eventTracker.findVersionTagForSequence(eventID));
+ }
+
+ @Test
+ public void returnsCorrectTagForEvent() {
+ EventID eventID = new EventID(new byte[0], 0L, 0L);
+ EventSequenceNumberHolder sequenceIdHolder = recordSequence(eventID);
+ assertEquals(sequenceIdHolder.getVersionTag(), eventTracker.findVersionTagForSequence(eventID));
+ }
+
+ @Test
+ public void returnsNoTagIfNoBulkOpWhenNoEventGiven() {
+ assertNull(eventTracker.findVersionTagForBulkOp(null));
+ }
+
+ @Test
+ public void returnsNoTagIfNoBulkOpForEventWithSequence() {
+ EventID eventID = new EventID(new byte[0], 0L, 1L);
+ assertNull(eventTracker.findVersionTagForBulkOp(eventID));
+ }
+
+ @Test
+ public void returnsNoTagIfBulkOpsDoNotMatchForEvent() {
+ putEvent("key", "value", new byte[0], 0, false, 0);
+ EventID eventIDWithoutBulkOp = new EventID(new byte[0], 0L, 1);
+ assertNull(eventTracker.findVersionTagForBulkOp(eventIDWithoutBulkOp));
+ }
+
+ @Test
+ public void returnsCorrectTagForEventWithBulkOp() {
+ EventID eventID = new EventID(new byte[0], 0L, 0L);
+ VersionTag tag = mock(VersionTag.class);
+ putEvent("key", "value", new byte[0], 0, false, 0, tag);
+ assertEquals(tag, eventTracker.findVersionTagForBulkOp(eventID));
+ }
+
+ @Test
+ public void executesABulkOperations() {
+ EventID eventID = new EventID(new byte[0], 0L, 1L);
+ Runnable bulkOperation = mock(Runnable.class);
+ eventTracker.syncBulkOp(bulkOperation, eventID, false);
+ verify(bulkOperation, times(1)).run();
+ }
+
+ @Test
+ public void executesRunnableIfNotPartOfATransaction() {
+ EventID eventID = new EventID(new byte[0], 0L, 1L);
+ Runnable bulkOperation = mock(Runnable.class);
+ eventTracker.syncBulkOp(bulkOperation, eventID, true);
+ verify(bulkOperation, times(1)).run();
+ }
+
+ private EventSequenceNumberHolder recordSequence(EventID eventID) {
+ EventSequenceNumberHolder sequenceIdHolder =
+ new EventSequenceNumberHolder(eventID.getSequenceID(), null);
+ ThreadIdentifier threadIdentifier = new ThreadIdentifier(new byte[0], eventID.getThreadID());
+ eventTracker.recordSequenceNumber(threadIdentifier, sequenceIdHolder);
+ return sequenceIdHolder;
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerDUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerDUnitTest.java
new file mode 100755
index 0000000..b85827c
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerDUnitTest.java
@@ -0,0 +1,489 @@
+/*
+ * 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.event;
+
+import static org.junit.Assert.*;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.AttributesFactory;
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheException;
+import org.apache.geode.cache.PartitionAttributesFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.Scope;
+import org.apache.geode.cache.server.CacheServer;
+import org.apache.geode.cache30.CacheSerializableRunnable;
+import org.apache.geode.cache30.ClientServerTestCase;
+import org.apache.geode.distributed.internal.DistributionConfig;
+import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.DistributedRegion;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+import org.apache.geode.test.dunit.Assert;
+import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.NetworkUtils;
+import org.apache.geode.test.dunit.SerializableRunnable;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.Wait;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
+
+/**
+ * Tests <code>EventTracker</code> management.
+ *
+ * @since GemFire 6.5
+ */
+@Category(DistributedTest.class)
+public class EventTrackerDUnitTest extends JUnit4CacheTestCase {
+
+ /** The port on which the <code>CacheServer</code> was started in this VM */
+ private static int cacheServerPort;
+
+ /** The <code>Cache</code>'s <code>ExpiryTask</code>'s ping interval */
+ private static final String MESSAGE_TRACKING_TIMEOUT = "5000";
+
+ @Override
+ public final void postTearDownCacheTestCase() throws Exception {
+ disconnectAllFromDS();
+ }
+
+ /**
+ * Tests <code>EventTracker</code> is created and destroyed when a <code>Region</code> is created
+ * and destroyed.
+ */
+ @Test
+ public void testEventTrackerCreateDestroy() throws CacheException {
+ // Verify the Cache's ExpiryTask contains no EventTrackers
+ GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
+ EventTrackerExpiryTask expiryTask = cache.getEventTrackerTask();
+ assertNotNull(expiryTask);
+
+ // We start with 3 event trackers:
+ // one for the PDX registry region
+ // one for ManagementConstants.MONITORING_REGION
+ // one for ManagementConstants.NOTIFICATION_REGION
+ final int EXPECTED_TRACKERS = 3;
+ assertEquals(EXPECTED_TRACKERS, expiryTask.getNumberOfTrackers());
+
+ // Create a distributed Region
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ LocalRegion region = (LocalRegion) createRegion(getName(), factory.create());
+
+ // Verify an EventTracker is created and is empty
+ EventTracker eventTracker = region.getEventTracker();
+ assertNotNull(eventTracker);
+ Map eventState = region.getEventState();
+ assertNotNull(eventState);
+ assertEquals(0, eventState.size());
+
+ // Verify it and the root region's EventTracker are added to the Cache's ExpiryTask's trackers
+ assertEquals(EXPECTED_TRACKERS + 2, expiryTask.getNumberOfTrackers());
+
+ // Destroy the Region
+ region.destroyRegion();
+
+ // Verify the EventTracker is removed from the Cache's ExpiryTask's trackers
+ assertEquals(EXPECTED_TRACKERS + 1, expiryTask.getNumberOfTrackers());
+ }
+
+ /**
+ * Tests adding threads to an <code>EventTracker</code>.
+ */
+ @Test
+ public void testEventTrackerAddThreadIdentifier() throws CacheException {
+ Host host = Host.getHost(0);
+ VM serverVM = host.getVM(0);
+ VM clientVM = host.getVM(1);
+ final String regionName = getName();
+
+ // Create Region in the server and verify tracker is created
+ serverVM.invoke(new CacheSerializableRunnable("Create server") {
+ public void run2() throws CacheException {
+ // Create a distributed Region
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ LocalRegion region = (LocalRegion) createRegion(regionName, factory.create());
+
+ // Verify an EventTracker is created
+ EventTracker eventTracker = region.getEventTracker();
+ assertNotNull(eventTracker);
+ try {
+ startCacheServer();
+ } catch (Exception ex) {
+ Assert.fail("While starting CacheServer", ex);
+ }
+ }
+ });
+
+ // Verify tracker in server contains no entries
+ serverVM.invoke(new CacheSerializableRunnable("Do puts") {
+ public void run2() throws CacheException {
+ LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
+ Map eventState = region.getEventState();
+ assertEquals(0, eventState.size());
+ }
+ });
+
+ // Create Create Region in the client
+ final int port = serverVM.invoke(() -> EventTrackerDUnitTest.getCacheServerPort());
+ final String hostName = NetworkUtils.getServerHostName(host);
+ clientVM.invoke(new CacheSerializableRunnable("Create client") {
+ public void run2() throws CacheException {
+ getCache();
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.LOCAL);
+ ClientServerTestCase.configureConnectionPool(factory, hostName, port, -1, false, -1, -1,
+ null);
+ createRegion(regionName, factory.create());
+ }
+ });
+
+ // Do puts in the client
+ clientVM.invoke(new CacheSerializableRunnable("Do puts") {
+ public void run2() throws CacheException {
+ Region region = getRootRegion().getSubregion(regionName);
+ for (int i = 0; i < 10; i++) {
+ region.put(i, i);
+ }
+ }
+ });
+
+ // Verify tracker in server contains an entry for client thread
+ serverVM.invoke(new CacheSerializableRunnable("Do puts") {
+ public void run2() throws CacheException {
+ LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
+ Map eventState = region.getEventState();
+ assertEquals(1, eventState.size());
+ }
+ });
+ }
+
+ /**
+ * Tests adding events to and removing events from an <code>EventTracker</code>.
+ */
+ @Test
+ public void testEventTrackerAddRemoveThreadIdentifier() throws CacheException {
+ Host host = Host.getHost(0);
+ VM serverVM = host.getVM(0);
+ VM clientVM = host.getVM(1);
+ final String regionName = getName();
+
+ // Create Region in the server and verify tracker is created
+ serverVM.invoke(new CacheSerializableRunnable("Create server") {
+ public void run2() throws CacheException {
+ // Set the message tracking timeout
+ System.setProperty(DistributionConfig.GEMFIRE_PREFIX + "messageTrackingTimeout",
+ MESSAGE_TRACKING_TIMEOUT);
+
+ // Create a distributed Region
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ LocalRegion region = (LocalRegion) createRegion(regionName, factory.create());
+
+ // Verify an EventTracker is created
+ EventTracker eventTracker = region.getEventTracker();
+ assertNotNull(eventTracker);
+ try {
+ startCacheServer();
+ } catch (Exception ex) {
+ Assert.fail("While starting CacheServer", ex);
+ }
+ }
+ });
+
+ // Verify tracker in server contains no entries
+ serverVM.invoke(new CacheSerializableRunnable("Do puts") {
+ public void run2() throws CacheException {
+ LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
+ Map eventState = region.getEventState();
+ assertEquals(0, eventState.size());
+ }
+ });
+
+ // Create Create Region in the client
+ final int port = serverVM.invoke(() -> EventTrackerDUnitTest.getCacheServerPort());
+ final String hostName = NetworkUtils.getServerHostName(host);
+ clientVM.invoke(new CacheSerializableRunnable("Create client") {
+ public void run2() throws CacheException {
+ getCache();
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.LOCAL);
+ ClientServerTestCase.configureConnectionPool(factory, hostName, port, -1, false, -1, -1,
+ null);
+ createRegion(regionName, factory.create());
+ }
+ });
+
+ // Do puts in the client
+ clientVM.invoke(new CacheSerializableRunnable("Do puts") {
+ public void run2() throws CacheException {
+ Region region = getRootRegion().getSubregion(regionName);
+ for (int i = 0; i < 10; i++) {
+ region.put(i, i);
+ }
+ }
+ });
+
+ // Verify tracker in server
+ serverVM.invoke(new CacheSerializableRunnable("Do puts") {
+ public void run2() throws CacheException {
+ // First verify it contains an entry
+ LocalRegion region = (LocalRegion) getRootRegion().getSubregion(regionName);
+ Map eventState = region.getEventState();
+ assertEquals(1, eventState.size());
+
+ // Pause for the message tracking timeout
+ int waitTime = Integer.parseInt(MESSAGE_TRACKING_TIMEOUT) * 3;
+ Wait.pause(waitTime);
+
+ // Verify the server no longer contains an entry
+ eventState = region.getEventState();
+ assertEquals(0, eventState.size());
+ }
+ });
+ }
+
+ /**
+ * Test to make sure we don't leak put all events in the event tracker after multiple putAlls
+ */
+ @Test
+ public void testPutAllHoldersInEventTracker() {
+ Host host = Host.getHost(0);
+ VM vm0 = host.getVM(0);
+ VM vm1 = host.getVM(1);
+ VM vm2 = host.getVM(2);
+
+ SerializableRunnable createRegion = new SerializableRunnable("createRegion") {
+
+ public void run() {
+ Cache cache = getCache();
+ RegionFactory<Object, Object> rf =
+ cache.createRegionFactory(RegionShortcut.PARTITION_REDUNDANT);
+ PartitionAttributesFactory paf = new PartitionAttributesFactory();
+ paf.setRedundantCopies(1);
+ paf.setTotalNumBuckets(3);
+ rf.setPartitionAttributes(paf.create());
+ rf.setConcurrencyChecksEnabled(true);
+ rf.create("partitioned");
+
+ rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
+ rf.setConcurrencyChecksEnabled(true);
+ rf.create("replicate");
+ try {
+ startCacheServer();
+ } catch (Exception ex) {
+ Assert.fail("While starting CacheServer", ex);
+ }
+ }
+ };
+
+ vm0.invoke(createRegion);
+ vm1.invoke(createRegion);
+
+ // Create Create Region in the client
+ final int port = vm0.invoke(() -> EventTrackerDUnitTest.getCacheServerPort());
+ final String hostName = NetworkUtils.getServerHostName(host);
+ vm2.invoke(new CacheSerializableRunnable("Create client") {
+ public void run2() throws CacheException {
+ getCache();
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.LOCAL);
+ ClientServerTestCase.configureConnectionPool(factory, hostName, port, -1, false, -1, -1,
+ null);
+ createRootRegion("partitioned", factory.create());
+ createRootRegion("replicate", factory.create());
+ }
+ });
+
+ doTwoPutAlls(vm2, "partitioned");
+ doTwoPutAlls(vm2, "replicate");
+
+ // Make sure that the event tracker for each bucket only records the last
+ // event.
+ checkBucketEventTracker(vm0, 0, 3);
+ checkBucketEventTracker(vm1, 0, 3);
+ checkBucketEventTracker(vm0, 1, 3);
+ checkBucketEventTracker(vm1, 1, 3);
+ checkBucketEventTracker(vm0, 2, 3);
+ checkBucketEventTracker(vm1, 2, 3);
+
+ checkReplicateEventTracker(vm0, 9);
+ checkReplicateEventTracker(vm1, 9);
+ }
+
+ private void doTwoPutAlls(VM vm, final String regionName) {
+ SerializableRunnable createData = new SerializableRunnable("putAlls") {
+
+ public void run() {
+ Cache cache = getCache();
+ Region region = cache.getRegion(regionName);
+
+ Map putAllMap = new HashMap();
+ for (int i = 0; i < 9; i++) {
+ putAllMap.put(i, i);
+ }
+ region.putAll(putAllMap);
+
+ putAllMap.clear();
+ for (int i = 10; i < 19; i++) {
+ putAllMap.put(i, i);
+ }
+ region.putAll(putAllMap);
+ }
+ };
+
+ vm.invoke(createData);
+ }
+
+ private SerializableRunnable checkReplicateEventTracker(VM vm, final int expectedEntryCount) {
+ SerializableRunnable checkEventTracker = new SerializableRunnable("checkEventTracker") {
+
+ public void run() {
+ Cache cache = getCache();
+ DistributedRegion region = (DistributedRegion) cache.getRegion("replicate");
+ checkEventTracker(region, expectedEntryCount);
+ }
+ };
+ vm.invoke(checkEventTracker);
+ return checkEventTracker;
+ }
+
+ private SerializableRunnable checkBucketEventTracker(VM vm, final int bucketNumber,
+ final int expectedEntryCount) {
+ SerializableRunnable checkEventTracker = new SerializableRunnable("checkEventTracker") {
+
+ public void run() {
+ Cache cache = getCache();
+ PartitionedRegion region = (PartitionedRegion) cache.getRegion("partitioned");
+ BucketRegion br = region.getBucketRegion(bucketNumber);
+
+ checkEventTracker(br, expectedEntryCount);
+ }
+ };
+ vm.invoke(checkEventTracker);
+ return checkEventTracker;
+ }
+
+ private void checkEventTracker(LocalRegion region, int numberOfEvents) {
+ EventTracker tracker = region.getEventTracker();
+ ConcurrentMap<ThreadIdentifier, BulkOperationHolder> memberToTags =
+ tracker.getRecordedBulkOpVersionTags();
+ assertEquals("memberToTags=" + memberToTags, 1, memberToTags.size());
+ BulkOperationHolder holder = memberToTags.values().iterator().next();
+ // We expect the holder to retain only the last putAll that was performed.
+ assertEquals("entryToVersionTags=" + holder.getEntryVersionTags(), numberOfEvents,
+ holder.getEntryVersionTags().size());
+ }
+
+ protected void startCacheServer() throws IOException {
+ CacheServer cacheServer = getCache().addCacheServer();
+ cacheServer.setPort(0);
+ cacheServer.start();
+ cacheServerPort = cacheServer.getPort();
+ }
+
+ protected static int getCacheServerPort() {
+ return cacheServerPort;
+ }
+
+ /**
+ * Tests event track is initialized after gii
+ */
+ @Test
+ public void testEventTrackerIsInitalized() throws CacheException {
+ Host host = Host.getHost(0);
+ VM vm0 = host.getVM(0);
+ VM vm1 = host.getVM(1);
+ VM vm2 = host.getVM(2);
+
+ createPRInVMs(vm0, vm1, vm2);
+
+ createPR();
+
+ doPutsInVMs(vm0, vm1, vm2);
+
+ doPuts();
+
+ verifyEventTrackerContent();
+
+ // close the region
+ getCache().getRegion(getName()).close();
+
+ // create the region again.
+ createPR();
+
+ for (int i = 0; i < 12; i++) {
+ waitEntryIsLocal(i);
+ }
+
+ // verify event track initialized after create region
+ verifyEventTrackerContent();
+
+ }
+
+ private void waitEntryIsLocal(int i) {
+ Awaitility.await().pollInterval(10, TimeUnit.MILLISECONDS).pollDelay(10, TimeUnit.MILLISECONDS)
+ .atMost(30, TimeUnit.SECONDS)
+ .until(() -> getCache().getRegion(getName()).getEntry(i) != null);
+ }
+
+ private void verifyEventTrackerContent() {
+ PartitionedRegion pr = (PartitionedRegion) getCache().getRegion(getName());
+ BucketRegion br = pr.getDataStore().getLocalBucketById(0);
+ Map<?, ?> eventStates = br.getEventState();
+ assertTrue(eventStates.size() == 4);
+ }
+
+ public void createPRInVMs(VM... vms) {
+ for (VM vm : vms) {
+ vm.invoke(() -> createPR());
+ }
+ }
+
+ private void createPR() {
+ PartitionAttributesFactory paf =
+ new PartitionAttributesFactory().setRedundantCopies(3).setTotalNumBuckets(4);
+ RegionFactory fact = getCache().createRegionFactory(RegionShortcut.PARTITION)
+ .setPartitionAttributes(paf.create());
+ fact.create(getName());
+ }
+
+ public void doPutsInVMs(VM... vms) {
+ for (VM vm : vms) {
+ vm.invoke(() -> doPuts());
+ }
+ }
+
+ private void doPuts() {
+ Region region = getCache().getRegion(getName());
+ for (int i = 0; i < 12; i++) {
+ region.put(i, i);
+ }
+ }
+}
[7/9] geode git commit: GEODE-2189 Docs: Update Swagger UI links
Added link to OpenAPI specification.
Posted by kh...@apache.org.
GEODE-2189 Docs: Update Swagger UI links
Added link to OpenAPI specification.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/e1840b16
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/e1840b16
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/e1840b16
Branch: refs/heads/develop
Commit: e1840b16c54f812918dfe66ba466a620181aa928
Parents: 20c6e3d
Author: Dave Barnes <db...@pivotal.io>
Authored: Fri Jul 14 10:05:22 2017 -0700
Committer: Ken Howe <kh...@pivotal.io>
Committed: Tue Jul 18 07:11:33 2017 -0700
----------------------------------------------------------------------
geode-docs/rest_apps/using_swagger.html.md.erb | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/e1840b16/geode-docs/rest_apps/using_swagger.html.md.erb
----------------------------------------------------------------------
diff --git a/geode-docs/rest_apps/using_swagger.html.md.erb b/geode-docs/rest_apps/using_swagger.html.md.erb
index 05f4129..ab0d84b 100644
--- a/geode-docs/rest_apps/using_swagger.html.md.erb
+++ b/geode-docs/rest_apps/using_swagger.html.md.erb
@@ -62,6 +62,5 @@ Specify an `http-service-port` for the developer REST service, as the default po
You can use the Swagger interface to try out additional Geode API endpoints and view sample responses.
-For more information on Swagger, see the [Swagger website](http://swagger.io/).
-
+For more information on Swagger, see the [Swagger website](http://swagger.io/) and the [OpenAPI specification](https://github.com/OAI/OpenAPI-Specification).
[5/9] geode git commit: GEODE-2920: added security tests for create
diskstore and create persistent region.
Posted by kh...@apache.org.
GEODE-2920: added security tests for create diskstore and create persistent region.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/20c6e3d4
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/20c6e3d4
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/20c6e3d4
Branch: refs/heads/develop
Commit: 20c6e3d40797cb97ca1ee3d4b0365e28cda95e2a
Parents: c5520f9
Author: Jinmei Liao <ji...@pivotal.io>
Authored: Mon Jul 17 10:41:19 2017 -0700
Committer: Ken Howe <kh...@pivotal.io>
Committed: Tue Jul 18 07:11:33 2017 -0700
----------------------------------------------------------------------
.../security/GfshCommandsSecurityTest.java | 93 +++++++++++++-------
1 file changed, 59 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/20c6e3d4/geode-core/src/test/java/org/apache/geode/management/internal/security/GfshCommandsSecurityTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/management/internal/security/GfshCommandsSecurityTest.java b/geode-core/src/test/java/org/apache/geode/management/internal/security/GfshCommandsSecurityTest.java
index c2be173..8635306 100644
--- a/geode-core/src/test/java/org/apache/geode/management/internal/security/GfshCommandsSecurityTest.java
+++ b/geode-core/src/test/java/org/apache/geode/management/internal/security/GfshCommandsSecurityTest.java
@@ -14,15 +14,11 @@
*/
package org.apache.geode.management.internal.security;
-import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_MANAGER;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
import java.util.List;
import org.apache.shiro.authz.permission.WildcardPermission;
-import org.assertj.core.api.SoftAssertions;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Rule;
@@ -35,7 +31,7 @@ import org.apache.geode.management.cli.Result;
import org.apache.geode.management.internal.cli.result.CommandResult;
import org.apache.geode.management.internal.cli.result.ErrorResultData;
import org.apache.geode.management.internal.cli.result.ResultBuilder;
-import org.apache.geode.security.TestSecurityManager;
+import org.apache.geode.security.SimpleTestSecurityManager;
import org.apache.geode.test.dunit.rules.ConnectionConfiguration;
import org.apache.geode.test.dunit.rules.GfshShellConnectionRule;
import org.apache.geode.test.dunit.rules.ServerStarterRule;
@@ -45,11 +41,9 @@ import org.apache.geode.test.junit.categories.SecurityTest;
@Category({IntegrationTest.class, SecurityTest.class})
public class GfshCommandsSecurityTest {
@ClassRule
- public static ServerStarterRule serverStarter = new ServerStarterRule().withJMXManager()
- .withProperty(SECURITY_MANAGER, TestSecurityManager.class.getName())
- .withProperty("security-json",
- "org/apache/geode/management/internal/security/cacheServer.json")
- .withRegion(RegionShortcut.REPLICATE_PERSISTENT, "persistentRegion");
+ public static ServerStarterRule serverStarter =
+ new ServerStarterRule().withJMXManager().withSecurityManager(SimpleTestSecurityManager.class)
+ .withRegion(RegionShortcut.REPLICATE_PERSISTENT, "persistentRegion");
@Rule
public GfshShellConnectionRule gfshConnection = new GfshShellConnectionRule(
@@ -61,73 +55,73 @@ public class GfshCommandsSecurityTest {
}
@Test
- @ConnectionConfiguration(user = "data-admin", password = "wrongPwd")
+ @ConnectionConfiguration(user = "data", password = "wrongPwd")
public void testInvalidCredentials() throws Exception {
- assertFalse(gfshConnection.isConnected());
+ assertThat(gfshConnection.isConnected()).isFalse();
}
@Test
- @ConnectionConfiguration(user = "data-admin", password = "1234567")
+ @ConnectionConfiguration(user = "data", password = "data")
public void testValidCredentials() throws Exception {
- assertTrue(gfshConnection.isConnected());
+ assertThat(gfshConnection.isConnected()).isTrue();
}
@Test
- @ConnectionConfiguration(user = "cluster-reader", password = "1234567")
+ @ConnectionConfiguration(user = "clusterRead", password = "clusterRead")
public void testClusterReader() throws Exception {
runCommandsPermittedAndForbiddenBy("CLUSTER:READ");
}
@Test
- @ConnectionConfiguration(user = "cluster-writer", password = "1234567")
+ @ConnectionConfiguration(user = "clusterWrite", password = "clusterWrite")
public void testClusterWriter() throws Exception {
runCommandsPermittedAndForbiddenBy("CLUSTER:WRITE");
}
@Test
- @ConnectionConfiguration(user = "cluster-manager", password = "1234567")
+ @ConnectionConfiguration(user = "clusterManage", password = "clusterManage")
public void testClusterManager() throws Exception {
runCommandsPermittedAndForbiddenBy("CLUSTER:MANAGE");
}
@Test
- @ConnectionConfiguration(user = "data-reader", password = "1234567")
+ @ConnectionConfiguration(user = "dataRead", password = "dataRead")
public void testDataReader() throws Exception {
runCommandsPermittedAndForbiddenBy("DATA:READ");
}
@Test
- @ConnectionConfiguration(user = "data-writer", password = "1234567")
+ @ConnectionConfiguration(user = "dataWrite", password = "dataWrite")
public void testDataWriter() throws Exception {
runCommandsPermittedAndForbiddenBy("DATA:WRITE");
}
@Test
- @ConnectionConfiguration(user = "data-manager", password = "1234567")
+ @ConnectionConfiguration(user = "dataManage", password = "dataManage")
public void testDataManager() throws Exception {
runCommandsPermittedAndForbiddenBy("DATA:MANAGE");
}
@Test
- @ConnectionConfiguration(user = "regionA-reader", password = "1234567")
+ @ConnectionConfiguration(user = "dataReadRegionA", password = "dataReadRegionA")
public void testRegionAReader() throws Exception {
runCommandsPermittedAndForbiddenBy("DATA:READ:RegionA");
}
@Test
- @ConnectionConfiguration(user = "regionA-writer", password = "1234567")
+ @ConnectionConfiguration(user = "dataWriteRegionA", password = "dataWriteRegionA")
public void testRegionAWriter() throws Exception {
runCommandsPermittedAndForbiddenBy("DATA:WRITE:RegionA");
}
@Test
- @ConnectionConfiguration(user = "regionA-manager", password = "1234567")
+ @ConnectionConfiguration(user = "dataManageRegionA", password = "dataManageRegionA")
public void testRegionAManager() throws Exception {
runCommandsPermittedAndForbiddenBy("DATA:MANAGE:RegionA");
}
@Test
- @ConnectionConfiguration(user = "super-user", password = "1234567")
+ @ConnectionConfiguration(user = "data,cluster", password = "data,cluster")
public void testRegionSuperUser() throws Exception {
runCommandsPermittedAndForbiddenBy("*");
}
@@ -135,18 +129,17 @@ public class GfshCommandsSecurityTest {
private void runCommandsPermittedAndForbiddenBy(String permission) throws Exception {
List<TestCommand> allPermitted =
TestCommand.getPermittedCommands(new WildcardPermission(permission, true));
- SoftAssertions softly = new SoftAssertions();
for (TestCommand permitted : allPermitted) {
System.out.println("Processing authorized command: " + permitted.getCommand());
CommandResult result = gfshConnection.executeCommand(permitted.getCommand());
- assertNotNull(result);
+ assertThat(result).isNotNull();
if (result.getResultData() instanceof ErrorResultData) {
- softly.assertThat(ResultBuilder.ERRORCODE_UNAUTHORIZED).describedAs(permitted.getCommand())
+ assertThat(ResultBuilder.ERRORCODE_UNAUTHORIZED).describedAs(permitted.getCommand())
.isNotEqualTo(((ErrorResultData) result.getResultData()).getErrorCode());
} else {
- softly.assertThat(Result.Status.OK).describedAs(permitted.toString())
+ assertThat(Result.Status.OK).describedAs(permitted.toString())
.isEqualTo(result.getStatus());
}
}
@@ -167,16 +160,13 @@ public class GfshCommandsSecurityTest {
continue;
}
- softly.assertThat(ResultBuilder.ERRORCODE_UNAUTHORIZED).describedAs(other.getCommand())
+ assertThat(ResultBuilder.ERRORCODE_UNAUTHORIZED).describedAs(other.getCommand())
.isEqualTo(((ErrorResultData) result.getResultData()).getErrorCode());
}
-
- softly.assertAll();
-
}
@Test
- @ConnectionConfiguration(user = "data-user", password = "1234567")
+ @ConnectionConfiguration(user = "data", password = "data")
public void testGetPostProcess() throws Exception {
gfshConnection.executeCommand("put --region=region1 --key=key2 --value=value2");
gfshConnection.executeCommand("put --region=region1 --key=key2 --value=value2");
@@ -186,4 +176,39 @@ public class GfshCommandsSecurityTest {
gfshConnection.executeCommand("query --query=\"select * from /region1\"");
}
+ @Test
+ @ConnectionConfiguration(user = "data", password = "data")
+ public void createDiskStore() throws Exception {
+ CommandResult result =
+ gfshConnection.executeCommand("create disk-store --name=disk1 --dir=disk1");
+
+ assertThat(result.getContent().toString()).contains("not authorized for CLUSTER:MANAGE:DISK");
+ }
+
+ @Test
+ @ConnectionConfiguration(user = "dataManage,clusterWriteDisk",
+ password = "dataManage,clusterWriteDisk")
+ public void createPartitionedPersistentRegionWithCorrectPermissions() throws Exception {
+ gfshConnection
+ .executeAndVerifyCommand("create region --name=region2 --type=PARTITION_PERSISTENT");
+ }
+
+ @Test
+ @ConnectionConfiguration(user = "dataManage", password = "dataManage")
+ public void createPartitionedPersistentRegionWithoutClusterWriteDisk() throws Exception {
+ CommandResult result =
+ gfshConnection.executeCommand("create region --name=region2 --type=PARTITION_PERSISTENT");
+
+ assertThat(result.getContent().toString()).contains("not authorized for CLUSTER:WRITE:DISK");
+ }
+
+ @Test
+ @ConnectionConfiguration(user = "clusterWriteDisk", password = "clusterWriteDisk")
+ public void createPartitionedPersistentRegionWithoutDataManage() throws Exception {
+ CommandResult result =
+ gfshConnection.executeCommand("create region --name=region2 --type=PARTITION_PERSISTENT");
+
+ assertThat(result.getContent().toString()).contains("not authorized for DATA:MANAGE");
+ }
+
}
[2/9] geode git commit: GEODE-2860: Refactor use of EventTracker
Posted by kh...@apache.org.
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTaskTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTaskTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTaskTest.java
new file mode 100644
index 0000000..1f969ef
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/event/EventTrackerExpiryTaskTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.event;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.CancelCriterion;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class EventTrackerExpiryTaskTest {
+ private static final long TIME_TO_LIVE = 100;
+
+ private EventTrackerExpiryTask task;
+
+ @Before
+ public void setup() {
+ task = new EventTrackerExpiryTask(TIME_TO_LIVE);
+ }
+
+ @Test
+ public void hasNoTrackersWhenInitialized() {
+ assertEquals(0, task.getNumberOfTrackers());
+ }
+
+ @Test
+ public void addsTrackersCorrectly() {
+ task.addTracker(NonDistributedEventTracker.getInstance());
+ assertEquals(1, task.getNumberOfTrackers());
+ }
+
+ @Test
+ public void removedTrackersCorrectly() {
+ task.addTracker(NonDistributedEventTracker.getInstance());
+ task.removeTracker(NonDistributedEventTracker.getInstance());
+ assertEquals(0, task.getNumberOfTrackers());
+ }
+
+ @Test
+ public void removesExpiredSequenceIdHolder() {
+ DistributedEventTracker tracker = constructTestTracker();
+ task.addTracker(tracker);
+ EventSequenceNumberHolder sequenceIdHolder = new EventSequenceNumberHolder(0L, null);
+ tracker.recordSequenceNumber(new ThreadIdentifier(new byte[0], 0L), sequenceIdHolder);
+ sequenceIdHolder.setEndOfLifeTimestamp(System.currentTimeMillis() - TIME_TO_LIVE);
+ task.run2();
+ assertEquals(0, tracker.getRecordedEvents().size());
+ }
+
+ @Test
+ public void doesNotRemoveNonExpiredSequenceIdHolder() {
+ DistributedEventTracker tracker = constructTestTracker();
+ task.addTracker(tracker);
+ EventSequenceNumberHolder sequenceIdHolder = new EventSequenceNumberHolder(0L, null);
+ tracker.recordSequenceNumber(new ThreadIdentifier(new byte[0], 0L), sequenceIdHolder);
+ sequenceIdHolder.setEndOfLifeTimestamp(System.currentTimeMillis() + 10000);
+ task.run2();
+ assertEquals(1, tracker.getRecordedEvents().size());
+ }
+
+ @Test
+ public void doesNotRemoveNewSequenceIdHolder() {
+ DistributedEventTracker tracker = constructTestTracker();
+ task.addTracker(tracker);
+ EventSequenceNumberHolder sequenceIdHolder = new EventSequenceNumberHolder(0L, null);
+ tracker.recordSequenceNumber(new ThreadIdentifier(new byte[0], 0L), sequenceIdHolder);
+ task.run2();
+ assertEquals(1, tracker.getRecordedEvents().size());
+ }
+
+ private DistributedEventTracker constructTestTracker() {
+ return new DistributedEventTracker(mock(InternalCache.class), mock(CancelCriterion.class),
+ "test region");
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/java/org/apache/geode/internal/cache/event/NonDistributedEventTrackerTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/event/NonDistributedEventTrackerTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/event/NonDistributedEventTrackerTest.java
new file mode 100644
index 0000000..7ba531f
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/event/NonDistributedEventTrackerTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.event;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class NonDistributedEventTrackerTest {
+ private NonDistributedEventTracker tracker = NonDistributedEventTracker.getInstance();
+
+ @Test
+ public void getStateReturnsNull() {
+ assertNull(tracker.getState());
+ }
+
+ @Test
+ public void hasSeenEventReturnsFalse() {
+ assertFalse(tracker.hasSeenEvent(mock(InternalCacheEvent.class)));
+ assertFalse(tracker.hasSeenEvent(mock(EventID.class)));
+ assertFalse(tracker.hasSeenEvent(mock(EventID.class), mock(InternalCacheEvent.class)));
+ }
+
+ @Test
+ public void findVersionTagForSequenceReturnsNull() {
+ assertNull(tracker.findVersionTagForSequence(mock(EventID.class)));
+ }
+
+ @Test
+ public void findVersionTagForBulkOpReturnsNull() {
+ assertNull(tracker.findVersionTagForBulkOp(mock(EventID.class)));
+ }
+
+ @Test
+ public void returnsCorrectName() {
+ assertEquals(NonDistributedEventTracker.NAME, tracker.getName());
+ }
+
+ @Test
+ public void syncBulkOpExecutesProvidedRunnable() {
+ Runnable runnable = mock(Runnable.class);
+ tracker.syncBulkOp(runnable, mock(EventID.class), false);
+ tracker.syncBulkOp(runnable, mock(EventID.class), true);
+ verify(runnable, times(2)).run();
+ }
+
+ @Test
+ public void isInitializedReturnsTrue() {
+ assertTrue(tracker.isInitialized());
+ }
+
+ @Test
+ public void isInitialImageProviderReturnsFalse() {
+ assertFalse(tracker.isInitialImageProvider(mock(DistributedMember.class)));
+ }
+
+ @Test
+ public void getRecordedBulkOpVersionTagsReturnsNull() {
+ assertNull(tracker.getRecordedBulkOpVersionTags());
+ }
+
+ @Test
+ public void getRecordedEventsReturnsNull() {
+ assertNull(tracker.getRecordedEvents());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
----------------------------------------------------------------------
diff --git a/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt b/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
index b84ac39..e2a708d 100644
--- a/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
+++ b/geode-core/src/test/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt
@@ -1103,8 +1103,8 @@ fromData,24,2a2bb7001e2a2bb8001fb500042a2bb900200100b50002b1
toData,24,2a2bb700212ab400042bb800222b2ab40002b900230200b1
org/apache/geode/internal/cache/InitialImageOperation$RequestImageMessage,2
-fromData,98,2a2bb700b42a2bb800b5b500172a2bb900b60100b500022a2bb900b70100b500552a2bb900b70100b500032a2bb900b70100b5004b2a2bb800b8c000b9b5003a2a2bb800b8c000bab500262a2bb800b8c000bbb5003b2a2bb800b8c000bcb50030b1
-toData,86,2a2bb700bd2ab400172bb800be2b2ab40002b900bf02002b2ab40055b900c002002b2ab40003b900c002002b2ab4004bb900c002002ab4003a2bb800c12ab400262bb800c12ab4003b2bb800c12ab400302bb800c1b1
+fromData,98,2a2bb700b52a2bb800b6b500172a2bb900b70100b500022a2bb900b80100b500552a2bb900b80100b500032a2bb900b80100b5004b2a2bb800b9c000bab5003a2a2bb800b9c000bbb500262a2bb800b9c000bcb5003b2a2bb800b9c000bdb50030b1
+toData,86,2a2bb700be2ab400172bb800bf2b2ab40002b900c002002b2ab40055b900c102002b2ab40003b900c102002b2ab4004bb900c102002ab4003a2bb800c22ab400262bb800c22ab4003b2bb800c22ab400302bb800c2b1
org/apache/geode/internal/cache/InitialImageOperation$RequestRVVMessage,2
fromData,34,2a2bb7001e2a2bb8001fb500052a2bb900200100b500022a2bb900210100b50003b1
@@ -1447,6 +1447,10 @@ org/apache/geode/internal/cache/control/ResourceAdvisor$ResourceProfileMessage,2
fromData,84,2a2bb7001d2a2bb9001e0100b500032bb9001e01003d1c029f00362a1cbd0004b50005033e1d2ab40005bea20020bb000459b7001f3a0419042bb800202ab400051d190453840301a7ffdda700082a01b50005b1
toData,71,2a2bb700212b2ab40003b9002202002ab40005c6002c2b2ab40005beb900220200033d1c2ab40005bea200132ab400051c322bb80023840201a7ffeaa7000a2b02b900220200b1
+org/apache/geode/internal/cache/event/EventSequenceNumberHolder,2
+fromData,22,2a2bb900100100b500042a2bb80011c00012b50005b1
+toData,19,2b2ab40004b9001303002ab400052bb80014b1
+
org/apache/geode/internal/cache/execute/FunctionRemoteContext,2
fromData,95,2bb800084d2cc100099900252a03b500072a2cc00009b8000ab500022ab40002c7001b2a2cc00009b5000ba700102a2cc0000cb500022a04b500072a2bb80008b500032a2bb8000db500042a2bb8000db500052a2bb8000eb6000fb50006b1
toData,73,2ab4000799000e2ab400022bb80010a700102ab40002b9001101002bb800102ab400032bb800102ab40004c000122bb800132ab40005c000122bb800132ab40006b800142bb80015b1
[4/9] geode git commit: GEODE-2860: Refactor use of EventTracker
Posted by kh...@apache.org.
GEODE-2860: Refactor use of EventTracker
* change EventTracker to an interface with two implementations
* move as much logic out of LocalRegion down into subclasses that
make use EventTracker
* move and refactor static inner classes in EventTracker into own
class files
* migrate some of event-focused classes into a new sub package
* add tests for existing logic from EventTracker
This closes #638
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/3f9be9a7
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/3f9be9a7
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/3f9be9a7
Branch: refs/heads/develop
Commit: 3f9be9a74fcdb1152692d72720ed2d78efec41d4
Parents: d02bf86
Author: Nick Reich <nr...@pivotal.io>
Authored: Fri Jun 30 16:05:58 2017 -0700
Committer: Ken Howe <kh...@pivotal.io>
Committed: Tue Jul 18 07:11:33 2017 -0700
----------------------------------------------------------------------
.../geode/internal/cache/BucketRegion.java | 36 +-
.../internal/cache/CreateRegionProcessor.java | 22 +-
.../geode/internal/cache/DistributedRegion.java | 46 +-
.../geode/internal/cache/EventStateHelper.java | 10 +-
.../geode/internal/cache/EventTracker.java | 790 -------------------
.../internal/cache/FindVersionTagOperation.java | 2 +-
.../geode/internal/cache/GemFireCacheImpl.java | 24 +-
.../apache/geode/internal/cache/HARegion.java | 5 +-
.../geode/internal/cache/InternalCache.java | 3 +-
.../geode/internal/cache/LocalRegion.java | 167 +---
.../geode/internal/cache/PartitionedRegion.java | 9 +-
.../cache/event/BulkOperationHolder.java | 79 ++
.../cache/event/DistributedEventTracker.java | 523 ++++++++++++
.../cache/event/EventSequenceNumberHolder.java | 124 +++
.../internal/cache/event/EventTracker.java | 136 ++++
.../cache/event/EventTrackerExpiryTask.java | 97 +++
.../cache/event/NonDistributedEventTracker.java | 135 ++++
.../cache/tier/sockets/BaseCommand.java | 4 +-
.../wan/serial/SerialGatewaySenderQueue.java | 6 +-
.../internal/cache/xmlcache/CacheCreation.java | 4 +-
.../internal/cache/BucketRegionJUnitTest.java | 4 +
.../cache/DistributedRegionJUnitTest.java | 11 +-
.../internal/cache/EventTrackerDUnitTest.java | 486 ------------
.../geode/internal/cache/EventTrackerTest.java | 94 ---
.../geode/internal/cache/IteratorDUnitTest.java | 2 +-
.../cache/PartitionedRegionDUnitTestCase.java | 4 +-
...onedRegionSingleNodeOperationsJUnitTest.java | 11 +-
.../event/DistributedEventTrackerTest.java | 328 ++++++++
.../cache/event/EventTrackerDUnitTest.java | 489 ++++++++++++
.../cache/event/EventTrackerExpiryTaskTest.java | 94 +++
.../event/NonDistributedEventTrackerTest.java | 89 +++
.../sanctionedDataSerializables.txt | 8 +-
32 files changed, 2262 insertions(+), 1580 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
index 31b341a..30ce9e7 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
@@ -30,7 +30,7 @@ import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.Version;
import org.apache.geode.internal.cache.BucketAdvisor.BucketProfile;
import org.apache.geode.internal.cache.CreateRegionProcessor.CreateRegionReplyProcessor;
-import org.apache.geode.internal.cache.EventTracker.EventSeqnoHolder;
+import org.apache.geode.internal.cache.event.EventSequenceNumberHolder;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
@@ -280,12 +280,6 @@ public class BucketRegion extends DistributedRegion implements Bucket {
}
@Override
- public void createEventTracker() {
- this.eventTracker = new EventTracker(this);
- this.eventTracker.start();
- }
-
- @Override
public void registerCreateRegionReplyProcessor(CreateRegionReplyProcessor processor) {
this.createRegionReplyProcessor = processor;
}
@@ -293,7 +287,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
@Override
protected void recordEventStateFromImageProvider(InternalDistributedMember provider) {
if (this.createRegionReplyProcessor != null) {
- Map<ThreadIdentifier, EventSeqnoHolder> providerEventStates =
+ Map<ThreadIdentifier, EventSequenceNumberHolder> providerEventStates =
this.createRegionReplyProcessor.getEventState(provider);
if (providerEventStates != null) {
recordEventState(provider, providerEventStates);
@@ -1632,7 +1626,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
if (this.isInitialized()) {
boolean callThem = callDispatchListenerEvent;
if (event.isPossibleDuplicate()
- && this.eventTracker.isInitialImageProvider(event.getDistributedMember())) {
+ && getEventTracker().isInitialImageProvider(event.getDistributedMember())) {
callThem = false;
}
super.invokeTXCallbacks(eventType, event, callThem);
@@ -1664,7 +1658,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
if (this.isInitialized()) {
boolean callThem = callDispatchListenerEvent;
if (event.isPossibleDuplicate()
- && this.eventTracker.isInitialImageProvider(event.getDistributedMember())) {
+ && this.getEventTracker().isInitialImageProvider(event.getDistributedMember())) {
callThem = false;
}
super.invokeDestroyCallbacks(eventType, event, callThem, notifyGateways);
@@ -1695,7 +1689,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
if (this.isInitialized()) {
boolean callThem = callDispatchListenerEvent;
if (event.isPossibleDuplicate()
- && this.eventTracker.isInitialImageProvider(event.getDistributedMember())) {
+ && this.getEventTracker().isInitialImageProvider(event.getDistributedMember())) {
callThem = false;
}
super.invokeInvalidateCallbacks(eventType, event, callThem);
@@ -1729,7 +1723,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
if (this.isInitialized()) {
boolean callThem = callDispatchListenerEvent;
if (callThem && event.isPossibleDuplicate()
- && this.eventTracker.isInitialImageProvider(event.getDistributedMember())) {
+ && this.getEventTracker().isInitialImageProvider(event.getDistributedMember())) {
callThem = false;
}
super.invokePutCallbacks(eventType, event, callThem, notifyGateways);
@@ -2439,5 +2433,23 @@ public class BucketRegion extends DistributedRegion implements Bucket {
return getPartitionedRegion().notifiesMultipleSerialGateways();
}
+ @Override
+ public boolean hasSeenEvent(EntryEventImpl event) {
+ ensureEventTrackerInitialization();
+ return super.hasSeenEvent(event);
+ }
+
+ // bug 41289 - wait for event tracker to be initialized before checkin
+ // so that an operation intended for a previous version of a bucket
+ // is not prematurely applied to a new version of the bucket
+ private void ensureEventTrackerInitialization() {
+ try {
+ getEventTracker().waitOnInitialization();
+ } catch (InterruptedException ie) {
+ stopper.checkCancelInProgress(ie);
+ Thread.currentThread().interrupt();
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/CreateRegionProcessor.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/CreateRegionProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/CreateRegionProcessor.java
index 1e38065..ee4e2df 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/CreateRegionProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/CreateRegionProcessor.java
@@ -49,7 +49,7 @@ import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.InitialImageAdvice;
-import org.apache.geode.internal.cache.EventTracker.EventSeqnoHolder;
+import org.apache.geode.internal.cache.event.EventSequenceNumberHolder;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
@@ -91,10 +91,8 @@ public class CreateRegionProcessor implements ProfileExchangeProcessor {
logger.debug("CreateRegionProcessor.initializeRegion, no recipients, msg not sent");
}
this.newRegion.getDistributionAdvisor().setInitialized();
- EventTracker tracker = ((LocalRegion) this.newRegion).getEventTracker();
- if (tracker != null) {
- tracker.setInitialized();
- }
+
+ ((LocalRegion) this.newRegion).getEventTracker().setInitialized();
return;
}
@@ -138,13 +136,11 @@ public class CreateRegionProcessor implements ProfileExchangeProcessor {
}
} finally {
replyProc.cleanup();
- EventTracker tracker = ((LocalRegion) this.newRegion).getEventTracker();
- if (tracker != null) {
- tracker.setInitialized();
- }
+ ((LocalRegion) this.newRegion).getEventTracker().setInitialized();
if (((LocalRegion) this.newRegion).isUsedForPartitionedRegionBucket()) {
if (logger.isDebugEnabled()) {
- logger.debug("initialized bucket event tracker: {}", tracker);
+ logger.debug("initialized bucket event tracker: {}",
+ ((LocalRegion) this.newRegion).getEventTracker());
}
}
}
@@ -203,12 +199,12 @@ public class CreateRegionProcessor implements ProfileExchangeProcessor {
.getDistributedSystem(), members);
}
- private final Map<DistributedMember, Map<ThreadIdentifier, EventSeqnoHolder>> remoteEventStates =
+ private final Map<DistributedMember, Map<ThreadIdentifier, EventSequenceNumberHolder>> remoteEventStates =
new ConcurrentHashMap<>();
private boolean allMembersSkippedChecks = true;
- public Map<ThreadIdentifier, EventSeqnoHolder> getEventState(
+ public Map<ThreadIdentifier, EventSequenceNumberHolder> getEventState(
InternalDistributedMember provider) {
return this.remoteEventStates.get(provider);
}
@@ -254,7 +250,7 @@ public class CreateRegionProcessor implements ProfileExchangeProcessor {
// Save all event states, need to initiate the event tracker from the GII provider
if (reply.eventState != null) {
remoteEventStates.put(reply.getSender(),
- (Map<ThreadIdentifier, EventSeqnoHolder>) reply.eventState);
+ (Map<ThreadIdentifier, EventSequenceNumberHolder>) reply.eventState);
}
if (lr.isUsedForPartitionedRegionBucket()) {
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java
index eb18134..d9ed4ed 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedRegion.java
@@ -90,6 +90,8 @@ import org.apache.geode.internal.cache.InitialImageOperation.GIIStatus;
import org.apache.geode.internal.cache.RemoteFetchVersionMessage.FetchVersionResponse;
import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceType;
import org.apache.geode.internal.cache.control.MemoryEvent;
+import org.apache.geode.internal.cache.event.DistributedEventTracker;
+import org.apache.geode.internal.cache.event.EventTracker;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionExecutor;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultWaiter;
@@ -98,7 +100,6 @@ import org.apache.geode.internal.cache.execute.LocalResultCollector;
import org.apache.geode.internal.cache.execute.RegionFunctionContextImpl;
import org.apache.geode.internal.cache.execute.ServerToClientFunctionResultSender;
import org.apache.geode.internal.cache.lru.LRUEntry;
-import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.persistence.CreatePersistentRegionProcessor;
import org.apache.geode.internal.cache.persistence.PersistenceAdvisor;
import org.apache.geode.internal.cache.persistence.PersistenceAdvisorImpl;
@@ -255,9 +256,10 @@ public class DistributedRegion extends LocalRegion implements CacheDistributionA
}
@Override
- public void createEventTracker() {
- this.eventTracker = new EventTracker(this);
- this.eventTracker.start();
+ protected EventTracker createEventTracker() {
+ EventTracker tracker = new DistributedEventTracker(cache, stopper, getName());
+ tracker.start();
+ return tracker;
}
/**
@@ -493,6 +495,38 @@ public class DistributedRegion extends LocalRegion implements CacheDistributionA
}
}
+ @Override
+ public boolean hasSeenEvent(EntryEventImpl event) {
+ boolean isDuplicate = false;
+
+ isDuplicate = getEventTracker().hasSeenEvent(event);
+ if (isDuplicate) {
+ markEventAsDuplicate(event);
+ } else {
+ // bug #48205 - a retried PR operation may already have a version assigned to it
+ // in another VM
+ if (event.isPossibleDuplicate() && event.getRegion().concurrencyChecksEnabled
+ && event.getVersionTag() == null && event.getEventId() != null) {
+ boolean isBulkOp = event.getOperation().isPutAll() || event.getOperation().isRemoveAll();
+ VersionTag tag =
+ FindVersionTagOperation.findVersionTag(event.getRegion(), event.getEventId(), isBulkOp);
+ event.setVersionTag(tag);
+ }
+ }
+ return isDuplicate;
+ }
+
+ private void markEventAsDuplicate(EntryEventImpl event) {
+ event.setPossibleDuplicate(true);
+ if (concurrencyChecksEnabled && event.getVersionTag() == null) {
+ if (event.isBulkOpInProgress()) {
+ event.setVersionTag(getEventTracker().findVersionTagForBulkOp(event.getEventId()));
+ } else {
+ event.setVersionTag(getEventTracker().findVersionTagForSequence(event.getEventId()));
+ }
+ }
+ }
+
void setGeneratedVersionTag(boolean generateVersionTag) {
// there is at-least one other persistent member, so turn on concurrencyChecks
enableConcurrencyChecks();
@@ -1038,9 +1072,7 @@ public class DistributedRegion extends LocalRegion implements CacheDistributionA
// makes sure all latches are released if they haven't been already
super.initialize(null, null, null);
} finally {
- if (this.eventTracker != null) {
- this.eventTracker.setInitialized();
- }
+ getEventTracker().setInitialized();
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/EventStateHelper.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/EventStateHelper.java b/geode-core/src/main/java/org/apache/geode/internal/cache/EventStateHelper.java
index 9bc5e3a..ec427bb 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/EventStateHelper.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/EventStateHelper.java
@@ -19,7 +19,7 @@ import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.CreateRegionProcessor.CreateRegionReplyMessage;
-import org.apache.geode.internal.cache.EventTracker.EventSeqnoHolder;
+import org.apache.geode.internal.cache.event.EventSequenceNumberHolder;
import org.apache.geode.internal.cache.InitialImageOperation.RegionStateMessage;
import org.apache.geode.internal.cache.ha.HARegionQueue.DispatchedAndCurrentEvents;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
@@ -89,7 +89,7 @@ public class EventStateHelper {
DispatchedAndCurrentEvents value = (DispatchedAndCurrentEvents) entry.getValue();
InternalDataSerializer.invokeToData(value, dop);
} else {
- EventSeqnoHolder value = (EventSeqnoHolder) entry.getValue();
+ EventSequenceNumberHolder value = (EventSequenceNumberHolder) entry.getValue();
InternalDataSerializer.invokeToData(value, dop);
}
}
@@ -130,11 +130,11 @@ public class EventStateHelper {
InternalDataSerializer.invokeFromData(value, dip);
eventState.put(key, value);
} else {
- EventSeqnoHolder value = new EventSeqnoHolder();
+ EventSequenceNumberHolder value = new EventSequenceNumberHolder();
InternalDataSerializer.invokeFromData(value, dip);
eventState.put(key, value);
- if (value.versionTag != null) {
- value.versionTag.replaceNullIDs(senderId);
+ if (value.getVersionTag() != null) {
+ value.getVersionTag().replaceNullIDs(senderId);
}
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/EventTracker.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/EventTracker.java b/geode-core/src/main/java/org/apache/geode/internal/cache/EventTracker.java
deleted file mode 100644
index b919043..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/EventTracker.java
+++ /dev/null
@@ -1,790 +0,0 @@
-/*
- * 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;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
-import org.apache.logging.log4j.Logger;
-
-import org.apache.geode.DataSerializable;
-import org.apache.geode.DataSerializer;
-import org.apache.geode.cache.client.PoolFactory;
-import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
-import org.apache.geode.internal.Assert;
-import org.apache.geode.internal.SystemTimer.SystemTimerTask;
-import org.apache.geode.internal.cache.ha.ThreadIdentifier;
-import org.apache.geode.internal.cache.versions.RegionVersionVector;
-import org.apache.geode.internal.cache.versions.VersionTag;
-import org.apache.geode.internal.logging.LogService;
-import org.apache.geode.internal.logging.log4j.LogMarker;
-import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
-
-/**
- * EventTracker tracks the last sequence number for a particular memberID:threadID. It is used to
- * avoid replaying events in client/server and partitioned-region configurations.
- *
- * @since GemFire 6.0
- */
-public class EventTracker {
- private static final Logger logger = LogService.getLogger();
-
- /**
- * a mapping of originator to the last event applied to this cache
- *
- * Keys are instances of {@link ThreadIdentifier}, values are instances of
- * {@link org.apache.geode.internal.cache.EventTracker.EventSeqnoHolder}.
- */
- protected final ConcurrentMap<ThreadIdentifier, EventSeqnoHolder> recordedEvents =
- new ConcurrentHashMap<ThreadIdentifier, EventSeqnoHolder>(100);
-
- /**
- * a mapping of originator to bulkOps
- *
- * Keys are instances of @link {@link ThreadIdentifier}
- */
- private final ConcurrentMap<ThreadIdentifier, Object> recordedBulkOps =
- new ConcurrentHashMap<ThreadIdentifier, Object>(100);
-
- /**
- * a mapping of originator to bulkOperation's last version tags. This map differs from
- * {@link #recordedBulkOps} in that the thread identifier used here is the base member id and
- * thread id of the bulk op, as opposed to the fake thread id which is assigned for each bucket.
- *
- * recordedBulkOps are also only tracked on the secondary for partitioned regions
- * recordedBulkOpVersionTags are tracked on both the primary and secondary.
- *
- * Keys are instances of @link {@link ThreadIdentifier}, values are instances of
- * {@link BulkOpHolder}.
- */
- private final ConcurrentMap<ThreadIdentifier, BulkOpHolder> recordedBulkOpVersionTags =
- new ConcurrentHashMap<ThreadIdentifier, BulkOpHolder>(100);
-
- /**
- * The member that the region corresponding to this tracker (if any) received its initial image
- * from (if a replicate)
- */
- private volatile InternalDistributedMember initialImageProvider;
-
- /**
- * The cache associated with this tracker
- */
- InternalCache cache;
-
- /**
- * The name of this tracker
- */
- String name;
-
- /**
- * whether or not this tracker has been initialized to allow entry operation. replicate region
- * does not initiate event tracker from its replicates.
- */
- volatile boolean initialized;
-
- /**
- * object used to wait for initialization
- */
- final StoppableCountDownLatch initializationLatch;
-
- /**
- * Initialize the EventTracker's timer task. This is stored in the cache for tracking and shutdown
- * purposes
- *
- * @param cache the cache to schedule tasks with
- */
- public static ExpiryTask startTrackerServices(InternalCache cache) {
- long expiryTime = Long.getLong(DistributionConfig.GEMFIRE_PREFIX + "messageTrackingTimeout",
- PoolFactory.DEFAULT_SUBSCRIPTION_MESSAGE_TRACKING_TIMEOUT / 3);
- ExpiryTask result = new ExpiryTask(cache, expiryTime);
- cache.getCCPTimer().scheduleAtFixedRate(result, expiryTime, expiryTime);
- // schedule(result, expiryTime);
- return result;
- }
-
- /**
- * Terminate the tracker's timer task
- *
- * @param cache the cache holding the tracker task
- */
- public static void stopTrackerServices(InternalCache cache) {
- cache.getEventTrackerTask().cancel();
- }
-
- /**
- * Create an event tracker
- *
- * @param region the cache region to associate with this tracker
- */
- public EventTracker(LocalRegion region) {
- this.cache = region.cache;
- this.name = "Event Tracker for " + region.getName();
- this.initializationLatch = new StoppableCountDownLatch(region.getStopper(), 1);
- }
-
- /** start this event tracker */
- public void start() {
- if (this.cache.getEventTrackerTask() != null) {
- this.cache.getEventTrackerTask().addTracker(this);
- }
- }
-
- /** stop this event tracker */
- public void stop() {
- if (this.cache.getEventTrackerTask() != null) {
- this.cache.getEventTrackerTask().removeTracker(this);
- }
- }
-
- /**
- * retrieve a deep copy of the state of the event tracker. Synchronization is not used while
- * copying the tracker's state.
- */
- public Map<ThreadIdentifier, EventSeqnoHolder> getState() {
- Map<ThreadIdentifier, EventSeqnoHolder> result =
- new HashMap<ThreadIdentifier, EventSeqnoHolder>(recordedEvents.size());
- for (Iterator<Map.Entry<ThreadIdentifier, EventSeqnoHolder>> it =
- recordedEvents.entrySet().iterator(); it.hasNext();) {
- Map.Entry<ThreadIdentifier, EventSeqnoHolder> entry = it.next();
- EventSeqnoHolder holder = entry.getValue();
- result.put(entry.getKey(), new EventSeqnoHolder(holder.lastSeqno, null)); // don't transfer
- // version tags -
- // adds too much
- // bulk just so we
- // can do client tag
- // recovery
- }
- return result;
- }
-
- /**
- * record the given state in the tracker.
- *
- * @param provider the member that provided this state
- * @param state a Map obtained from getState();
- */
- public void recordState(InternalDistributedMember provider,
- Map<ThreadIdentifier, EventSeqnoHolder> state) {
- this.initialImageProvider = provider;
- StringBuffer sb = null;
- if (logger.isDebugEnabled()) {
- sb = new StringBuffer(200);
- sb.append("Recording initial state for ").append(this.name).append(": ");
- }
- for (Iterator<Map.Entry<ThreadIdentifier, EventSeqnoHolder>> it =
- state.entrySet().iterator(); it.hasNext();) {
- Map.Entry<ThreadIdentifier, EventSeqnoHolder> entry = it.next();
- if (sb != null) {
- sb.append("\n ").append(entry.getKey().expensiveToString()).append("; sequenceID=")
- .append(entry.getValue());
- }
- // record only if we haven't received an event that is newer
- recordSeqno(entry.getKey(), entry.getValue(), true);
- }
- if (sb != null) {
- logger.debug(sb);
- }
- // fix for bug 41622 - hang in GII. This keeps ops from waiting for the
- // full GII to complete
- setInitialized();
- }
-
- /**
- * Use this method to ensure that the tracker is put in an initialized state
- */
- public void setInitialized() {
- this.initializationLatch.countDown();
- this.initialized = true;
- }
-
- /**
- * Wait for the tracker to finishe being initialized
- */
- public void waitOnInitialization() throws InterruptedException {
- this.initializationLatch.await();
- }
-
- /**
- * Record an event sequence id if it is higher than what we currently have. This is intended for
- * use during initial image transfer.
- *
- * @param membershipID the key of an entry in the map obtained from getEventState()
- * @param evhObj the value of an entry in the map obtained from getEventState()
- */
- protected void recordSeqno(ThreadIdentifier membershipID, EventSeqnoHolder evhObj) {
- recordSeqno(membershipID, evhObj, false);
- }
-
- /**
- * Record an event sequence id if it is higher than what we currently have. This is intended for
- * use during initial image transfer.
- *
- * @param threadID the key of an entry in the map obtained from getEventState()
- * @param evh the value of an entry in the map obtained from getEventState()
- * @param ifAbsent only record this state if there's not already an entry for this memberID
- */
- private void recordSeqno(ThreadIdentifier threadID, EventSeqnoHolder evh, boolean ifAbsent) {
- boolean removed;
- if (logger.isDebugEnabled()) {
- logger.debug("recording {} {}", threadID.expensiveToString(), evh.toString());
- }
- do {
- removed = false;
- EventSeqnoHolder oldEvh = recordedEvents.putIfAbsent(threadID, evh);
- if (oldEvh != null) {
- synchronized (oldEvh) {
- if (oldEvh.removed) {
- // need to wait for an entry being removed by the sweeper to go away
- removed = true;
- continue;
- } else {
- if (ifAbsent) {
- break;
- }
- oldEvh.endOfLifeTimer = 0;
- if (oldEvh.lastSeqno < evh.lastSeqno) {
- oldEvh.lastSeqno = evh.lastSeqno;
- oldEvh.versionTag = evh.versionTag;
- // Exception e = oldEvh.context;
- // oldEvh.context = new Exception("stack trace");
- // oldEvh.context.initCause(e);
- }
- }
- }
- } else {
- evh.endOfLifeTimer = 0;
- // evh.context = new Exception("stack trace");
- }
- } while (removed);
- }
-
- /** record the event's threadid/sequenceid to prevent replay */
- public void recordEvent(InternalCacheEvent event) {
- EventID eventID = event.getEventId();
- if (ignoreEvent(event, eventID)) {
- return; // not tracked
- }
-
- LocalRegion lr = (LocalRegion) event.getRegion();
-
- ThreadIdentifier membershipID =
- new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
-
- VersionTag tag = null;
- if (lr.getServerProxy() == null/* && event.hasClientOrigin() */) { // clients do not need to
- // store version tags for
- // replayed events
- tag = event.getVersionTag();
- RegionVersionVector v = ((LocalRegion) event.getRegion()).getVersionVector();
- // bug #46453 - make sure ID references are canonical before storing
- if (v != null && tag != null) {
- tag.setMemberID(v.getCanonicalId(tag.getMemberID()));
- if (tag.getPreviousMemberID() != null) {
- tag.setPreviousMemberID(v.getCanonicalId(tag.getPreviousMemberID()));
- }
- }
- }
-
- EventSeqnoHolder newEvh = new EventSeqnoHolder(eventID.getSequenceID(), tag);
- if (logger.isTraceEnabled()) {
- logger.trace("region event tracker recording {}", event);
- }
- recordSeqno(membershipID, newEvh);
-
- // If this is a bulkOp, and concurrency checks are enabled, we need to
- // save the version tag in case we retry.
- // Make recordBulkOp version tag after recordSeqno, so that recordBulkOpStart
- // in a retry bulk op would not incorrectly remove the saved version tag in
- // recordedBulkOpVersionTags
- if (lr.getConcurrencyChecksEnabled()
- && (event.getOperation().isPutAll() || event.getOperation().isRemoveAll())
- && lr.getServerProxy() == null) {
- recordBulkOpEvent(event, membershipID);
- }
- }
-
- /**
- * Record a version tag for a bulk operation
- */
- private void recordBulkOpEvent(InternalCacheEvent event, ThreadIdentifier tid) {
- EventID eventID = event.getEventId();
-
- VersionTag tag = event.getVersionTag();
- if (tag == null) {
- return;
- }
-
- if (logger.isDebugEnabled()) {
- logger.debug("recording bulkOp event {} {} {} op={}", tid.expensiveToString(), eventID, tag,
- event.getOperation());
- }
-
- RegionVersionVector v = ((LocalRegion) event.getRegion()).getVersionVector();
- // bug #46453 - make sure ID references are canonical before storing
- if (v != null) {
- tag.setMemberID(v.getCanonicalId(tag.getMemberID()));
- if (tag.getPreviousMemberID() != null) {
- tag.setPreviousMemberID(v.getCanonicalId(tag.getPreviousMemberID()));
- }
- }
-
- // Loop until we can successfully update the recorded bulk operations
- // For this thread id.
- boolean retry = false;
- do {
- BulkOpHolder bulkOpTracker = recordedBulkOpVersionTags.get(tid);
- if (bulkOpTracker == null) {
- bulkOpTracker = new BulkOpHolder();
- BulkOpHolder old = recordedBulkOpVersionTags.putIfAbsent(tid, bulkOpTracker);
- if (old != null) {
- retry = true;
- continue;
- }
- }
- synchronized (bulkOpTracker) {
- if (bulkOpTracker.removed) {
- retry = true;
- continue;
- }
-
- // Add the version tag for bulkOp event.
- bulkOpTracker.putVersionTag(eventID, event.getVersionTag());
- retry = false;
- }
- } while (retry);
- }
-
- public boolean hasSeenEvent(InternalCacheEvent event) {
- // ClientProxyMembershipID membershipID = event.getContext();
- EventID eventID = event.getEventId();
- if (ignoreEvent(event, eventID)) {
- return false; // not tracked
- }
- return hasSeenEvent(eventID, event);
- }
-
- public boolean hasSeenEvent(EventID eventID) {
- return hasSeenEvent(eventID, null);
- }
-
- public boolean hasSeenEvent(EventID eventID, InternalCacheEvent tagHolder) {
- ThreadIdentifier membershipID =
- new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
- // if (membershipID == null || eventID == null) {
- // return false;
- // }
-
- EventSeqnoHolder evh = recordedEvents.get(membershipID);
- if (evh == null) {
- return false;
- }
-
- synchronized (evh) {
- if (evh.removed || evh.lastSeqno < eventID.getSequenceID()) {
- return false;
- }
- // log at fine because partitioned regions can send event multiple times
- // during normal operation during bucket region initialization
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_BRIDGE_SERVER)) {
- logger.trace(LogMarker.DISTRIBUTION_BRIDGE_SERVER,
- "Cache encountered replay of event with ID {}. Highest recorded for this source is {}",
- eventID, evh.lastSeqno);
- }
- // bug #44956 - recover version tag for duplicate event
- if (evh.lastSeqno == eventID.getSequenceID() && tagHolder != null && evh.versionTag != null) {
- ((EntryEventImpl) tagHolder).setVersionTag(evh.versionTag);
- }
- return true;
- } // synchronized
- }
-
- public VersionTag findVersionTag(EventID eventID) {
- ThreadIdentifier threadID =
- new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
-
- EventSeqnoHolder evh = recordedEvents.get(threadID);
- if (evh == null) {
- if (logger.isDebugEnabled()) {
- logger.debug("search for version tag failed as no event is recorded for {}",
- threadID.expensiveToString());
- }
- return null;
- }
-
- synchronized (evh) {
- if (logger.isDebugEnabled()) {
- logger.debug("search for version tag located last event for {}: {}",
- threadID.expensiveToString(), evh);
- }
- if (evh.lastSeqno != eventID.getSequenceID()) {
- return null;
- }
- // log at fine because partitioned regions can send event multiple times
- // during normal operation during bucket region initialization
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_BRIDGE_SERVER) && evh.versionTag == null) {
- logger.trace(LogMarker.DISTRIBUTION_BRIDGE_SERVER,
- "Could not recover version tag. Found event holder with no version tag for {}",
- eventID);
- }
- return evh.versionTag;
- } // synchronized
- }
-
- public VersionTag findVersionTagForGateway(EventID eventID) {
- ThreadIdentifier threadID =
- new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
-
- EventSeqnoHolder evh = recordedEvents.get(threadID);
- if (evh == null) {
- if (logger.isDebugEnabled()) {
- logger.debug("search for version tag failed as no event is recorded for {}",
- threadID.expensiveToString());
- }
- return null;
- }
-
- synchronized (evh) {
- if (logger.isDebugEnabled()) {
- logger.debug("search for version tag located last event for {}: {} {}",
- threadID.expensiveToString(), evh, eventID.getSequenceID());
- }
-
- if (evh.lastSeqno < eventID.getSequenceID()) {
- return null;
- }
- // log at fine because partitioned regions can send event multiple times
- // during normal operation during bucket region initialization
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_BRIDGE_SERVER) && evh.versionTag == null) {
- logger.trace(LogMarker.DISTRIBUTION_BRIDGE_SERVER,
- "Could not recover version tag. Found event holder with no version tag for {}",
- eventID);
- }
- return evh.versionTag;
- } // synchronized
- }
-
-
- public VersionTag findVersionTagForBulkOp(EventID eventID) {
- ThreadIdentifier threadID =
- new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
-
- BulkOpHolder evh = recordedBulkOpVersionTags.get(threadID);
- if (evh == null) {
- if (logger.isDebugEnabled()) {
- logger.debug("search for version tag failed as no events are recorded for {}",
- threadID.expensiveToString());
- }
- return null;
- }
-
- synchronized (evh) {
- if (logger.isDebugEnabled()) {
- logger.debug("search for version tag located event holder for {}: {}",
- threadID.expensiveToString(), evh);
- }
- return evh.entryVersionTags.get(eventID);
- } // synchronized
- }
-
- /**
- * @return true if the event should not be tracked, false otherwise
- */
- private boolean ignoreEvent(InternalCacheEvent event, EventID eventID) {
- if (eventID == null) {
- return true;
- } else {
- boolean isVersioned = (event.getVersionTag() != null);
- boolean isClient = event.hasClientOrigin();
- if (isVersioned && isClient) {
- return false; // version tags for client events are kept for retries by the client
- }
- boolean isEntry = event.getOperation().isEntry();
- boolean isPr = event.getRegion().getAttributes().getDataPolicy().withPartitioning()
- || ((LocalRegion) event.getRegion()).isUsedForPartitionedRegionBucket();
- return (!isClient && // ignore if it originated on a server, and
- isEntry && // it affects an entry and
- !isPr); // is not on a PR
- }
- }
-
- /**
- * A routine to provide synchronization running based on <memberShipID, threadID> of the
- * requesting client
- *
- * @param r - a Runnable to wrap the processing of the bulk op
- * @param eventID - the base event ID of the bulk op
- *
- * @since GemFire 5.7
- */
- public void syncBulkOp(Runnable r, EventID eventID) {
- Assert.assertTrue(eventID != null);
- ThreadIdentifier membershipID =
- new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
-
- Object opSyncObj = null;
- do {
- opSyncObj = recordedBulkOps.putIfAbsent(membershipID, new Object());
- if (opSyncObj == null) {
- opSyncObj = recordedBulkOps.get(membershipID);
- }
- } while (opSyncObj == null);
-
- synchronized (opSyncObj) {
- try {
- recordBulkOpStart(membershipID, eventID);
- // Perform the bulk op
- r.run();
- } finally {
- recordedBulkOps.remove(membershipID);
- }
- }
- }
-
- /**
- * Called when a new bulkOp is started on the local region. Used to clear event tracker state from
- * the last bulkOp.
- */
- public void recordBulkOpStart(ThreadIdentifier tid, EventID eventID) {
- if (logger.isDebugEnabled()) {
- logger.debug("recording bulkOp start for {}", tid.expensiveToString());
- }
- EventSeqnoHolder evh = recordedEvents.get(tid);
- if (evh == null) {
- return;
- }
- synchronized (evh) {
- // only remove it when a new bulk op occurs
- if (eventID.getSequenceID() > evh.lastSeqno) {
- this.recordedBulkOpVersionTags.remove(tid);
- }
- }
- }
-
- /**
- * @return the initialized
- */
- public boolean isInitialized() {
- return this.initialized;
- }
-
- /**
- * @param mbr the member in question
- * @return true if the given member provided the initial image event state for this tracker
- */
- public boolean isInitialImageProvider(DistributedMember mbr) {
- return (this.initialImageProvider != null) && (mbr != null)
- && this.initialImageProvider.equals(mbr);
- }
-
- /**
- * Test method for getting the set of recorded version tags.
- */
- protected ConcurrentMap<ThreadIdentifier, BulkOpHolder> getRecordedBulkOpVersionTags() {
- return recordedBulkOpVersionTags;
- }
-
- @Override
- public String toString() {
- return "" + this.name + "(initialized=" + this.initialized + ")";
- }
-
- /**
- * A sequence number tracker to keep events from clients from being re-applied to the cache if
- * they've already been seen.
- *
- * @since GemFire 5.5
- */
- static class EventSeqnoHolder implements DataSerializable {
- private static final long serialVersionUID = 8137262960763308046L;
-
- /** event sequence number. These */
- long lastSeqno = -1;
-
- /** millisecond timestamp */
- transient long endOfLifeTimer;
-
- /** whether this entry is being removed */
- transient boolean removed;
-
- /**
- * version tag, if any, for the operation
- */
- VersionTag versionTag;
-
- // for debugging
- // transient Exception context;
-
- EventSeqnoHolder(long id, VersionTag versionTag) {
- this.lastSeqno = id;
- this.versionTag = versionTag;
- }
-
- public EventSeqnoHolder() {}
-
- @Override
- public String toString() {
- StringBuilder result = new StringBuilder();
- result.append("seqNo").append(this.lastSeqno);
- if (this.versionTag != null) {
- result.append(",").append(this.versionTag);
- }
- return result.toString();
- }
-
- public void fromData(DataInput in) throws IOException, ClassNotFoundException {
- lastSeqno = in.readLong();
- versionTag = (VersionTag) DataSerializer.readObject(in);
- }
-
- public void toData(DataOutput out) throws IOException {
- out.writeLong(lastSeqno);
- DataSerializer.writeObject(versionTag, out);
- }
- }
-
- /**
- * A holder for the version tags generated for a bulk operation (putAll or removeAll). These
- * version tags are retrieved when a bulk op is retried.
- *
- * @since GemFire 7.0 protected for test purposes only.
- */
- protected static class BulkOpHolder {
- /**
- * Whether this object was removed by the cleanup thread.
- */
- public boolean removed;
-
- /**
- * public for tests only
- */
- public Map<EventID, VersionTag> entryVersionTags = new HashMap<EventID, VersionTag>();
-
- /** millisecond timestamp */
- transient long endOfLifeTimer;
-
- /**
- * creates a new instance to save status of a putAllOperation
- */
- BulkOpHolder() {
- // do nothing
- }
-
- public void putVersionTag(EventID eventId, VersionTag versionTag) {
- entryVersionTags.put(eventId, versionTag);
- this.endOfLifeTimer = 0;
- }
-
-
- @Override
- public String toString() {
- return "BulkOpHolder tags=" + this.entryVersionTags;
- }
- }
-
- public static class ExpiryTask extends SystemTimerTask {
-
- InternalCache cache;
- long expiryTime;
- List trackers = new LinkedList();
-
- public ExpiryTask(InternalCache cache, long expiryTime) {
- this.cache = cache;
- this.expiryTime = expiryTime;
- }
-
- void addTracker(EventTracker tracker) {
- synchronized (trackers) {
- trackers.add(tracker);
- }
- }
-
- void removeTracker(EventTracker tracker) {
- synchronized (trackers) {
- trackers.remove(tracker);
- }
- }
-
- int getNumberOfTrackers() {
- return trackers.size();
- }
-
- @Override
- public void run2() {
- long now = System.currentTimeMillis();
- long timeout = now - expiryTime;
- final boolean traceEnabled = logger.isTraceEnabled();
- synchronized (trackers) {
- for (Iterator it = trackers.iterator(); it.hasNext();) {
- EventTracker tracker = (EventTracker) it.next();
- if (traceEnabled) {
- logger.trace("{} sweeper: starting", tracker.name);
- }
- for (Iterator it2 = tracker.recordedEvents.entrySet().iterator(); it2.hasNext();) {
- Map.Entry e = (Map.Entry) it2.next();
- EventSeqnoHolder evh = (EventSeqnoHolder) e.getValue();
- synchronized (evh) {
- if (evh.endOfLifeTimer == 0) {
- evh.endOfLifeTimer = now; // a new holder - start the timer
- }
- if (evh.endOfLifeTimer <= timeout) {
- evh.removed = true;
- evh.lastSeqno = -1;
- if (traceEnabled) {
- logger.trace("{} sweeper: removing {}", tracker.name, e.getKey());
- }
- it2.remove();
- }
- }
- }
-
- // Remove bulk operations we're tracking
- for (Iterator<Map.Entry<ThreadIdentifier, BulkOpHolder>> it2 =
- tracker.recordedBulkOpVersionTags.entrySet().iterator(); it2.hasNext();) {
- Map.Entry<ThreadIdentifier, BulkOpHolder> e = it2.next();
- BulkOpHolder evh = e.getValue();
- synchronized (evh) {
- if (evh.endOfLifeTimer == 0) {
- evh.endOfLifeTimer = now; // a new holder - start the timer
- }
- // Remove the PutAll tracker only if the put all is complete
- // and it has expired.
- if (evh.endOfLifeTimer <= timeout) {
- evh.removed = true;
- if (logger.isTraceEnabled()) {
- logger.trace("{} sweeper: removing bulkOp {}", tracker.name, e.getKey());
- }
- it2.remove();
- }
- }
- }
- if (traceEnabled) {
- logger.trace("{} sweeper: done", tracker.name);
- }
- }
- }
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/FindVersionTagOperation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/FindVersionTagOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/FindVersionTagOperation.java
index 199aafc..6434667 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/FindVersionTagOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/FindVersionTagOperation.java
@@ -132,7 +132,7 @@ public class FindVersionTagOperation {
result = r.findVersionTagForClientBulkOp(eventId);
} else {
- result = r.findVersionTagForClientEvent(eventId);
+ result = r.findVersionTagForEvent(eventId);
}
if (result != null) {
result.replaceNullIDs(r.getVersionMember());
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
index 2dda38c..de5fd88 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
@@ -77,6 +77,9 @@ import javax.transaction.TransactionManager;
import com.sun.jna.Native;
import com.sun.jna.Platform;
import org.apache.commons.lang.StringUtils;
+
+import org.apache.geode.internal.cache.event.EventTracker;
+import org.apache.geode.internal.cache.event.EventTrackerExpiryTask;
import org.apache.geode.internal.security.SecurityServiceFactory;
import org.apache.logging.log4j.Logger;
@@ -474,7 +477,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
/**
* a system timer task for cleaning up old bridge thread event entries
*/
- private final EventTracker.ExpiryTask recordedEventSweeper;
+ private final EventTrackerExpiryTask recordedEventSweeper;
private final TombstoneService tombstoneService;
@@ -898,7 +901,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
getOffHeapEvictor());
}
- this.recordedEventSweeper = EventTracker.startTrackerServices(this);
+ this.recordedEventSweeper = createEventTrackerExpiryTask();
this.tombstoneService = TombstoneService.initialize(this);
TypeRegistry.init();
@@ -941,6 +944,18 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
} // synchronized
}
+ /**
+ * Initialize the EventTracker's timer task. This is stored for tracking and shutdown purposes
+ */
+ private EventTrackerExpiryTask createEventTrackerExpiryTask() {
+ long lifetimeInMillis =
+ Long.getLong(DistributionConfig.GEMFIRE_PREFIX + "messageTrackingTimeout",
+ PoolFactory.DEFAULT_SUBSCRIPTION_MESSAGE_TRACKING_TIMEOUT / 3);
+ EventTrackerExpiryTask task = new EventTrackerExpiryTask(lifetimeInMillis);
+ getCCPTimer().scheduleAtFixedRate(task, lifetimeInMillis, lifetimeInMillis);
+ return task;
+ }
+
@Override
public SecurityService getSecurityService() {
return this.securityService;
@@ -2347,8 +2362,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
this.cachePerfStats.close();
TXLockService.destroyServices();
-
- EventTracker.stopTrackerServices(this);
+ getEventTrackerTask().cancel();
synchronized (this.ccpTimerMutex) {
if (this.ccpTimer != null) {
@@ -2744,7 +2758,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
* @return the sweeper task
*/
@Override
- public EventTracker.ExpiryTask getEventTrackerTask() {
+ public EventTrackerExpiryTask getEventTrackerTask() {
return this.recordedEventSweeper;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/HARegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/HARegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/HARegion.java
index 4cf8f41..baddff8 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/HARegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/HARegion.java
@@ -39,6 +39,8 @@ import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
+import org.apache.geode.internal.cache.event.EventTracker;
+import org.apache.geode.internal.cache.event.NonDistributedEventTracker;
import org.apache.geode.internal.cache.ha.HARegionQueue;
import org.apache.geode.internal.cache.ha.ThreadIdentifier;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
@@ -112,8 +114,9 @@ public class HARegion extends DistributedRegion {
boolean ifOld, Object expectedOldValue, boolean requireOldValue) {}
@Override
- public void createEventTracker() {
+ public EventTracker createEventTracker() {
// event trackers aren't needed for HARegions
+ return NonDistributedEventTracker.getInstance();
}
/**
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
index 4c229db..aed439c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
@@ -53,6 +53,7 @@ import org.apache.geode.distributed.internal.membership.InternalDistributedMembe
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.control.ResourceAdvisor;
+import org.apache.geode.internal.cache.event.EventTrackerExpiryTask;
import org.apache.geode.internal.cache.extension.Extensible;
import org.apache.geode.internal.cache.persistence.BackupManager;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
@@ -237,7 +238,7 @@ public interface InternalCache extends Cache, Extensible<Cache>, CacheTime {
TXEntryStateFactory getTXEntryStateFactory();
- EventTracker.ExpiryTask getEventTrackerTask();
+ EventTrackerExpiryTask getEventTrackerTask();
void removeDiskStore(DiskStoreImpl diskStore);
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
index 3b3047f..6bee770 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
@@ -131,6 +131,8 @@ import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceT
import org.apache.geode.internal.cache.control.MemoryEvent;
import org.apache.geode.internal.cache.control.MemoryThresholds;
import org.apache.geode.internal.cache.control.ResourceListener;
+import org.apache.geode.internal.cache.event.EventTracker;
+import org.apache.geode.internal.cache.event.NonDistributedEventTracker;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionExecutor;
import org.apache.geode.internal.cache.execute.DistributedRegionFunctionResultSender;
import org.apache.geode.internal.cache.execute.LocalResultCollector;
@@ -326,9 +328,6 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
*/
private final boolean supportsTX;
- /** tracks threadID->seqno information for this region */
- EventTracker eventTracker;
-
/**
* tracks region-level version information for members
*/
@@ -440,6 +439,8 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
private final ImageState imageState;
+ private final EventTracker eventTracker;
+
/**
* Register interest count to track if any register interest is in progress for this region. This
* count will be incremented when register interest starts and decremented when register interest
@@ -646,13 +647,16 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
getDataPolicy().withReplication() || getDataPolicy().isPreloaded(),
getAttributes().getDataPolicy().withPersistence(), this.stopper);
- createEventTracker();
-
// prevent internal regions from participating in a TX, bug 38709
this.supportsTX = !isSecret() && !isUsedForPartitionedRegionAdmin() && !isUsedForMetaRegion()
|| isMetaRegionWithTransactions();
this.testCallable = internalRegionArgs.getTestCallable();
+ eventTracker = createEventTracker();
+ }
+
+ protected EventTracker createEventTracker() {
+ return NonDistributedEventTracker.getInstance();
}
private RegionMap createRegionMap(InternalRegionArguments internalRegionArgs) {
@@ -676,27 +680,11 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
}
/**
- * initialize the event tracker. Not all region implementations want or need one of these. Regions
- * that require one should reimplement this method and create one like so:
- *
- * <pre>
- * {@code
- * this.eventTracker = new EventTracker(this.cache);
- * this.eventTracker.start();
- * }
- * </pre>
+ * Other region classes may track events using different mechanisms than EventTrackers or may not
+ * track events at all
*/
- void createEventTracker() {
- // if LocalRegion is changed to have an event tracker, then the initialize()
- // method should be changed to set it to "initialized" state when the
- // region finishes initialization
- }
-
- /**
- * Other region classes may track events using different mechanisms than EventTrackers
- */
- EventTracker getEventTracker() {
- return this.eventTracker;
+ public EventTracker getEventTracker() {
+ return eventTracker;
}
/** returns the regions version-vector */
@@ -2559,9 +2547,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
}
}
- if (this.eventTracker != null) {
- this.eventTracker.stop();
- }
+ getEventTracker().stop();
if (logger.isTraceEnabled(LogMarker.RVV) && getVersionVector() != null) {
logger.trace(LogMarker.RVV, "version vector for {} is {}", getName(),
getVersionVector().fullToString());
@@ -2666,9 +2652,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
}
this.cache.setRegionByPath(getFullPath(), null);
- if (this.eventTracker != null) {
- this.eventTracker.stop();
- }
+ getEventTracker().stop();
if (this.diskRegion != null) {
this.diskRegion.prepareForClose(this);
@@ -5923,11 +5907,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* is installed in the receiver of the image.
*/
public Map<? extends DataSerializable, ? extends DataSerializable> getEventState() {
- if (this.eventTracker != null) {
- return this.eventTracker.getState();
- } else {
- return null;
- }
+ return getEventTracker().getState();
}
/**
@@ -5939,9 +5919,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* @param state a Map obtained from getEventState()
*/
protected void recordEventState(InternalDistributedMember provider, Map state) {
- if (this.eventTracker != null) {
- this.eventTracker.recordState(provider, state);
- }
+ getEventTracker().recordState(provider, state);
}
/**
@@ -5965,9 +5943,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* record the event's sequenceId in Region's event state to prevent replay.
*/
public void recordEvent(InternalCacheEvent event) {
- if (this.eventTracker != null) {
- this.eventTracker.recordEvent(event);
- }
+ getEventTracker().recordEvent(event);
}
/**
@@ -5976,64 +5952,16 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* @return true if the Region's event state has seen the event
*/
public boolean hasSeenEvent(EntryEventImpl event) {
- boolean isDuplicate = false;
-
- if (this.eventTracker != null) {
- // bug 41289 - wait for event tracker to be initialized before checkin
- // so that an operation intended for a previous version of a bucket
- // is not prematurely applied to a new version of the bucket
- if (this.isUsedForPartitionedRegionBucket()) {
- try {
- this.eventTracker.waitOnInitialization();
- } catch (InterruptedException ie) {
- this.stopper.checkCancelInProgress(ie);
- Thread.currentThread().interrupt();
- }
- }
-
- isDuplicate = this.eventTracker.hasSeenEvent(event);
- if (isDuplicate) {
- event.setPossibleDuplicate(true);
- if (getConcurrencyChecksEnabled() && event.getVersionTag() == null) {
- if (event.isBulkOpInProgress()) {
- event.setVersionTag(findVersionTagForClientBulkOp(event.getEventId()));
- } else {
- event.setVersionTag(findVersionTagForClientEvent(event.getEventId()));
- }
- }
- } else {
- // bug #48205 - a retried PR operation may already have a version assigned to it
- // in another VM
- if (event.isPossibleDuplicate() && event.getRegion().concurrencyChecksEnabled
- && event.getVersionTag() == null && event.getEventId() != null) {
- boolean isBulkOp = event.getOperation().isPutAll() || event.getOperation().isRemoveAll();
- VersionTag tag = FindVersionTagOperation.findVersionTag(event.getRegion(),
- event.getEventId(), isBulkOp);
- event.setVersionTag(tag);
- }
- }
- }
-
- return isDuplicate;
+ return getEventTracker().hasSeenEvent(event);
}
/**
- * tries to find the version tag for a replayed client event
+ * tries to find the version tag for a event
*
* @return the version tag, if known. Null if not
*/
- public VersionTag findVersionTagForClientEvent(EventID eventId) {
- if (this.eventTracker != null) {
- return this.eventTracker.findVersionTag(eventId);
- }
- return null;
- }
-
- public VersionTag findVersionTagForGatewayEvent(EventID eventId) {
- if (this.eventTracker != null) {
- return this.eventTracker.findVersionTagForGateway(eventId);
- }
- return null;
+ public VersionTag findVersionTagForEvent(EventID eventId) {
+ return getEventTracker().findVersionTagForSequence(eventId);
}
/**
@@ -6042,13 +5970,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* @return the version tag, if known. Null if not
*/
public VersionTag findVersionTagForClientBulkOp(EventID eventId) {
- if (eventId == null) {
- return null;
- }
- if (this.eventTracker != null) {
- return this.eventTracker.findVersionTagForBulkOp(eventId);
- }
- return null;
+ return getEventTracker().findVersionTagForBulkOp(eventId);
}
/**
@@ -6061,25 +5983,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* @return true if the Region's event state has seen the event
*/
public boolean hasSeenEvent(EventID eventID) {
- if (eventID == null) {
- return false;
- }
- boolean isDuplicate = false;
- if (this.eventTracker != null) {
- // bug 41289 - wait for event tracker to be initialized before checkin
- // so that an operation intended for a previous version of a bucket
- // is not prematurely applied to a new version of the bucket
- if (this.isUsedForPartitionedRegionBucket()) {
- try {
- this.eventTracker.waitOnInitialization();
- } catch (InterruptedException ie) {
- this.stopper.checkCancelInProgress(ie);
- Thread.currentThread().interrupt();
- }
- }
- isDuplicate = this.eventTracker.hasSeenEvent(eventID, null);
- }
- return isDuplicate;
+ return getEventTracker().hasSeenEvent(eventID);
}
/**
@@ -6092,16 +5996,12 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* @since GemFire 5.7
*/
void syncBulkOp(Runnable task, EventID eventId) {
- if (this.eventTracker != null && !isTX()) {
- this.eventTracker.syncBulkOp(task, eventId);
- } else {
- task.run();
- }
+ getEventTracker().syncBulkOp(task, eventId, isTX());
}
public void recordBulkOpStart(ThreadIdentifier membershipID, EventID eventID) {
- if (this.eventTracker != null && !isTX()) {
- this.eventTracker.recordBulkOpStart(membershipID, eventID);
+ if (!isTX()) {
+ getEventTracker().recordBulkOpStart(eventID, membershipID);
}
}
@@ -7058,14 +6958,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
* @return true if event state has been transfered to this region from another cache
*/
boolean isEventTrackerInitialized() {
- return this.eventTracker != null && this.eventTracker.isInitialized();
- }
-
- /**
- * @return true if this region has an event tracker
- */
- boolean hasEventTracker() {
- return this.eventTracker != null;
+ return getEventTracker().isInitialized();
}
public void acquireDestroyLock() {
@@ -7115,9 +7008,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
this.destroyedSubregionSerialNumbers = collectSubregionSerialNumbers();
try {
- if (this.eventTracker != null) {
- this.eventTracker.stop();
- }
+ getEventTracker().stop();
if (this.diskRegion != null) {
// This was needed to fix bug 30937
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
index c3aec13..27b442d 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
@@ -5448,16 +5448,11 @@ public class PartitionedRegion extends LocalRegion
}
@Override
- void createEventTracker() {
- // PR buckets maintain their own trackers. None is needed at this level
- }
-
- @Override
- public VersionTag findVersionTagForClientEvent(EventID eventId) {
+ public VersionTag findVersionTagForEvent(EventID eventId) {
if (this.dataStore != null) {
Set<Map.Entry<Integer, BucketRegion>> bucketMap = this.dataStore.getAllLocalBuckets();
for (Map.Entry<Integer, BucketRegion> entry : bucketMap) {
- VersionTag result = entry.getValue().findVersionTagForClientEvent(eventId);
+ VersionTag result = entry.getValue().findVersionTagForEvent(eventId);
if (result != null) {
return result;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/event/BulkOperationHolder.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/event/BulkOperationHolder.java b/geode-core/src/main/java/org/apache/geode/internal/cache/event/BulkOperationHolder.java
new file mode 100644
index 0000000..53c4bb5
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/event/BulkOperationHolder.java
@@ -0,0 +1,79 @@
+/*
+ * 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.event;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.versions.VersionTag;
+
+/**
+ * A holder for the version tags generated for a bulk operation (putAll or removeAll). These version
+ * tags are retrieved when a bulk op is retried.
+ *
+ * @since GemFire 7.0 protected for test purposes only.
+ */
+public class BulkOperationHolder {
+ /**
+ * Whether this object was removed by the cleanup thread.
+ */
+ private boolean removed;
+
+ /**
+ * public for tests only
+ */
+ private Map<EventID, VersionTag> entryVersionTags = new HashMap<>();
+
+ /** millisecond timestamp */
+ private transient long endOfLifeTimestamp;
+
+ /**
+ * creates a new instance to save status of a putAllOperation
+ */
+ BulkOperationHolder() {
+ // do nothing
+ }
+
+ void putVersionTag(EventID eventId, VersionTag versionTag) {
+ entryVersionTags.put(eventId, versionTag);
+ this.endOfLifeTimestamp = 0;
+ }
+
+ public Map<EventID, VersionTag> getEntryVersionTags() {
+ return entryVersionTags;
+ }
+
+ @Override
+ public String toString() {
+ return "BulkOperationHolder tags=" + this.entryVersionTags;
+ }
+
+ public synchronized boolean expire(long now, long expirationTime) {
+ if (endOfLifeTimestamp == 0) {
+ endOfLifeTimestamp = now; // a new holder - start the timer
+ }
+ boolean expired = false;
+ if (endOfLifeTimestamp <= expirationTime) {
+ removed = true;
+ expired = true;
+ }
+ return expired;
+ }
+
+ public boolean isRemoved() {
+ return removed;
+ }
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/3f9be9a7/geode-core/src/main/java/org/apache/geode/internal/cache/event/DistributedEventTracker.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/event/DistributedEventTracker.java b/geode-core/src/main/java/org/apache/geode/internal/cache/event/DistributedEventTracker.java
new file mode 100644
index 0000000..ee0e8ff
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/event/DistributedEventTracker.java
@@ -0,0 +1,523 @@
+/*
+ * 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.event;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.CancelCriterion;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.Assert;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.cache.ha.ThreadIdentifier;
+import org.apache.geode.internal.cache.versions.RegionVersionVector;
+import org.apache.geode.internal.cache.versions.VersionTag;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.internal.logging.log4j.LogMarker;
+import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
+
+
+public class DistributedEventTracker implements EventTracker {
+ private static final Logger logger = LogService.getLogger();
+
+ /**
+ * a mapping of originator to the last event applied to this cache
+ *
+ * Keys are instances of {@link ThreadIdentifier}, values are instances of
+ * {@link EventSequenceNumberHolder}.
+ */
+ private final ConcurrentMap<ThreadIdentifier, EventSequenceNumberHolder> recordedEvents =
+ new ConcurrentHashMap<>(100);
+
+ /**
+ * a mapping of originator to bulkOps
+ *
+ * Keys are instances of @link {@link ThreadIdentifier}
+ */
+ private final ConcurrentMap<ThreadIdentifier, Object> recordedBulkOps =
+ new ConcurrentHashMap<>(100);
+
+ /**
+ * a mapping of originator to bulkOperation's last version tags. This map differs from
+ * {@link #recordedBulkOps} in that the thread identifier used here is the base member id and
+ * thread id of the bulk op, as opposed to the fake thread id which is assigned for each bucket.
+ *
+ * recordedBulkOps are also only tracked on the secondary for partitioned regions
+ * recordedBulkOpVersionTags are tracked on both the primary and secondary.
+ *
+ * Keys are instances of @link {@link ThreadIdentifier}, values are instances of
+ * {@link BulkOperationHolder}.
+ */
+ private final ConcurrentMap<ThreadIdentifier, BulkOperationHolder> recordedBulkOpVersionTags =
+ new ConcurrentHashMap<>(100);
+
+ /**
+ * The member that the region corresponding to this tracker (if any) received its initial image
+ * from (if a replicate)
+ */
+ private volatile InternalDistributedMember initialImageProvider;
+
+ /**
+ * The cache associated with this tracker
+ */
+ private InternalCache cache;
+
+ /**
+ * The name of this tracker
+ */
+ private String name;
+
+ /**
+ * whether or not this tracker has been initialized to allow entry operation. replicate region
+ * does not initiate event tracker from its replicates.
+ */
+ private volatile boolean initialized;
+
+ /**
+ * object used to wait for initialization
+ */
+ private final StoppableCountDownLatch initializationLatch;
+
+ /**
+ * Create an event tracker
+ *
+ * @param cache the cache of the region to associate with this tracker
+ * @param stopper the CancelCriterion for the region
+ * @param regionName name of the region
+ */
+ public DistributedEventTracker(InternalCache cache, CancelCriterion stopper, String regionName) {
+
+ this.cache = cache;
+ this.name = "Event Tracker for " + regionName;
+ this.initializationLatch = new StoppableCountDownLatch(stopper, 1);
+ }
+
+ @Override
+ public void start() {
+ if (cache.getEventTrackerTask() != null) {
+ cache.getEventTrackerTask().addTracker(this);
+ }
+ }
+
+ @Override
+ public void stop() {
+ if (cache.getEventTrackerTask() != null) {
+ cache.getEventTrackerTask().removeTracker(this);
+ }
+ }
+
+ @Override
+ public Map<ThreadIdentifier, EventSequenceNumberHolder> getState() {
+ Map<ThreadIdentifier, EventSequenceNumberHolder> result = new HashMap<>(recordedEvents.size());
+ for (Map.Entry<ThreadIdentifier, EventSequenceNumberHolder> entry : recordedEvents.entrySet()) {
+ EventSequenceNumberHolder holder = entry.getValue();
+ // don't transfer version tags - adds too much bulk just so we can do client tag recovery
+ result.put(entry.getKey(),
+ new EventSequenceNumberHolder(holder.getLastSequenceNumber(), null));
+ }
+ return result;
+ }
+
+ @Override
+ public void recordState(InternalDistributedMember provider,
+ Map<ThreadIdentifier, EventSequenceNumberHolder> state) {
+ this.initialImageProvider = provider;
+ StringBuffer sb = null;
+ if (logger.isDebugEnabled()) {
+ sb = new StringBuffer(200);
+ sb.append("Recording initial state for ").append(this.name).append(": ");
+ }
+ for (Map.Entry<ThreadIdentifier, EventSequenceNumberHolder> entry : state.entrySet()) {
+ if (sb != null) {
+ sb.append("\n ").append(entry.getKey().expensiveToString()).append("; sequenceID=")
+ .append(entry.getValue());
+ }
+ // record only if we haven't received an event that is newer
+ recordSequenceNumber(entry.getKey(), entry.getValue(), true);
+ }
+ if (sb != null) {
+ logger.debug(sb);
+ }
+ // fix for bug 41622 - hang in GII. This keeps ops from waiting for the
+ // full GII to complete
+ setInitialized();
+ }
+
+ @Override
+ public void setInitialized() {
+ initializationLatch.countDown();
+ initialized = true;
+ }
+
+ @Override
+ public void waitOnInitialization() throws InterruptedException {
+ initializationLatch.await();
+ }
+
+ /**
+ * Record an event sequence id if it is higher than what we currently have. This is intended for
+ * use during initial image transfer.
+ *
+ * @param membershipID the key of an entry in the map obtained from getEventState()
+ * @param evhObj the value of an entry in the map obtained from getEventState()
+ */
+ protected void recordSequenceNumber(ThreadIdentifier membershipID,
+ EventSequenceNumberHolder evhObj) {
+ recordSequenceNumber(membershipID, evhObj, false);
+ }
+
+ /**
+ * Record an event sequence id if it is higher than what we currently have. This is intended for
+ * use during initial image transfer.
+ *
+ * @param threadID the key of an entry in the map obtained from getEventState()
+ * @param evh the value of an entry in the map obtained from getEventState()
+ * @param ifAbsent only record this state if there's not already an entry for this memberID
+ */
+ private void recordSequenceNumber(ThreadIdentifier threadID, EventSequenceNumberHolder evh,
+ boolean ifAbsent) {
+ boolean removed;
+ if (logger.isDebugEnabled()) {
+ logger.debug("recording {} {}", threadID.expensiveToString(), evh.toString());
+ }
+ do {
+ removed = false;
+ EventSequenceNumberHolder oldEvh = recordedEvents.putIfAbsent(threadID, evh);
+ if (oldEvh != null) {
+ synchronized (oldEvh) {
+ if (oldEvh.isRemoved()) {
+ // need to wait for an entry being removed by the sweeper to go away
+ removed = true;
+ continue;
+ } else {
+ if (ifAbsent) {
+ break;
+ }
+ oldEvh.setEndOfLifeTimestamp(0);
+ if (oldEvh.getLastSequenceNumber() < evh.getLastSequenceNumber()) {
+ oldEvh.setLastSequenceNumber(evh.getLastSequenceNumber());
+ oldEvh.setVersionTag(evh.getVersionTag());
+ }
+ }
+ }
+ } else {
+ evh.setEndOfLifeTimestamp(0);
+ }
+ } while (removed);
+ }
+
+ @Override
+ public void recordEvent(InternalCacheEvent event) {
+ EventID eventID = event.getEventId();
+ if (ignoreEvent(event, eventID)) {
+ return; // not tracked
+ }
+
+ LocalRegion lr = (LocalRegion) event.getRegion();
+ ThreadIdentifier membershipID = createThreadIDFromEvent(eventID);
+
+ VersionTag tag = null;
+ if (lr.getServerProxy() == null) {
+ tag = event.getVersionTag();
+ RegionVersionVector v = ((LocalRegion) event.getRegion()).getVersionVector();
+ canonicalizeIDs(tag, v);
+ }
+
+ EventSequenceNumberHolder newEvh = new EventSequenceNumberHolder(eventID.getSequenceID(), tag);
+ if (logger.isTraceEnabled()) {
+ logger.trace("region event tracker recording {}", event);
+ }
+ recordSequenceNumber(membershipID, newEvh);
+
+ // If this is a bulkOp, and concurrency checks are enabled, we need to
+ // save the version tag in case we retry.
+ // Make recordBulkOp version tag after recordSequenceNumber, so that recordBulkOpStart
+ // in a retry bulk op would not incorrectly remove the saved version tag in
+ // recordedBulkOpVersionTags
+ if (lr.getConcurrencyChecksEnabled()
+ && (event.getOperation().isPutAll() || event.getOperation().isRemoveAll())
+ && lr.getServerProxy() == null) {
+ recordBulkOpEvent(event, membershipID);
+ }
+ }
+
+ private ThreadIdentifier createThreadIDFromEvent(EventID eventID) {
+ return new ThreadIdentifier(eventID.getMembershipID(), eventID.getThreadID());
+ }
+
+ /**
+ * Record a version tag for a bulk operation.
+ */
+ private void recordBulkOpEvent(InternalCacheEvent event, ThreadIdentifier threadID) {
+ EventID eventID = event.getEventId();
+
+ VersionTag tag = event.getVersionTag();
+ if (tag == null) {
+ return;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("recording bulkOp event {} {} {} op={}", threadID.expensiveToString(), eventID,
+ tag, event.getOperation());
+ }
+
+ RegionVersionVector versionVector = ((LocalRegion) event.getRegion()).getVersionVector();
+ canonicalizeIDs(tag, versionVector);
+
+ // Loop until we can successfully update the recorded bulk operations
+ // For this thread id.
+ boolean retry = false;
+ do {
+ BulkOperationHolder bulkOpTracker = recordedBulkOpVersionTags.get(threadID);
+ if (bulkOpTracker == null) {
+ bulkOpTracker = new BulkOperationHolder();
+ BulkOperationHolder old = recordedBulkOpVersionTags.putIfAbsent(threadID, bulkOpTracker);
+ if (old != null) {
+ retry = true;
+ continue;
+ }
+ }
+ synchronized (bulkOpTracker) {
+ if (bulkOpTracker.isRemoved()) {
+ retry = true;
+ continue;
+ }
+
+ // Add the version tag for bulkOp event.
+ bulkOpTracker.putVersionTag(eventID, event.getVersionTag());
+ retry = false;
+ }
+ } while (retry);
+ }
+
+ private void canonicalizeIDs(VersionTag tag, RegionVersionVector versionVector) {
+ if (tag != null && versionVector != null) {
+ tag.setMemberID(versionVector.getCanonicalId(tag.getMemberID()));
+ if (tag.getPreviousMemberID() != null) {
+ tag.setPreviousMemberID(versionVector.getCanonicalId(tag.getPreviousMemberID()));
+ }
+ }
+ }
+
+ @Override
+ public boolean hasSeenEvent(InternalCacheEvent event) {
+ EventID eventID = event.getEventId();
+ if (ignoreEvent(event, eventID)) {
+ return false; // not tracked
+ }
+ return hasSeenEvent(eventID, event);
+ }
+
+ @Override
+ public boolean hasSeenEvent(EventID eventID) {
+ return hasSeenEvent(eventID, null);
+ }
+
+ @Override
+ public boolean hasSeenEvent(EventID eventID, InternalCacheEvent tagHolder) {
+ if (eventID == null) {
+ return false;
+ }
+
+ EventSequenceNumberHolder evh = getSequenceHolderForEvent(eventID);
+ if (evh == null) {
+ return false;
+ }
+
+ synchronized (evh) {
+ if (evh.isRemoved() || evh.getLastSequenceNumber() < eventID.getSequenceID()) {
+ return false;
+ }
+ // log at fine because partitioned regions can send event multiple times
+ // during normal operation during bucket region initialization
+ if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_BRIDGE_SERVER)) {
+ logger.trace(LogMarker.DISTRIBUTION_BRIDGE_SERVER,
+ "Cache encountered replay of event with ID {}. Highest recorded for this source is {}",
+ eventID, evh.getLastSequenceNumber());
+ }
+ // bug #44956 - recover version tag for duplicate event
+ if (evh.getLastSequenceNumber() == eventID.getSequenceID() && tagHolder != null
+ && evh.getVersionTag() != null) {
+ ((EntryEventImpl) tagHolder).setVersionTag(evh.getVersionTag());
+ }
+ return true;
+ }
+ }
+
+ private EventSequenceNumberHolder getSequenceHolderForEvent(EventID eventID) {
+ ThreadIdentifier membershipID = createThreadIDFromEvent(eventID);
+ return recordedEvents.get(membershipID);
+ }
+
+ @Override
+ public VersionTag findVersionTagForSequence(EventID eventID) {
+ EventSequenceNumberHolder evh = getSequenceHolderForEvent(eventID);
+ if (evh == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("search for version tag failed as no event is recorded for {}",
+ createThreadIDFromEvent(eventID).expensiveToString());
+ }
+ return null;
+ }
+
+ synchronized (evh) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("search for version tag located last event for {}: {}",
+ createThreadIDFromEvent(eventID).expensiveToString(), evh);
+ }
+ if (evh.getLastSequenceNumber() != eventID.getSequenceID()) {
+ return null;
+ }
+ // log at fine because partitioned regions can send event multiple times
+ // during normal operation during bucket region initialization
+ if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_BRIDGE_SERVER)
+ && evh.getVersionTag() == null) {
+ logger.trace(LogMarker.DISTRIBUTION_BRIDGE_SERVER,
+ "Could not recover version tag. Found event holder with no version tag for {}",
+ eventID);
+ }
+ return evh.getVersionTag();
+ }
+ }
+
+ @Override
+ public VersionTag findVersionTagForBulkOp(EventID eventID) {
+ if (eventID == null) {
+ return null;
+ }
+ ThreadIdentifier threadID = createThreadIDFromEvent(eventID);
+ BulkOperationHolder evh = recordedBulkOpVersionTags.get(threadID);
+ if (evh == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("search for version tag failed as no events are recorded for {}",
+ threadID.expensiveToString());
+ }
+ return null;
+ }
+
+ synchronized (evh) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("search for version tag located event holder for {}: {}",
+ threadID.expensiveToString(), evh);
+ }
+ return evh.getEntryVersionTags().get(eventID);
+ }
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * @return true if the event should not be tracked, false otherwise
+ */
+ private boolean ignoreEvent(InternalCacheEvent event, EventID eventID) {
+ if (eventID == null) {
+ return true;
+ } else {
+ boolean isVersioned = (event.getVersionTag() != null);
+ boolean isClient = event.hasClientOrigin();
+ if (isVersioned && isClient) {
+ return false; // version tags for client events are kept for retries by the client
+ }
+ boolean isEntry = event.getOperation().isEntry();
+ boolean isPr = event.getRegion().getAttributes().getDataPolicy().withPartitioning()
+ || ((LocalRegion) event.getRegion()).isUsedForPartitionedRegionBucket();
+ return (!isClient && // ignore if it originated on a server, and
+ isEntry && // it affects an entry and
+ !isPr); // is not on a PR
+ }
+ }
+
+ @Override
+ public void syncBulkOp(Runnable r, EventID eventID, boolean partOfTransaction) {
+ if (partOfTransaction) {
+ r.run();
+ return;
+ }
+ Assert.assertTrue(eventID != null);
+ ThreadIdentifier membershipID = createThreadIDFromEvent(eventID);
+ Object opSyncObj = null;
+ do {
+ opSyncObj = recordedBulkOps.putIfAbsent(membershipID, new Object());
+ if (opSyncObj == null) {
+ opSyncObj = recordedBulkOps.get(membershipID);
+ }
+ } while (opSyncObj == null);
+
+ synchronized (opSyncObj) {
+ try {
+ recordBulkOpStart(eventID, membershipID);
+ // Perform the bulk op
+ r.run();
+ } finally {
+ recordedBulkOps.remove(membershipID);
+ }
+ }
+ }
+
+ @Override
+ public void recordBulkOpStart(EventID eventID, ThreadIdentifier tid) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("recording bulkOp start for {}", tid.expensiveToString());
+ }
+ EventSequenceNumberHolder evh = recordedEvents.get(tid);
+ if (evh == null) {
+ return;
+ }
+ synchronized (evh) {
+ // only remove it when a new bulk op occurs
+ if (eventID.getSequenceID() > evh.getLastSequenceNumber()) {
+ this.recordedBulkOpVersionTags.remove(tid);
+ }
+ }
+ }
+
+ @Override
+ public boolean isInitialized() {
+ return this.initialized;
+ }
+
+ @Override
+ public boolean isInitialImageProvider(DistributedMember mbr) {
+ return (this.initialImageProvider != null) && (mbr != null)
+ && this.initialImageProvider.equals(mbr);
+ }
+
+ public ConcurrentMap<ThreadIdentifier, BulkOperationHolder> getRecordedBulkOpVersionTags() {
+ return recordedBulkOpVersionTags;
+ }
+
+ @Override
+ public ConcurrentMap<ThreadIdentifier, EventSequenceNumberHolder> getRecordedEvents() {
+ return recordedEvents;
+ }
+
+ @Override
+ public String toString() {
+ return "" + this.name + "(initialized=" + this.initialized + ")";
+ }
+
+}