You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by zh...@apache.org on 2018/04/10 05:13:01 UTC
[geode] branch feature/GEODE-4942 updated (b5bb4b9 -> bc9a168)
This is an automated email from the ASF dual-hosted git repository.
zhouxj pushed a change to branch feature/GEODE-4942
in repository https://gitbox.apache.org/repos/asf/geode.git.
omit b5bb4b9 GEODE-4942: when sender is starting, and not running yet, put event at primary should be saved for QueueRemovalMessage
add d00dbd7 GEODE-4947: JdbcConnector tests run against external database as AcceptanceTest category (#1747)
add f13f56c GEODE-5003: Add tests for CompositeResultData (#1743)
add ced3d00 GEODE-4830: modify the message when no jndi-binding is found. (#1732)
add 514b1d3 GEODE-5000: do not request/apply cluster config when creating client … (#1739)
add 85c67b2 GEODE-4819: Separating authorization out from protobuf handlers
add 6f564c0 GEODE-5005: Add unit tests for InfoResultData (#1745)
add c0dc8a3 GEODE-5004: Add unit tests for ErrorResultData (#1744)
add a6a7275 GEODE-5021: Add parameters to connection in jdbc-1.0.xsd (#1748)
add a956147 GEODE-1279: Rename FinalStaticArrayShouldNotCauseSegFaultRegressionTest
add 76dd7e2 GEODE_1279: Rename MembershipAttributesAreSerializableRegressionTest
add eab8252 COMMIT THEN REVIEW. fix spelling: PARTITIION -> PARTITION
add 947e37b GEODE-5026: Do not run acceptanceTest in parallel under docker
add 7b3e809 Merge pull request #1751 from jdeppe-pivotal/feature/GEODE-5026
add 804c053 GEODE-4518: Replace DSCODE with an enumeration. (#1738)
add 2d45dad GEODE-4999: Added explicit tomcat 8.5 directory for tcserver support (#1737)
add f6f58d3 Revert "GEODE-4999: Added explicit tomcat 8.5 directory for tcserver support (#1737)"
add c0be3ec GEODE-5006: Remove unused ObjectResultData (#1756)
add a7701e8 GEODE-5007: Add unit tests to TabularResultData (#1757)
add f0fe90e GEODE-5024: Use debug version of gradle-dockerized-test-plugin (#1749)
add 2c44745 GEODE-5029: Improve docs around query timeout and low memory (#1755)
add ebce148 GEODE-5027 Bump version to 1.6.0 and add v1.5.0 for testing
add e47df10 GEODE-3820 CI failure: DataSerializerPropogationDUnitTest.testDataSerializersEventIdVerification
add 2e89e84 GEODE-4995 ClusterStartupRule is inefficient when shutting down the cluster
add dfab3ba GEODE-5027 Bump version to 1.6.0 and add v1.5.0 for testing
new bc9a168 GEODE-4942: when sender is starting, and not running yet, put event at primary should be saved for QueueRemovalMessage
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (b5bb4b9)
\
N -- N -- N refs/heads/feature/GEODE-4942 (bc9a168)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
build.gradle | 4 +-
ci/pipelines/develop.yml | 4 +-
.../source/subnavs/geode-subnav.erb | 5 +-
geode-connectors/build.gradle | 10 +
.../geode/connectors/jdbc/internal/SqlHandler.java | 4 +-
.../jdbc/internal/SqlStatementFactory.java | 14 +-
.../geode.apache.org/schema/jdbc/jdbc-1.0.xsd | 1 +
.../jdbc/JdbcAsyncWriterIntegrationTest.java | 18 +-
...JdbcDUnitTest.java => JdbcDistributedTest.java} | 168 +++---
.../connectors/jdbc/JdbcLoaderIntegrationTest.java | 35 +-
.../connectors/jdbc/JdbcWriterIntegrationTest.java | 18 +-
.../jdbc/MySqlJdbcAsyncWriterIntegrationTest.java | 47 ++
.../connectors/jdbc/MySqlJdbcDistributedTest.java | 100 ++++
.../jdbc/MySqlJdbcLoaderIntegrationTest.java | 58 ++
.../jdbc/MySqlJdbcWriterIntegrationTest.java | 47 ++
.../PostgresJdbcAsyncWriterIntegrationTest.java | 47 ++
.../jdbc/PostgresJdbcDistributedTest.java | 88 +++
.../jdbc/PostgresJdbcLoaderIntegrationTest.java | 57 ++
.../jdbc/PostgresJdbcWriterIntegrationTest.java | 47 ++
.../MySqlTableMetaDataManagerIntegrationTest.java | 42 ++
...ostgresTableMetaDataManagerIntegrationTest.java | 43 ++
.../jdbc/internal/SqlStatementFactoryTest.java | 17 +-
.../TableMetaDataManagerIntegrationTest.java | 28 +-
.../jdbc/internal/TestConfigService.java | 14 +-
.../test/junit/rules/DatabaseConnectionRule.java | 21 +-
.../junit/rules/InMemoryDerbyConnectionRule.java | 40 +-
.../test/junit/rules/MySqlConnectionRule.java | 72 +++
.../test/junit/rules/PostgresConnectionRule.java | 45 ++
.../junit/rules/SqlDatabaseConnectionRule.java | 123 ++++
.../geode/connectors/jdbc/internal/mysql.yml | 28 +
.../geode/connectors/jdbc/internal/postgres.yml | 27 +
.../org/apache/geode/connectors/jdbc/mysql.yml | 28 +
.../org/apache/geode/connectors/jdbc/postgres.yml | 27 +
.../main/java/org/apache/geode/DataSerializer.java | 20 +-
.../asyncqueue/internal/AsyncEventQueueStats.java | 5 +
.../java/org/apache/geode/internal/DSCODE.java | 160 ++---
.../geode/internal/InternalDataSerializer.java | 655 ++++++++++++---------
.../java/org/apache/geode/internal/Version.java | 72 +--
.../internal/cache/CacheDistributionAdvisor.java | 2 +-
.../cache/CachedDeserializableFactory.java | 2 +-
.../internal/cache/ClusterConfigurationLoader.java | 2 +-
.../geode/internal/cache/GemFireCacheImpl.java | 43 +-
.../org/apache/geode/internal/cache/Token.java | 2 +-
.../cache/snapshot/RegionSnapshotServiceImpl.java | 2 +-
.../apache/geode/internal/cache/tier/Acceptor.java | 7 -
.../geode/internal/cache/tier/ConnectionProxy.java | 2 +-
.../cache/tier/sockets/ChunkedMessage.java | 4 +-
.../tier/sockets/ClientDataSerializerMessage.java | 2 +-
.../cache/tier/sockets/ClientTombstoneMessage.java | 2 +-
.../tier/sockets/ClientUpdateMessageImpl.java | 4 +-
.../cache/tier/sockets/CommandInitializer.java | 99 +---
.../cache/tier/sockets/HAEventWrapper.java | 4 +-
.../geode/internal/cache/tier/sockets/Part.java | 2 +-
.../tier/sockets/ServerSideHandshakeFactory.java | 2 +-
.../wan/AbstractGatewaySenderEventProcessor.java | 1 +
.../internal/cache/wan/GatewaySenderStats.java | 16 +
.../internal/offheap/AbstractStoredObject.java | 3 +-
.../apache/geode/internal/offheap/DataType.java | 403 +++++++------
.../internal/offheap/OffHeapRegionEntryHelper.java | 4 +-
.../internal/offheap/OffHeapStoredObject.java | 2 +-
.../org/apache/geode/internal/util/BlobHelper.java | 2 +-
.../cli/commands/ListJndiBindingCommand.java | 19 +-
.../cli/commands/dto/RegionAttributesInfo.java | 162 -----
.../internal/cli/commands/dto/RegionDetails.java | 245 --------
.../cli/commands/dto/RegionMemberDetails.java | 205 -------
.../cli/result/CliJsonSerializableFactory.java | 46 --
.../internal/cli/result/CommandResult.java | 161 +----
.../internal/cli/result/ObjectResultData.java | 86 ---
.../internal/cli/result/ResultBuilder.java | 37 --
.../management/internal/cli/result/ResultData.java | 1 -
.../management/internal/cli/util/JsonUtil.java | 48 --
.../main/java/org/apache/geode/pdx/FieldType.java | 4 +-
.../org/apache/geode/pdx/internal/EnumInfo.java | 2 -
.../apache/geode/pdx/internal/PdxInstanceEnum.java | 2 +-
.../apache/geode/pdx/internal/PdxInstanceImpl.java | 4 +-
.../apache/geode/pdx/internal/PdxReaderImpl.java | 7 +-
.../org/apache/geode/pdx/internal/PdxString.java | 18 +-
.../apache/geode/pdx/internal/PdxWriterImpl.java | 6 +-
...ArrayShouldNotCauseSegFaultRegressionTest.java} | 55 +-
...ipAttributesAreSerializableRegressionTest.java} | 30 +-
.../DSCODETest.java} | 38 +-
.../apache/geode/internal/VersionJUnitTest.java | 22 +
.../geode/internal/cache/GemFireCacheImplTest.java | 26 +-
.../rollingupgrade/RollingUpgrade2DUnitTest.java | 230 +++-----
.../DataSerializerPropagationDUnitTest.java | 16 +-
.../geode/internal/offheap/DataTypeJUnitTest.java | 40 +-
.../internal/offheap/MemoryBlockNodeJUnitTest.java | 2 +-
.../offheap/OffHeapRegionEntryHelperJUnitTest.java | 2 +-
.../offheap/OffHeapStoredObjectJUnitTest.java | 8 +-
.../offheap/TinyStoredObjectJUnitTest.java | 3 +-
.../cli/commands/ListJndiBindingCommandTest.java | 74 +++
.../cli/result/CompositeResultDataTest.java | 132 +++++
.../internal/cli/result/ErrorResultDataTest.java | 75 +++
.../internal/cli/result/InfoResultDataTest.java | 63 ++
.../internal/cli/result/TabularResultDataTest.java | 47 +-
.../apache/geode/pdx/PdxSerializableJUnitTest.java | 20 +-
.../geode/test/dunit/rules/ClusterStartupRule.java | 1 +
.../apache/geode/test/junit/rules/VMProvider.java | 9 +
.../apache/geode/codeAnalysis/excludedClasses.txt | 3 +-
.../codeAnalysis/sanctionedDataSerializables.txt | 2 +-
.../query_additional/advanced_querying.html.md.erb | 12 +-
.../query_additional/query_timeout.html.md.erb | 36 ++
.../monitor_queries_for_low_memory.html.md.erb | 27 +-
geode-docs/reference/topics/cache_xml.html.md.erb | 2 +-
.../reference/topics/client-cache.html.md.erb | 4 +-
geode-old-versions/build.gradle | 1 +
.../test/dunit/standalone/VersionManager.java | 1 +
.../AuthorizingFunctionServiceImplTest.java | 3 -
.../geode/internal/cache/wan/WANTestBase.java | 15 +
.../ParallelGatewaySenderOperationsDUnitTest.java | 38 +-
gradle/docker.gradle | 1 -
111 files changed, 2720 insertions(+), 2222 deletions(-)
rename geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/{JdbcDUnitTest.java => JdbcDistributedTest.java} (85%)
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/MySqlJdbcAsyncWriterIntegrationTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/MySqlJdbcDistributedTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/MySqlJdbcLoaderIntegrationTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/MySqlJdbcWriterIntegrationTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/PostgresJdbcAsyncWriterIntegrationTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/PostgresJdbcDistributedTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/PostgresJdbcLoaderIntegrationTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/PostgresJdbcWriterIntegrationTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/MySqlTableMetaDataManagerIntegrationTest.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/connectors/jdbc/internal/PostgresTableMetaDataManagerIntegrationTest.java
rename geode-core/src/main/java/org/apache/geode/management/internal/cli/result/CliJsonSerializableIds.java => geode-connectors/src/test/java/org/apache/geode/test/junit/rules/DatabaseConnectionRule.java (68%)
rename geode-core/src/main/java/org/apache/geode/management/internal/cli/result/CliJsonSerializable.java => geode-connectors/src/test/java/org/apache/geode/test/junit/rules/InMemoryDerbyConnectionRule.java (52%)
create mode 100644 geode-connectors/src/test/java/org/apache/geode/test/junit/rules/MySqlConnectionRule.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/test/junit/rules/PostgresConnectionRule.java
create mode 100644 geode-connectors/src/test/java/org/apache/geode/test/junit/rules/SqlDatabaseConnectionRule.java
create mode 100644 geode-connectors/src/test/resources/org/apache/geode/connectors/jdbc/internal/mysql.yml
create mode 100644 geode-connectors/src/test/resources/org/apache/geode/connectors/jdbc/internal/postgres.yml
create mode 100644 geode-connectors/src/test/resources/org/apache/geode/connectors/jdbc/mysql.yml
create mode 100644 geode-connectors/src/test/resources/org/apache/geode/connectors/jdbc/postgres.yml
delete mode 100644 geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/dto/RegionAttributesInfo.java
delete mode 100644 geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/dto/RegionDetails.java
delete mode 100644 geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/dto/RegionMemberDetails.java
delete mode 100644 geode-core/src/main/java/org/apache/geode/management/internal/cli/result/CliJsonSerializableFactory.java
delete mode 100644 geode-core/src/main/java/org/apache/geode/management/internal/cli/result/ObjectResultData.java
rename geode-core/src/test/java/org/apache/geode/cache/{FinalStaticArrayShouldNotCauseSegFaultTest.java => FinalStaticArrayShouldNotCauseSegFaultRegressionTest.java} (60%)
rename geode-core/src/test/java/org/apache/geode/cache/{MembershipAttributesAreSerializableTest.java => MembershipAttributesAreSerializableRegressionTest.java} (75%)
copy geode-core/src/test/java/org/apache/geode/{management/internal/cli/result/TabularResultDataTest.java => internal/DSCODETest.java} (55%)
create mode 100644 geode-core/src/test/java/org/apache/geode/management/internal/cli/commands/ListJndiBindingCommandTest.java
create mode 100644 geode-core/src/test/java/org/apache/geode/management/internal/cli/result/CompositeResultDataTest.java
create mode 100644 geode-core/src/test/java/org/apache/geode/management/internal/cli/result/ErrorResultDataTest.java
create mode 100644 geode-core/src/test/java/org/apache/geode/management/internal/cli/result/InfoResultDataTest.java
create mode 100644 geode-docs/developing/query_additional/query_timeout.html.md.erb
--
To stop receiving notification emails like this one, please contact
zhouxj@apache.org.
[geode] 01/01: GEODE-4942: when sender is starting,
and not running yet,
put event at primary should be saved for QueueRemovalMessage
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhouxj pushed a commit to branch feature/GEODE-4942
in repository https://gitbox.apache.org/repos/asf/geode.git
commit bc9a168cdb3890e4b974d990ffbe66e12b09224b
Author: zhouxh <gz...@pivotal.io>
AuthorDate: Wed Apr 4 18:07:54 2018 -0700
GEODE-4942: when sender is starting, and not running yet, put event at primary should be saved for QueueRemovalMessage
---
.../asyncqueue/internal/AsyncEventQueueStats.java | 5 ++
.../geode/internal/cache/EntryEventImpl.java | 3 +
.../internal/cache/PartitionedRegionDataStore.java | 6 +-
.../internal/cache/wan/AbstractGatewaySender.java | 15 ++++-
.../wan/AbstractGatewaySenderEventProcessor.java | 50 +++++++++++++-
.../internal/cache/wan/GatewaySenderStats.java | 16 +++++
.../ConcurrentParallelGatewaySenderQueue.java | 9 +++
.../wan/parallel/ParallelGatewaySenderQueue.java | 20 +++++-
.../geode/internal/cache/wan/WANTestBase.java | 78 +++++++++++++++++++---
.../ParallelGatewaySenderOperationsDUnitTest.java | 38 +++++++++--
.../SerialGatewaySenderOperationsDUnitTest.java | 49 +++++++++++++-
11 files changed, 263 insertions(+), 26 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueStats.java b/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueStats.java
index dee2c92..b8259a3 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueStats.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueStats.java
@@ -47,6 +47,9 @@ public class AsyncEventQueueStats extends GatewaySenderStats {
f.createIntCounter(EVENTS_NOT_QUEUED_CONFLATED,
"Number of events received but not added to the event queue because the queue already contains an event with the event's key.",
"operations"),
+ f.createIntCounter(NOT_QUEUED_EVENTS, "Number of events not added to queue.", "events"),
+ f.createIntCounter(NOT_QUEUED_EVENTS_AT_YET_RUNNING_PRIMARY_SENDER,
+ "Number of events not added to primary queue due to sender yet runing.", "events"),
f.createIntCounter(EVENTS_CONFLATED_FROM_BATCHES,
"Number of events conflated from batches.", "operations"),
f.createIntCounter(EVENTS_DISTRIBUTED,
@@ -122,6 +125,8 @@ public class AsyncEventQueueStats extends GatewaySenderStats {
unprocessedTokenMapSizeId = type.nameToId(UNPROCESSED_TOKEN_MAP_SIZE);
conflationIndexesMapSizeId = type.nameToId(CONFLATION_INDEXES_MAP_SIZE);
notQueuedEventsId = type.nameToId(NOT_QUEUED_EVENTS);
+ notQueuedEventsAtYetRunningPrimarySenderId =
+ type.nameToId(NOT_QUEUED_EVENTS_AT_YET_RUNNING_PRIMARY_SENDER);
eventsFilteredId = type.nameToId(EVENTS_FILTERED);
eventsConflatedFromBatchesId = type.nameToId(EVENTS_CONFLATED_FROM_BATCHES);
loadBalancesCompletedId = type.nameToId(LOAD_BALANCES_COMPLETED);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/EntryEventImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/EntryEventImpl.java
index c91d236..664d054 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/EntryEventImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/EntryEventImpl.java
@@ -2158,6 +2158,9 @@ public class EntryEventImpl implements InternalEntryEvent, InternalCacheEvent,
if (this.getInhibitDistribution()) {
buf.append(";inhibitDistribution");
}
+ if (this.tailKey != -1) {
+ buf.append(";tailKey=" + tailKey);
+ }
buf.append("]");
return buf.toString();
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java
index ef8eb99..d468ef4 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java
@@ -2450,10 +2450,10 @@ public class PartitionedRegionDataStore implements HasCachePerfStats {
return sizeOfLocalPrimaries;
}
- public int getSizeOfLocalBuckets(boolean includeSecondary) {
+ public int getSizeOfLocalBuckets() {
int sizeOfLocal = 0;
- Set<BucketRegion> primaryBuckets = getAllLocalBucketRegions();
- for (BucketRegion br : primaryBuckets) {
+ Set<BucketRegion> allLocalBuckets = getAllLocalBucketRegions();
+ for (BucketRegion br : allLocalBuckets) {
sizeOfLocal += br.size();
}
return sizeOfLocal;
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
index a134e1e..034d810 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySender.java
@@ -849,7 +849,10 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
// If this gateway is not running, return
if (!isRunning()) {
if (isDebugEnabled) {
- logger.debug("Returning back without putting into the gateway sender queue");
+ logger.debug("Returning back without putting into the gateway sender queue:" + event);
+ }
+ if (this.eventProcessor != null) {
+ this.eventProcessor.registerEventDroppedInPrimaryQueue(event);
}
return;
}
@@ -962,7 +965,10 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
// The sender may have stopped, after we have checked the status in the beginning.
if (!isRunning()) {
if (isDebugEnabled) {
- logger.debug("Returning back without putting into the gateway sender queue");
+ logger.debug("Returning back without putting into the gateway sender queue:" + event);
+ }
+ if (this.eventProcessor != null) {
+ this.eventProcessor.registerEventDroppedInPrimaryQueue(event);
}
return;
}
@@ -1251,6 +1257,11 @@ public abstract class AbstractGatewaySender implements GatewaySender, Distributi
return localProcessor == null ? 0 : localProcessor.eventQueueSize();
}
+ public int getEventSecondaryQueueSize() {
+ AbstractGatewaySenderEventProcessor localProcessor = this.eventProcessor;
+ return localProcessor == null ? 0 : localProcessor.eventSecondaryQueueSize();
+ }
+
public void setEnqueuedAllTempQueueEvents(boolean enqueuedAllTempQueueEvents) {
this.enqueuedAllTempQueueEvents = enqueuedAllTempQueueEvents;
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
index 9309e43..eea7480 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
@@ -33,6 +33,7 @@ import org.apache.geode.GemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
@@ -49,6 +50,7 @@ import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.PartitionedRegion;
+import org.apache.geode.internal.cache.PartitionedRegionHelper;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue;
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue;
@@ -261,15 +263,57 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
}
// This should be local size instead of pr size
- if (this.queue instanceof ParallelGatewaySenderQueue) {
- return ((ParallelGatewaySenderQueue) queue).localSize();
- }
if (this.queue instanceof ConcurrentParallelGatewaySenderQueue) {
return ((ConcurrentParallelGatewaySenderQueue) queue).localSize();
}
return this.queue.size();
}
+ public int eventSecondaryQueueSize() {
+ if (queue == null) {
+ return 0;
+ }
+
+ // if parallel, get both primary and secondary queues' size, then substract primary queue's size
+ if (this.queue instanceof ConcurrentParallelGatewaySenderQueue) {
+ int size = ((ConcurrentParallelGatewaySenderQueue) queue).localSize(true)
+ - ((ConcurrentParallelGatewaySenderQueue) queue).localSize(false);
+ return size;
+ }
+ return this.queue.size();
+ }
+
+ public void registerEventDroppedInPrimaryQueue(EntryEventImpl event) {
+ if (queue == null) {
+ return;
+ }
+ if (this.queue instanceof ConcurrentParallelGatewaySenderQueue) {
+ ConcurrentParallelGatewaySenderQueue cpgsq = (ConcurrentParallelGatewaySenderQueue) queue;
+ PartitionedRegion prQ = cpgsq.getRegion(event.getRegion().getFullPath());
+ if (prQ == null) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("shadow partitioned region " + event.getRegion().getFullPath()
+ + " is not created yet.");
+ }
+ return;
+ }
+ int bucketId = PartitionedRegionHelper.getHashKey((EntryOperation) event);
+ long shadowKey = event.getTailKey();
+
+ ParallelGatewaySenderQueue pgsq =
+ (ParallelGatewaySenderQueue) cpgsq.getQueueByBucket(bucketId);
+ boolean isPrimary = prQ.getRegionAdvisor().getBucketAdvisor(bucketId).isPrimary();
+ if (isPrimary) {
+ pgsq.addRemovedEvent(prQ, bucketId, shadowKey);
+ this.sender.getStatistics().incEventsNotQueuedAtYetRunningPrimarySender();
+ if (logger.isDebugEnabled()) {
+ logger.debug("register dropped event for primary queue. BucketId is " + bucketId
+ + ", shadowKey is " + shadowKey + ", prQ is " + prQ.getFullPath());
+ }
+ }
+ }
+ }
+
/**
* @return the sender
*/
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderStats.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderStats.java
index c7fd370..adaf928 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderStats.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderStats.java
@@ -84,6 +84,8 @@ public class GatewaySenderStats {
protected static final String EVENTS_FILTERED = "eventsFiltered";
protected static final String NOT_QUEUED_EVENTS = "notQueuedEvent";
+ protected static final String NOT_QUEUED_EVENTS_AT_YET_RUNNING_PRIMARY_SENDER =
+ "notQueuedEventAtYetRunningPrimarySender";
protected static final String LOAD_BALANCES_COMPLETED = "loadBalancesCompleted";
protected static final String LOAD_BALANCES_IN_PROGRESS = "loadBalancesInProgress";
@@ -135,6 +137,8 @@ public class GatewaySenderStats {
protected static int eventsFilteredId;
/** Id of not queued events */
protected static int notQueuedEventsId;
+ /** Id of not queued events due to the primary sender is yet running */
+ protected static int notQueuedEventsAtYetRunningPrimarySenderId;
/** Id of events conflated in batch */
protected static int eventsConflatedFromBatchesId;
/** Id of load balances completed */
@@ -213,6 +217,8 @@ public class GatewaySenderStats {
f.createIntGauge(CONFLATION_INDEXES_MAP_SIZE,
"Current number of entries in the conflation indexes map.", "events"),
f.createIntCounter(NOT_QUEUED_EVENTS, "Number of events not added to queue.", "events"),
+ f.createIntCounter(NOT_QUEUED_EVENTS_AT_YET_RUNNING_PRIMARY_SENDER,
+ "Number of events not added to primary queue due to sender yet runing.", "events"),
f.createIntCounter(EVENTS_FILTERED,
"Number of events filtered through GatewayEventFilter.", "events"),
f.createIntCounter(LOAD_BALANCES_COMPLETED, "Number of load balances completed",
@@ -249,6 +255,8 @@ public class GatewaySenderStats {
unprocessedTokenMapSizeId = type.nameToId(UNPROCESSED_TOKEN_MAP_SIZE);
conflationIndexesMapSizeId = type.nameToId(CONFLATION_INDEXES_MAP_SIZE);
notQueuedEventsId = type.nameToId(NOT_QUEUED_EVENTS);
+ notQueuedEventsAtYetRunningPrimarySenderId =
+ type.nameToId(NOT_QUEUED_EVENTS_AT_YET_RUNNING_PRIMARY_SENDER);
eventsFilteredId = type.nameToId(EVENTS_FILTERED);
eventsConflatedFromBatchesId = type.nameToId(EVENTS_CONFLATED_FROM_BATCHES);
loadBalancesCompletedId = type.nameToId(LOAD_BALANCES_COMPLETED);
@@ -599,6 +607,14 @@ public class GatewaySenderStats {
return this.stats.getInt(notQueuedEventsId);
}
+ public void incEventsNotQueuedAtYetRunningPrimarySender() {
+ this.stats.incInt(notQueuedEventsAtYetRunningPrimarySenderId, 1);
+ }
+
+ public int getEventsNotQueuedAtYetRunningPrimarySender() {
+ return this.stats.getInt(notQueuedEventsAtYetRunningPrimarySenderId);
+ }
+
public void incEventsFiltered() {
this.stats.incInt(eventsFilteredId, 1);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java
index 4fc940c..e556910 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java
@@ -121,6 +121,11 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
return this.processors[0].getQueue().size();
}
+ public String displayContent() {
+ ParallelGatewaySenderQueue pgsq = (ParallelGatewaySenderQueue) (processors[0].getQueue());
+ return pgsq.displayContent();
+ }
+
public int localSize() {
return localSize(false);
}
@@ -190,6 +195,10 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
return processors[index];
}
+ public RegionQueue getQueueByBucket(int bucketId) {
+ return getPGSProcessor(bucketId).getQueue();
+ }
+
public BlockingQueue<GatewaySenderEventImpl> getBucketTmpQueue(int bucketId) {
return getPGSProcessor(bucketId).getBucketTmpQueue(bucketId);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
index 3aa8534..89880fc 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
@@ -1112,7 +1112,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
// This method may need synchronization in case it is used by
// ConcurrentParallelGatewaySender
- protected void addRemovedEvent(PartitionedRegion prQ, int bucketId, Object key) {
+ public void addRemovedEvent(PartitionedRegion prQ, int bucketId, Object key) {
StoppableReentrantLock lock = buckToDispatchLock;
if (lock != null) {
lock.lock();
@@ -1405,12 +1405,28 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
return localSize(false);
}
+ public String displayContent() {
+ int size = 0;
+ StringBuffer sb = new StringBuffer();
+ for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) {
+ if (prQ != null && prQ.getDataStore() != null) {
+ Set<BucketRegion> allLocalBuckets = prQ.getDataStore().getAllLocalBucketRegions();
+ for (BucketRegion br : allLocalBuckets) {
+ if (br.size() > 0) {
+ sb.append("bucketId=" + br.getId() + ":" + br.keySet() + ";");
+ }
+ }
+ }
+ }
+ return sb.toString();
+ }
+
public int localSize(boolean includeSecondary) {
int size = 0;
for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) {
if (prQ != null && prQ.getDataStore() != null) {
if (includeSecondary) {
- size += prQ.getDataStore().getSizeOfLocalBuckets(true);
+ size += prQ.getDataStore().getSizeOfLocalBuckets();
} else {
size += prQ.getDataStore().getSizeOfLocalPrimaryBuckets();
}
diff --git a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
index 226595b..3799083 100644
--- a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
+++ b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/WANTestBase.java
@@ -933,6 +933,8 @@ public class WANTestBase extends DistributedTestCase {
}
props.setProperty(MCAST_PORT, "0");
props.setProperty(LOCATORS, "localhost[" + locPort + "]");
+ String logLevel = System.getProperty(LOG_LEVEL, "info");
+ props.setProperty(LOG_LEVEL, logLevel);
InternalDistributedSystem ds = test.getSystem(props);
cache = CacheFactory.create(ds);
}
@@ -1155,6 +1157,21 @@ public class WANTestBase extends DistributedTestCase {
return stats;
}
+ public static List<Integer> getSenderStatsForDroppedEvents(String senderId) {
+ AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId);
+ GatewaySenderStats statistics = sender.getStatistics();
+ ArrayList<Integer> stats = new ArrayList<Integer>();
+ int eventNotQueued = statistics.getEventsNotQueuedAtYetRunningPrimarySender();
+ if (eventNotQueued > 0) {
+ logger.info(
+ "Found " + eventNotQueued + " not queued events due to primary sender is yet running");
+ }
+ stats.add(eventNotQueued);
+ stats.add(statistics.getEventsNotQueued());
+ stats.add(statistics.getEventsNotQueuedConflated());
+ return stats;
+ }
+
public static void checkQueueStats(String senderId, final int queueSize, final int eventsReceived,
final int eventsQueued, final int eventsDistributed) {
GatewaySenderStats statistics = getGatewaySenderStats(senderId);
@@ -2746,11 +2763,21 @@ public class WANTestBase extends DistributedTestCase {
public static void validateQueueSizeStat(String id, final int queueSize) {
final AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(id);
- Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ Awaitility.await().atMost(60, TimeUnit.SECONDS)
.until(() -> assertEquals(queueSize, sender.getEventQueueSize()));
assertEquals(queueSize, sender.getEventQueueSize());
}
+ public static void validateSecondaryQueueSizeStat(String id, final int queueSize) {
+ final AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(id);
+ Awaitility.await().atMost(120, TimeUnit.SECONDS)
+ .until(() -> assertEquals(
+ "Expected unprocessedEventMap is drained but actual is "
+ + sender.getStatistics().getUnprocessedEventMapSize(),
+ queueSize, sender.getStatistics().getUnprocessedEventMapSize()));
+ assertEquals(queueSize, sender.getStatistics().getUnprocessedEventMapSize());
+ }
+
/**
* This method is specifically written for pause and stop operations. This method validates that
* the region size remains same for at least minimum number of verification attempts and also it
@@ -3053,6 +3080,31 @@ public class WANTestBase extends DistributedTestCase {
});
}
+ public static Integer getSecondaryQueueContentSize(final String senderId) {
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ GatewaySender sender = null;
+ for (GatewaySender s : senders) {
+ if (s.getId().equals(senderId)) {
+ sender = s;
+ break;
+ }
+ }
+ AbstractGatewaySender abstractSender = (AbstractGatewaySender) sender;
+ int size = abstractSender.getEventSecondaryQueueSize();
+ return size;
+ }
+
+ public static String displayQueueContent(final RegionQueue queue) {
+ if (queue instanceof ParallelGatewaySenderQueue) {
+ ParallelGatewaySenderQueue pgsq = (ParallelGatewaySenderQueue) queue;
+ return pgsq.displayContent();
+ } else if (queue instanceof ConcurrentParallelGatewaySenderQueue) {
+ ConcurrentParallelGatewaySenderQueue pgsq = (ConcurrentParallelGatewaySenderQueue) queue;
+ return pgsq.displayContent();
+ }
+ return null;
+ }
+
public static Integer getQueueContentSize(final String senderId) {
return getQueueContentSize(senderId, false);
}
@@ -3135,14 +3187,22 @@ public class WANTestBase extends DistributedTestCase {
((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0];
Set<BucketRegion> buckets = ((PartitionedRegion) regionQueue.getRegion()).getDataStore()
.getAllLocalPrimaryBucketRegions();
- for (final BucketRegion bucket : buckets) {
- Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> {
- assertEquals("Expected bucket entries for bucket: " + bucket.getId()
- + " is: 0 but actual entries: " + bucket.keySet().size() + " This bucket isPrimary: "
- + bucket.getBucketAdvisor().isPrimary() + " KEYSET: " + bucket.keySet(), 0,
- bucket.keySet().size());
- });
- } // for loop ends
+ final AbstractGatewaySender abstractSender = (AbstractGatewaySender) sender;
+ RegionQueue queue = abstractSender.getEventProcessor().queue;
+ Awaitility.await().atMost(60, TimeUnit.SECONDS).until(() -> {
+ assertEquals("Expected events in all primary queues are drained but actual is "
+ + abstractSender.getEventQueueSize() + ". Queue content is: "
+ + displayQueueContent(queue), 0, abstractSender.getEventQueueSize());
+ });
+ assertEquals("Expected events in all primary queues after drain is 0", 0,
+ abstractSender.getEventQueueSize());
+ Awaitility.await().atMost(120, TimeUnit.SECONDS).until(() -> {
+ assertEquals("Expected events in all secondary queues are drained but actual is "
+ + abstractSender.getEventSecondaryQueueSize() + ". Queue content is: "
+ + displayQueueContent(queue), 0, abstractSender.getEventSecondaryQueueSize());
+ });
+ assertEquals("Except events in all secondary queues after drain is 0", 0,
+ abstractSender.getEventSecondaryQueueSize());
} finally {
exp.remove();
exp1.remove();
diff --git a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
index eaef4f9..f5b98b7 100644
--- a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
+++ b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java
@@ -20,6 +20,8 @@ import static org.apache.geode.internal.cache.tier.sockets.Message.MAX_MESSAGE_S
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.assertj.core.api.Assertions.assertThat;
+import java.util.ArrayList;
+
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -407,18 +409,42 @@ public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase {
vm2.invoke(() -> validateRegionSizeRemainsSame(getTestMethodName() + "_PR", 200));
// SECOND RUN: start async puts on region
- AsyncInvocation async = vm4.invokeAsync(() -> doPuts(getTestMethodName() + "_PR", 5000));
-
- // when puts are happening by another thread, start the senders
- startSenderInVMsAsync("ln", vm4, vm5, vm6, vm7);
-
- async.join();
+ ArrayList<Integer> vm4List = null;
+ ArrayList<Integer> vm5List = null;
+ ArrayList<Integer> vm6List = null;
+ ArrayList<Integer> vm7List = null;
+ boolean foundDroppedAtYetStartedPrimarySender = false;
+ int count = 0;
+
+ do {
+ stopSenders();
+ AsyncInvocation async = vm4.invokeAsync(() -> doPuts(getTestMethodName() + "_PR", 5000));
+
+ // when puts are happening by another thread, start the senders
+ startSenderInVMsAsync("ln", vm4, vm5, vm6, vm7);
+
+ async.join();
+ vm4List =
+ (ArrayList<Integer>) vm4.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln"));
+ vm5List =
+ (ArrayList<Integer>) vm5.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln"));
+ vm6List =
+ (ArrayList<Integer>) vm6.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln"));
+ vm7List =
+ (ArrayList<Integer>) vm7.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln"));
+ if (vm4List.get(0) + vm5List.get(0) + vm6List.get(0) + vm7List.get(0) > 0) {
+ foundDroppedAtYetStartedPrimarySender = true;
+ }
+ count++;
+ } while (foundDroppedAtYetStartedPrimarySender == false && count < 5);
+ assertThat(foundDroppedAtYetStartedPrimarySender);
// verify all the buckets on all the sender nodes are drained
validateParallelSenderQueueAllBucketsDrained();
// verify that the queue size ultimately becomes zero. That means all the events propagate to
// remote site.
+
vm4.invoke(() -> validateQueueContents("ln", 0));
}
diff --git a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
index ee43b83..8df5650 100644
--- a/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
+++ b/geode-wan/src/test/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
@@ -269,6 +269,53 @@ public class SerialGatewaySenderOperationsDUnitTest extends WANTestBase {
}
@Test
+ public void testRestartSerialGatewaySendersWhilePutting() throws Throwable {
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ createCacheInVMs(nyPort, vm2, vm3);
+ createReceiverInVMs(vm2, vm3);
+
+ createSenderCaches(lnPort);
+
+ createSenderVM4();
+ createSenderVM5();
+
+ createReceiverRegions();
+
+ createSenderRegions();
+
+ vm7.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 20));
+
+ startSenderInVMs("ln", vm4, vm5);
+
+ vm7.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 20));
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 20));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 20));
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+
+ vm4.invoke(() -> SerialGatewaySenderOperationsDUnitTest.verifySenderStoppedState("ln"));
+ vm5.invoke(() -> SerialGatewaySenderOperationsDUnitTest.verifySenderStoppedState("ln"));
+
+ vm4.invoke(() -> WANTestBase.validateQueueSizeStat("ln", 0));
+ vm5.invoke(() -> WANTestBase.validateQueueSizeStat("ln", 0));
+
+ // do a lot of puts while senders are restarting
+ AsyncInvocation async = vm7.invokeAsync(() -> doPuts(getTestMethodName() + "_RR", 5000));
+
+ startSenderInVMsAsync("ln", vm4, vm5);
+ async.join();
+
+ vm4.invoke(() -> WANTestBase.validateQueueSizeStat("ln", 0));
+ vm5.invoke(() -> WANTestBase.validateQueueSizeStat("ln", 0));
+ vm4.invoke(() -> WANTestBase.validateSecondaryQueueSizeStat("ln", 0));
+ vm5.invoke(() -> WANTestBase.validateSecondaryQueueSizeStat("ln", 0));
+ }
+
+ @Test
public void testStopOneSerialGatewaySenderBothPrimary() throws Throwable {
Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
@@ -298,7 +345,7 @@ public class SerialGatewaySenderOperationsDUnitTest extends WANTestBase {
vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 200));
vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_RR", 200));
- // Do some puts while restarting a sender
+ // Do some puts from both vm4 and vm5 while restarting a sender
AsyncInvocation asyncPuts =
vm4.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_RR", 300));
--
To stop receiving notification emails like this one, please contact
zhouxj@apache.org.