You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bo...@apache.org on 2018/06/29 19:43:38 UTC
[geode] branch feature/GEM-1772 updated (311d712 -> 7c17f1c)
This is an automated email from the ASF dual-hosted git repository.
boglesby pushed a change to branch feature/GEM-1772
in repository https://gitbox.apache.org/repos/asf/geode.git.
discard 311d712 GEM-1772: Added additional debugging
discard 0483f40 GEM-1772: Added additional debugging
discard 9435365 GEM-1772: Added additional debugging
discard 0b07ac3 GEM-1772: Added debugging
add 1380562 Introduces Serializable Consumer for Cluster junit (#2067)
add 982d332 GEODE-5285: Make PdxInstanceImplTest a unit test instead of an integration test (#2064)
add f5eab20 GEODE-5330: only create one temp folder (#2068)
add 7d7a651 Fix test failure related to #2067
add ec0501d GEODE-5109: Ignoring the test
add 13087e2 GEODE-5333: Destroy failed connection in ConnectionConnector (#2072)
add f2aeed4 User Guide: correct a CQ typo
add 908a5ef GEODE-5335: Do not resolve addresses when calling netstat and lsof (#2070)
add 226e406 GEODE-4791: Fix intellij with gradle 4.8 by removing provided configuration
add 880a9d6 Write a test that hangs without the fix for GEODE-3563. (#2057)
add 87c0ec3 GEODE-4511: rework ClientAuthDUnitRule to illustrate when authenticat… (#2077)
add af0ac3d GEODE-1655: CI Failure in AutoConnectionSourceDUnitTest.testClientMembershipListener
add 6ae3cad GEODE-5346: Fix race in LauncherLifecycleCommandsDUnitTest (#2080)
add d8538ee GEODE-1198 CI Failure: DistributedSystemDUnitTest.testConflictingUDPPort
new 3e9a15d GEM-1772: Added debugging
new 9147b1e GEM-1772: Added additional debugging
new 9157dd2 GEM-1772: Added additional debugging
new 68e529e GEM-1772: Added additional debugging
new 7c17f1c GEM-1772: Added additional debugging
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 (311d712)
\
N -- N -- N refs/heads/feature/GEM-1772 (7c17f1c)
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 5 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:
extensions/session-testing-war/build.gradle | 3 +-
geode-assembly/build.gradle | 5 +-
.../LauncherLifecycleCommandsDUnitTest.java | 50 ++--
.../geode/connectors/jdbc/JdbcDistributedTest.java | 4 +-
geode-core/build.gradle | 10 +-
.../cache/client/internal/ConnectionConnector.java | 63 ++--
.../client/internal/ConnectionFactoryImpl.java | 23 +-
.../distributed/internal/DistributionAdvisor.java | 40 ++-
.../internal/cache/AbstractUpdateOperation.java | 5 +-
.../apache/geode/internal/cache/BucketRegion.java | 2 +-
.../internal/cache/DistributedCacheOperation.java | 19 +-
.../apache/geode/internal/cache/LocalRegion.java | 2 +-
.../geode/internal/cache/PartitionedRegion.java | 2 +-
.../geode/internal/cache/StateFlushOperation.java | 5 +-
.../org/apache/geode/internal/cache/TXState.java | 2 +-
.../internal/cache/tx/RemotePutAllMessage.java | 2 +-
.../java/org/apache/geode/internal/util/Hex.java | 16 +
.../internal/cli/functions/NetstatFunction.java | 14 +-
.../geode/pdx/internal/LonerTypeRegistration.java | 10 +-
.../geode/pdx/internal/PeerTypeRegistration.java | 2 +-
.../apache/geode/pdx/internal/TypeRegistry.java | 5 +
.../internal/AutoConnectionSourceDUnitTest.java | 19 +-
.../client/internal/ConnectionConnectorTest.java | 114 +++++++
.../distributed/DistributedSystemDUnitTest.java | 11 +-
.../geode/internal/tcp/TCPConduitDUnitTest.java | 128 ++++++++
.../management/internal/cli/NetstatDUnitTest.java | 149 ++++++++--
.../geode/pdx/internal/PdxInstanceImplTest.java | 329 +++++++++++++++++----
.../apache/geode/security/ClientAuthDUnitTest.java | 64 ++--
...tAuthorizationLegacyConfigurationDUnitTest.java | 14 +-
...aAuthorizationUsingLegacySecurityDUnitTest.java | 23 +-
...onUsingLegacySecurityWithFailoverDUnitTest.java | 13 +-
.../geode/test/dunit/DistributedTestUtils.java | 17 +-
...CallableIF.java => SerializableConsumerIF.java} | 9 +-
.../geode/test/dunit/rules/ClusterStartupRule.java | 48 +--
...tartupRuleCanSpecifyOlderVersionsDUnitTest.java | 8 +-
.../geode/test/junit/rules/GfshCommandRule.java | 38 ++-
geode-cq/build.gradle | 3 +-
.../commands/DescribeClientCommandDUnitTest.java | 5 +-
.../how_continuous_querying_works.html.md.erb | 3 +-
.../test/dunit/standalone/VersionManager.java | 0
.../dunit/standalone/VersionManagerJUnitTest.java | 0
.../lucene/LuceneQueriesReindexDUnitTest.java | 2 +
geode-old-versions/build.gradle | 9 +-
geode-protobuf/build.gradle | 3 +-
geode-pulse/build.gradle | 8 +-
geode-rebalancer/build.gradle | 6 +-
geode-wan/build.gradle | 3 +-
geode-web-api/build.gradle | 6 +-
geode-web/build.gradle | 11 +-
gradle/ide.gradle | 2 -
gradle/java.gradle | 15 +-
gradle/wrapper/gradle-wrapper.properties | 4 +-
52 files changed, 988 insertions(+), 360 deletions(-)
create mode 100644 geode-core/src/test/java/org/apache/geode/cache/client/internal/ConnectionConnectorTest.java
create mode 100644 geode-core/src/test/java/org/apache/geode/internal/tcp/TCPConduitDUnitTest.java
copy geode-core/src/test/java/org/apache/geode/test/dunit/{SerializableCallableIF.java => SerializableConsumerIF.java} (80%)
rename {geode-old-versions => geode-junit}/src/main/java/org/apache/geode/test/dunit/standalone/VersionManager.java (100%)
rename {geode-old-versions => geode-junit}/src/test/java/org/apache/geode/test/dunit/standalone/VersionManagerJUnitTest.java (100%)
[geode] 02/05: GEM-1772: Added additional debugging
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boglesby pushed a commit to branch feature/GEM-1772
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 9147b1e25d2398a63f175e2c273e731d749ef39a
Author: Barry Oglesby <bo...@pivotal.io>
AuthorDate: Wed Jun 20 13:01:11 2018 -0700
GEM-1772: Added additional debugging
---
.../distributed/internal/ReplyProcessor21.java | 6 ++--
.../distributed/internal/direct/DirectChannel.java | 8 ++---
.../internal/cache/DistributedCacheOperation.java | 14 +++++---
.../internal/cache/DistributedPutAllOperation.java | 14 +++++---
.../internal/cache/InitialImageOperation.java | 37 ++++++++++++----------
.../cache/partitioned/PutAllPRMessage.java | 16 +++++-----
6 files changed, 54 insertions(+), 41 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java
index d3cf1ad..fecfa41 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java
@@ -396,9 +396,9 @@ public class ReplyProcessor21 implements MembershipListener {
protected void process(DistributionMessage msg, boolean warn) {
if (msg instanceof PutAllPRMessage.PutAllReplyMessage) {
- //if (logger.isDebugEnabled()) {
- logger.warn("XXX {} got process({}) from {}", this, msg, msg.getSender());
- //}
+ // if (logger.isDebugEnabled()) {
+ logger.warn("XXX {} got process({}) from {}", this, msg, msg.getSender());
+ // }
}
if (msg instanceof ReplyMessage) {
ReplyException ex = ((ReplyMessage) msg).getException();
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
index eaa062e..664b600 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
@@ -317,10 +317,10 @@ public class DirectChannel {
directMsg.registerProcessor();
}
if (msg instanceof DistributedPutAllOperation.PutAllMessage) {
- //if (logger.isDebugEnabled()) {
- logger.warn("XXX DirectChannel sendToMany Sending ({}) to {} peers ({}) via tcp/ip",
- msg, p_destinations.length, Arrays.toString(p_destinations));
- //}
+ // if (logger.isDebugEnabled()) {
+ logger.warn("XXX DirectChannel sendToMany Sending ({}) to {} peers ({}) via tcp/ip",
+ msg, p_destinations.length, Arrays.toString(p_destinations));
+ // }
}
try {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
index 35dca1d..86ad5e0 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
@@ -453,10 +453,12 @@ public abstract class DistributedCacheOperation {
}
if (this instanceof DistributedPutAllOperation) {
- //if (logger.isDebugEnabled()) {
- logger.warn("XXX DistributedPutAllOperation _distribute recipients for {}: {} with adjunct messages to: {}", this, recipients,
- adjunctRecipients);
- //}
+ // if (logger.isDebugEnabled()) {
+ logger.warn(
+ "XXX DistributedPutAllOperation _distribute recipients for {}: {} with adjunct messages to: {}",
+ this, recipients,
+ adjunctRecipients);
+ // }
}
if (shouldAck) {
@@ -607,7 +609,9 @@ public abstract class DistributedCacheOperation {
}
if (failures != null && !failures.isEmpty()) {
- logger.warn("XXX DistributedPutAllOperation _distribute Failed sending ({}) to {} while processing event:{}", msg, failures, event);
+ logger.warn(
+ "XXX DistributedPutAllOperation _distribute Failed sending ({}) to {} while processing event:{}",
+ msg, failures, event);
}
Set<InternalDistributedMember> adjunctRecipientsWithNoCacheServer =
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
index 29c38af..70a97eb 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
@@ -1175,13 +1175,17 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation {
public void run() {
final boolean isDebugEnabled = logger.isDebugEnabled();
for (int i = 0; i < putAllDataSize; ++i) {
- //if (isDebugEnabled) {
- logger.warn("XXX PutAllMessage basicOperateOnRegion putAll processing {} with {} sender={}", putAllData[i],
- putAllData[i].versionTag, sender);
- //}
+ // if (isDebugEnabled) {
+ logger.warn(
+ "XXX PutAllMessage basicOperateOnRegion putAll processing {} with {} sender={}",
+ putAllData[i],
+ putAllData[i].versionTag, sender);
+ // }
putAllData[i].setSender(sender);
doEntryPut(putAllData[i], rgn);
- logger.warn("XXX PutAllMessage basicOperateOnRegion putAll done processing {} with {} sender={}", putAllData[i],
+ logger.warn(
+ "XXX PutAllMessage basicOperateOnRegion putAll done processing {} with {} sender={}",
+ putAllData[i],
putAllData[i].versionTag, sender);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java
index a265cdd..5621090 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java
@@ -563,7 +563,8 @@ public class InitialImageOperation {
}
if (this.gotImage) {
// TODO add localizedString
- logger.info("XXX {} is done getting image from {}. isDeltaGII is {}", this.region.getName(),
+ logger.info("XXX {} is done getting image from {}. isDeltaGII is {}",
+ this.region.getName(),
recipient, this.isDeltaGII);
} else {
// TODO add localizedString
@@ -1304,9 +1305,11 @@ public class InitialImageOperation {
// bug 37461: don't allow abort flag to be reset
boolean isAborted = this.abort; // volatile fetch
if (!isAborted) {
- logger.warn("XXX ImageProcessor.process about to process chunk entries=" + m.entries);
+ logger
+ .warn("XXX ImageProcessor.process about to process chunk entries=" + m.entries);
isAborted = !processChunk(m.entries, m.getSender(), m.remoteVersion);
- logger.warn("XXX ImageProcessor.process processed chunk entries isAborted=" + isAborted);
+ logger.warn(
+ "XXX ImageProcessor.process processed chunk entries isAborted=" + isAborted);
if (isAborted) {
this.abort = true; // volatile store
} else {
@@ -1672,9 +1675,9 @@ public class InitialImageOperation {
final int seriesNum = 0;
// chunkEntries returns false if didn't finish
- //if (isGiiDebugEnabled) {
- logger.warn("XXX RequestImageMessage: Starting chunkEntries for {}", rgn.getFullPath());
- //}
+ // if (isGiiDebugEnabled) {
+ logger.warn("XXX RequestImageMessage: Starting chunkEntries for {}", rgn.getFullPath());
+ // }
final InitialImageFlowControl flowControl =
InitialImageFlowControl.register(dm, getSender());
@@ -1732,17 +1735,17 @@ public class InitialImageOperation {
&& rgn.getConcurrencyChecksEnabled()) {
synchronized (rgn.getCache().getTombstoneService().getBlockGCLock()) {
if (goWithFullGII(rgn, this.versionVector)) {
- //if (isGiiDebugEnabled) {
- logger.warn("XXX Doing full GII for {}", rgn.getFullPath());
- //}
+ // if (isGiiDebugEnabled) {
+ logger.warn("XXX Doing full GII for {}", rgn.getFullPath());
+ // }
this.versionVector = null; // full GII
} else {
// lock GIILock only for deltaGII
int count = rgn.getCache().getTombstoneService().incrementGCBlockCount();
markedOngoingGII = true;
- //if (isGiiDebugEnabled) {
+ // if (isGiiDebugEnabled) {
logger.warn("XX Doing delta GII for {}", rgn.getFullPath());
- //}
+ // }
}
}
}
@@ -1780,7 +1783,8 @@ public class InitialImageOperation {
if (this.last && rgn.getVersionVector() != null) {
gcVersions = rgn.getVersionVector().getMemberToGCVersion();
}
- logger.warn("XXX RequestImageMessage.executeWith about to send chunk " + (msgNum+1));
+ logger.warn("XXX RequestImageMessage.executeWith about to send chunk "
+ + (msgNum + 1));
replyWithData(dm, entries, seriesNum, msgNum++, numSeries, this.last,
flowControlId,
versionVector != null, holderToSend, gcVersions);
@@ -1794,10 +1798,11 @@ public class InitialImageOperation {
});
- //if (isGiiDebugEnabled) {
- logger.warn("XXX RequestImageMessage: ended chunkEntries for {}; finished = {}", rgn.getFullPath(),
- finished);
- //}
+ // if (isGiiDebugEnabled) {
+ logger.warn("XXX RequestImageMessage: ended chunkEntries for {}; finished = {}",
+ rgn.getFullPath(),
+ finished);
+ // }
// Call to chunkEntries above will have sent at least one
// reply with last==true for the last message. (unless doing abortTest or
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
index cca7891..aae61a2 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
@@ -194,9 +194,9 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
PutAllResponse p = new PutAllResponse(r.getSystem(), recipients);
initMessage(r, recipients, false, p);
setTransactionDistributed(r.getCache().getTxManager().isDistributed());
- //if (logger.isDebugEnabled()) {
- logger.warn("XXX PutAllPRMessage.send: recipient is {}, msg is {}", recipient, this);
- //}
+ // if (logger.isDebugEnabled()) {
+ logger.warn("XXX PutAllPRMessage.send: recipient is {}, msg is {}", recipient, this);
+ // }
Set failures = r.getDistributionManager().putOutgoing(this);
if (failures != null && failures.size() > 0) {
@@ -402,11 +402,11 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
baseEvent.setContext(this.bridgeContext);
}
baseEvent.setPossibleDuplicate(this.posDup);
- //if (logger.isDebugEnabled()) {
- logger.warn(
- "XXX PutAllPRMessage.doLocalPutAll: eventSender is {}, baseEvent is {}, msg is {}",
- eventSender, baseEvent, this);
- //}
+ // if (logger.isDebugEnabled()) {
+ logger.warn(
+ "XXX PutAllPRMessage.doLocalPutAll: eventSender is {}, baseEvent is {}, msg is {}",
+ eventSender, baseEvent, this);
+ // }
dpao = new DistributedPutAllOperation(baseEvent, putAllPRDataSize, false);
}
[geode] 04/05: GEM-1772: Added additional debugging
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boglesby pushed a commit to branch feature/GEM-1772
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 68e529e270f5cd49cd58411236473b89d2bf8bc1
Author: Barry Oglesby <bo...@pivotal.io>
AuthorDate: Wed Jun 27 15:17:40 2018 -0700
GEM-1772: Added additional debugging
---
.../geode/internal/cache/StateFlushOperation.java | 29 +++++++++++++++++++---
1 file changed, 26 insertions(+), 3 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java
index 21b2367..11e2e2d 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java
@@ -98,6 +98,11 @@ public class StateFlushOperation {
private DistributionManager dm;
+ private static void log(String message) {
+ // System.out.println(Thread.currentThread().getName() + ": " + message);
+ logger.warn("XXX " + message);
+ }
+
/** flush current ops to the given members for the given region */
public static void flushTo(Set<InternalDistributedMember> targets, DistributedRegion region) {
DistributionManager dm = region.getDistributionManager();
@@ -223,6 +228,7 @@ public class StateFlushOperation {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {} with processor {}", smm,
gfprocessor);
}
+ log("StateFlushOperation.flush about to send smm=" + smm);
Set failures = this.dm.putOutgoing(smm);
if (failures != null) {
if (failures.contains(target)) {
@@ -238,6 +244,7 @@ public class StateFlushOperation {
try {
gfprocessor.waitForReplies();
+ log("StateFlushOperation.flush received replies for smm=" + smm);
if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Finished processing {}", smm);
}
@@ -342,6 +349,8 @@ public class StateFlushOperation {
@Override
protected void process(ClusterDistributionManager dm) {
+ log("StateMarkerMessage.process message=" + this + "; relay="
+ + dm.getDistributionManagerId().equals(relayRecipient));
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Processing {}", this);
if (dm.getDistributionManagerId().equals(relayRecipient)) {
try {
@@ -373,6 +382,7 @@ public class StateFlushOperation {
if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {}", ga);
}
+ log("StateMarkerMessage.process about to send message=" + ga);
dm.putOutgoing(ga);
}
} else {
@@ -386,6 +396,7 @@ public class StateFlushOperation {
gr.setRecipient((InternalDistributedMember) relayRecipient);
gr.requestingMember = this.getSender();
gr.processorId = processorId;
+ log("StateMarkerMessage.process created message=" + gr);
try {
Set<DistributedRegion> regions = getRegions(dm);
for (DistributedRegion r : regions) {
@@ -431,6 +442,7 @@ public class StateFlushOperation {
if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {}", gr);
}
+ log("StateMarkerMessage.process sending message=" + gr);
dm.putOutgoing(gr);
}
}
@@ -444,7 +456,9 @@ public class StateFlushOperation {
// ops
}
try {
+ log("StateMarkerMessage.waitForCurrentOperations waiting region=" + r.getFullPath());
r.getDistributionAdvisor().waitForCurrentOperations();
+ log("StateMarkerMessage.waitForCurrentOperations done waiting region=" + r.getFullPath());
} catch (RegionDestroyedException e) {
// continue with the next region
}
@@ -565,11 +579,17 @@ public class StateFlushOperation {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Waiting for channel states: {}",
channelStateDescription(channelState));
}
+ log("StateStabilizationMessage.run waiting for channel states: "
+ + channelStateDescription(channelState));
for (;;) {
dm.getCancelCriterion().checkCancelInProgress(null);
boolean interrupted = Thread.interrupted();
try {
+ log("StateStabilizationMessage.run about to waitForMessageState sender="
+ + getSender());
dm.getMembershipManager().waitForMessageState(getSender(), channelState);
+ log("StateStabilizationMessage.run done waitForMessageState sender="
+ + getSender());
break;
} catch (InterruptedException ignore) {
interrupted = true;
@@ -611,6 +631,7 @@ public class StateFlushOperation {
if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {}", ga);
}
+ log("StateStabilizationMessage.run about to send message=" + ga);
if (requestingMember.equals(dm.getDistributionManagerId())) {
ga.dmProcess(dm);
} else {
@@ -678,9 +699,8 @@ public class StateFlushOperation {
@Override
public void process(final DistributionManager dm, final ReplyProcessor21 processor) {
- if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
- logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Processing {}", this);
- }
+ log("StateStabilizedMessage.process message=" + this);
+ logger.warn("XXX Processing {}", this);
super.process(dm, processor);
}
@@ -761,6 +781,7 @@ public class StateFlushOperation {
@Override
public void memberDeparted(DistributionManager distributionManager,
final InternalDistributedMember id, final boolean crashed) {
+ log("StateFlushReplyProcessor.memberDeparted id=" + id);
super.memberDeparted(distributionManager, id, crashed);
}
@@ -768,6 +789,8 @@ public class StateFlushOperation {
protected void processActiveMembers(Set activeMembers) {
super.processActiveMembers(activeMembers);
if (!activeMembers.contains(this.targetMember)) {
+ log("StateFlushReplyProcessor.processActiveMembers targetMemberHasLeft targetMember="
+ + this.targetMember);
targetMemberHasLeft = true;
}
}
[geode] 03/05: GEM-1772: Added additional debugging
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boglesby pushed a commit to branch feature/GEM-1772
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 9157dd23b2db58e6cd2a02a71fef9aab2dab7982
Author: Barry Oglesby <bo...@pivotal.io>
AuthorDate: Wed Jun 20 18:20:42 2018 -0700
GEM-1772: Added additional debugging
---
.../apache/geode/distributed/internal/direct/DirectChannel.java | 7 +++++++
.../src/main/java/org/apache/geode/internal/tcp/MsgStreamer.java | 7 +++++++
2 files changed, 14 insertions(+)
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
index 664b600..83abd90 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
@@ -342,6 +342,10 @@ public class DirectChannel {
final List cons = new ArrayList(destinations.length);
ConnectExceptions ce = getConnections(mgr, msg, destinations, orderedMsg, retry, ackTimeout,
ackSDTimeout, cons);
+ if (msg instanceof DistributedPutAllOperation.PutAllMessage) {
+ logger
+ .warn("XXX DirectChannel sendToMany got connections=" + cons + "; exceptions=" + ce);
+ }
if (directReply && msg.getProcessorId() > 0) { // no longer a direct-reply message?
directReply = false;
}
@@ -397,6 +401,9 @@ public class DirectChannel {
startTime = System.currentTimeMillis();
}
ms.reserveConnections(startTime, ackTimeout, ackSDTimeout);
+ if (msg instanceof DistributedPutAllOperation.PutAllMessage) {
+ logger.warn("XXX DirectChannel sendToMany reserved connections");
+ }
int result = ms.writeMessage();
if (bytesWritten == 0) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgStreamer.java b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgStreamer.java
index af54679..59dc3b3 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgStreamer.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/tcp/MsgStreamer.java
@@ -37,6 +37,7 @@ import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.ObjToByteArraySerializer;
import org.apache.geode.internal.Version;
+import org.apache.geode.internal.cache.DistributedPutAllOperation;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
@@ -313,8 +314,14 @@ public class MsgStreamer extends OutputStream
for (Iterator it = this.cons.iterator(); it.hasNext();) {
Connection con = (Connection) it.next();
try {
+ if (this.msg instanceof DistributedPutAllOperation.PutAllMessage) {
+ logger.warn("XXX MsgStreamer realFlush about to write con=" + con + "; msg=" + this.msg);
+ }
con.sendPreserialized(this.buffer,
lastFlushForMessage && this.msg.containsRegionContentChange(), conflationMsg);
+ if (this.msg instanceof DistributedPutAllOperation.PutAllMessage) {
+ logger.warn("XXX MsgStreamer realFlush completed write con=" + con + "; msg=" + this.msg);
+ }
} catch (IOException ex) {
it.remove();
if (this.ce == null)
[geode] 01/05: GEM-1772: Added debugging
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boglesby pushed a commit to branch feature/GEM-1772
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 3e9a15d8a2682cd320fe82a2926a92b674da986a
Author: Barry Oglesby <bo...@pivotal.io>
AuthorDate: Tue Jun 19 12:00:09 2018 -0700
GEM-1772: Added debugging
---
.../distributed/internal/ReplyProcessor21.java | 7 ++--
.../distributed/internal/direct/DirectChannel.java | 9 +++--
.../internal/cache/DistributedCacheOperation.java | 12 ++++---
.../internal/cache/DistributedPutAllOperation.java | 8 +++--
.../internal/cache/InitialImageOperation.java | 38 ++++++++++++----------
.../cache/partitioned/PutAllPRMessage.java | 20 +++++++-----
.../geode/internal/i18n/LocalizedStrings.java | 2 +-
7 files changed, 56 insertions(+), 40 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java
index 3ffd31d..d3cf1ad 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyProcessor21.java
@@ -35,6 +35,7 @@ import org.apache.geode.i18n.StringId;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.DSFIDNotFoundException;
import org.apache.geode.internal.Version;
+import org.apache.geode.internal.cache.partitioned.PutAllPRMessage;
import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
@@ -394,8 +395,10 @@ public class ReplyProcessor21 implements MembershipListener {
}
protected void process(DistributionMessage msg, boolean warn) {
- if (logger.isDebugEnabled()) {
- logger.debug("{} got process({}) from {}", this, msg, msg.getSender());
+ if (msg instanceof PutAllPRMessage.PutAllReplyMessage) {
+ //if (logger.isDebugEnabled()) {
+ logger.warn("XXX {} got process({}) from {}", this, msg, msg.getSender());
+ //}
}
if (msg instanceof ReplyMessage) {
ReplyException ex = ((ReplyMessage) msg).getException();
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
index 970957f..eaa062e 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/direct/DirectChannel.java
@@ -47,6 +47,7 @@ import org.apache.geode.distributed.internal.membership.InternalDistributedMembe
import org.apache.geode.distributed.internal.membership.MembershipManager;
import org.apache.geode.i18n.StringId;
import org.apache.geode.internal.cache.DirectReplyMessage;
+import org.apache.geode.internal.cache.DistributedPutAllOperation;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.AlertAppender;
@@ -315,9 +316,11 @@ public class DirectChannel {
if (!directReply && directMsg != null) {
directMsg.registerProcessor();
}
- if (logger.isDebugEnabled()) {
- logger.debug("Sending ({}) to {} peers ({}) via tcp/ip",
- msg, p_destinations.length, Arrays.toString(p_destinations));
+ if (msg instanceof DistributedPutAllOperation.PutAllMessage) {
+ //if (logger.isDebugEnabled()) {
+ logger.warn("XXX DirectChannel sendToMany Sending ({}) to {} peers ({}) via tcp/ip",
+ msg, p_destinations.length, Arrays.toString(p_destinations));
+ //}
}
try {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
index 4fee0a0..35dca1d 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
@@ -452,9 +452,11 @@ public abstract class DistributedCacheOperation {
}
}
- if (logger.isDebugEnabled()) {
- logger.debug("recipients for {}: {} with adjunct messages to: {}", this, recipients,
- adjunctRecipients);
+ if (this instanceof DistributedPutAllOperation) {
+ //if (logger.isDebugEnabled()) {
+ logger.warn("XXX DistributedPutAllOperation _distribute recipients for {}: {} with adjunct messages to: {}", this, recipients,
+ adjunctRecipients);
+ //}
}
if (shouldAck) {
@@ -604,8 +606,8 @@ public abstract class DistributedCacheOperation {
cachelessNodes.addAll(cachelessNodesWithNoCacheServer);
}
- if (failures != null && !failures.isEmpty() && logger.isDebugEnabled()) {
- logger.debug("Failed sending ({}) to {} while processing event:{}", msg, failures, event);
+ if (failures != null && !failures.isEmpty()) {
+ logger.warn("XXX DistributedPutAllOperation _distribute Failed sending ({}) to {} while processing event:{}", msg, failures, event);
}
Set<InternalDistributedMember> adjunctRecipientsWithNoCacheServer =
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
index 3fd97d1..29c38af 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedPutAllOperation.java
@@ -1175,12 +1175,14 @@ public class DistributedPutAllOperation extends AbstractUpdateOperation {
public void run() {
final boolean isDebugEnabled = logger.isDebugEnabled();
for (int i = 0; i < putAllDataSize; ++i) {
- if (isDebugEnabled) {
- logger.debug("putAll processing {} with {} sender={}", putAllData[i],
+ //if (isDebugEnabled) {
+ logger.warn("XXX PutAllMessage basicOperateOnRegion putAll processing {} with {} sender={}", putAllData[i],
putAllData[i].versionTag, sender);
- }
+ //}
putAllData[i].setSender(sender);
doEntryPut(putAllData[i], rgn);
+ logger.warn("XXX PutAllMessage basicOperateOnRegion putAll done processing {} with {} sender={}", putAllData[i],
+ putAllData[i].versionTag, sender);
}
}
}, ev.getEventId());
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java
index 1f04280..a265cdd 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InitialImageOperation.java
@@ -563,11 +563,11 @@ public class InitialImageOperation {
}
if (this.gotImage) {
// TODO add localizedString
- logger.info("{} is done getting image from {}. isDeltaGII is {}", this.region.getName(),
+ logger.info("XXX {} is done getting image from {}. isDeltaGII is {}", this.region.getName(),
recipient, this.isDeltaGII);
} else {
// TODO add localizedString
- logger.info("{} failed to get image from {}", this.region.getName(), recipient);
+ logger.info("XXX {} failed to get image from {}", this.region.getName(), recipient);
}
if (this.region.getDataPolicy().withPersistence()) {
logger.info(LocalizedMessage.create(
@@ -1304,7 +1304,9 @@ public class InitialImageOperation {
// bug 37461: don't allow abort flag to be reset
boolean isAborted = this.abort; // volatile fetch
if (!isAborted) {
+ logger.warn("XXX ImageProcessor.process about to process chunk entries=" + m.entries);
isAborted = !processChunk(m.entries, m.getSender(), m.remoteVersion);
+ logger.warn("XXX ImageProcessor.process processed chunk entries isAborted=" + isAborted);
if (isAborted) {
this.abort = true; // volatile store
} else {
@@ -1670,10 +1672,9 @@ public class InitialImageOperation {
final int seriesNum = 0;
// chunkEntries returns false if didn't finish
- if (isGiiDebugEnabled) {
- logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
- "RequestImageMessage: Starting chunkEntries for {}", rgn.getFullPath());
- }
+ //if (isGiiDebugEnabled) {
+ logger.warn("XXX RequestImageMessage: Starting chunkEntries for {}", rgn.getFullPath());
+ //}
final InitialImageFlowControl flowControl =
InitialImageFlowControl.register(dm, getSender());
@@ -1731,18 +1732,17 @@ public class InitialImageOperation {
&& rgn.getConcurrencyChecksEnabled()) {
synchronized (rgn.getCache().getTombstoneService().getBlockGCLock()) {
if (goWithFullGII(rgn, this.versionVector)) {
- if (isGiiDebugEnabled) {
- logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE, "have to do fullGII");
- }
+ //if (isGiiDebugEnabled) {
+ logger.warn("XXX Doing full GII for {}", rgn.getFullPath());
+ //}
this.versionVector = null; // full GII
} else {
// lock GIILock only for deltaGII
int count = rgn.getCache().getTombstoneService().incrementGCBlockCount();
markedOngoingGII = true;
- if (isGiiDebugEnabled) {
- logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE, "There're {} Delta GII on going",
- count);
- }
+ //if (isGiiDebugEnabled) {
+ logger.warn("XX Doing delta GII for {}", rgn.getFullPath());
+ //}
}
}
}
@@ -1780,9 +1780,11 @@ public class InitialImageOperation {
if (this.last && rgn.getVersionVector() != null) {
gcVersions = rgn.getVersionVector().getMemberToGCVersion();
}
+ logger.warn("XXX RequestImageMessage.executeWith about to send chunk " + (msgNum+1));
replyWithData(dm, entries, seriesNum, msgNum++, numSeries, this.last,
flowControlId,
versionVector != null, holderToSend, gcVersions);
+ logger.warn("XXX RequestImageMessage.executeWith sent chunk " + (msgNum));
}
return !abort;
} catch (CancelException e) {
@@ -1792,11 +1794,10 @@ public class InitialImageOperation {
});
- if (isGiiDebugEnabled) {
- logger.trace(LogMarker.INITIAL_IMAGE_VERBOSE,
- "RequestImageMessage: ended chunkEntries for {}; finished = {}", rgn.getFullPath(),
+ //if (isGiiDebugEnabled) {
+ logger.warn("XXX RequestImageMessage: ended chunkEntries for {}; finished = {}", rgn.getFullPath(),
finished);
- }
+ //}
// Call to chunkEntries above will have sent at least one
// reply with last==true for the last message. (unless doing abortTest or
@@ -1870,6 +1871,7 @@ public class InitialImageOperation {
// null chunk signals receiver that we are aborting
ImageReplyMessage.send(getSender(), processorId, rex, dm, null, 0, 0, 1, true, 0, false,
null, null);
+ logger.warn("XXX RequestImageMessage.process sent failure message");
} // !success
if (internalAfterSentImageReply != null
@@ -1934,6 +1936,7 @@ public class InitialImageOperation {
&& it.hasNext()) {
RegionEntry mapEntry = (RegionEntry) it.next();
Object key = mapEntry.getKey();
+ logger.warn("XXX RequestImageMessage.chunkEntries checking key=" + key);
if (rgn.checkEntryNotValid(mapEntry)) { // entry was just removed
continue;
}
@@ -2001,6 +2004,7 @@ public class InitialImageOperation {
entry.setLastModified(rgn.getDistributionManager(), mapEntry.getLastModified());
}
+ logger.warn("XXX RequestImageMessage.chunkEntries adding to chunk entry=" + entry);
chunkEntries.add(entry);
currentChunkSize += entry.calcSerializedSize();
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
index b6450b3..cca7891 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutAllPRMessage.java
@@ -194,9 +194,9 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
PutAllResponse p = new PutAllResponse(r.getSystem(), recipients);
initMessage(r, recipients, false, p);
setTransactionDistributed(r.getCache().getTxManager().isDistributed());
- if (logger.isDebugEnabled()) {
- logger.debug("PutAllPRMessage.send: recipient is {}, msg is {}", recipient, this);
- }
+ //if (logger.isDebugEnabled()) {
+ logger.warn("XXX PutAllPRMessage.send: recipient is {}, msg is {}", recipient, this);
+ //}
Set failures = r.getDistributionManager().putOutgoing(this);
if (failures != null && failures.size() > 0) {
@@ -317,6 +317,7 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
long startTime) throws EntryExistsException, ForceReattemptException, DataLocationException {
boolean sendReply = true;
+ logger.warn("XXX PutAllPRMessage.operateOnPartitionedRegion invoked msg=" + this);
InternalDistributedMember eventSender = getSender();
long lastModified = 0L;
@@ -330,6 +331,7 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
if (sendReply) {
sendReply(getSender(), getProcessorId(), dm, null, pr, startTime);
}
+ logger.warn("XXX PutAllPRMessage.operateOnPartitionedRegion completed msg=" + this);
return false;
}
@@ -400,11 +402,11 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
baseEvent.setContext(this.bridgeContext);
}
baseEvent.setPossibleDuplicate(this.posDup);
- if (logger.isDebugEnabled()) {
- logger.debug(
- "PutAllPRMessage.doLocalPutAll: eventSender is {}, baseEvent is {}, msg is {}",
+ //if (logger.isDebugEnabled()) {
+ logger.warn(
+ "XXX PutAllPRMessage.doLocalPutAll: eventSender is {}, baseEvent is {}, msg is {}",
eventSender, baseEvent, this);
- }
+ //}
dpao = new DistributedPutAllOperation(baseEvent, putAllPRDataSize, false);
}
@@ -474,8 +476,8 @@ public class PutAllPRMessage extends PartitionMessageWithDirectReply {
try {
didPut = r.getDataView().putEntryOnRemote(ev, false, false, null, false,
lastModified, true);
- if (didPut && logger.isDebugEnabled()) {
- logger.debug("PutAllPRMessage.doLocalPutAll:putLocally success for {}", ev);
+ if (didPut) {
+ logger.warn("XXX PutAllPRMessage.doLocalPutAll:putLocally success for {}", ev);
}
} catch (ConcurrentCacheModificationException e) {
didPut = true;
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 cd86460..e1d5e9b 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
@@ -5568,7 +5568,7 @@ public class LocalizedStrings {
public static final StringId CacheClientProxy_CQEXCEPTION_WHILE_CLOSING_NON_DURABLE_CQS_0 =
new StringId(4334, "CqException while closing non durable Cqs. {0}");
public static final StringId InitialImageOperation_REGION_0_REQUESTING_INITIAL_IMAGE_FROM_1 =
- new StringId(4336, "Region {0} requesting initial image from {1}");
+ new StringId(4336, "XXX Region {0} requesting initial image from {1}");
public static final StringId CacheClientUpdater_ERROR_WHILE_PROCESSING_THE_CQ_MESSAGE_PROBLEM_WITH_READING_MESSAGE_FOR_CQ_0 =
new StringId(4337,
"Error while processing the CQ Message. Problem with reading message for CQ# : {0}");
[geode] 05/05: GEM-1772: Added additional debugging
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boglesby pushed a commit to branch feature/GEM-1772
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 7c17f1c97136267a77ea1f7660b5f5658439dee4
Author: Barry Oglesby <bo...@pivotal.io>
AuthorDate: Fri Jun 29 12:42:01 2018 -0700
GEM-1772: Added additional debugging
---
.../distributed/internal/DistributionAdvisor.java | 40 ++++++++++++++++++++--
.../internal/cache/AbstractUpdateOperation.java | 5 +--
.../apache/geode/internal/cache/BucketRegion.java | 2 +-
.../internal/cache/DistributedCacheOperation.java | 19 ++++++++--
.../apache/geode/internal/cache/LocalRegion.java | 2 +-
.../geode/internal/cache/PartitionedRegion.java | 2 +-
.../geode/internal/cache/StateFlushOperation.java | 5 +--
.../org/apache/geode/internal/cache/TXState.java | 2 +-
.../internal/cache/tx/RemotePutAllMessage.java | 2 +-
9 files changed, 66 insertions(+), 13 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java
index 5ce42ed..ce9ed7c 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java
@@ -14,6 +14,8 @@
*/
package org.apache.geode.distributed.internal;
+import static org.apache.geode.internal.cache.DistributedCacheOperation.DISTRIBUTED_PUT_ALL_IN_PROGRESS;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
@@ -40,6 +42,7 @@ import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.Version;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.DistributedRegion;
+import org.apache.geode.internal.cache.StateFlushOperation;
import org.apache.geode.internal.cache.UpdateAttributesProcessor;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.versions.VersionSource;
@@ -626,6 +629,13 @@ public class DistributionAdvisor {
synchronized (this.opCountLock) {
previousVersionOpCount += currentVersionOpCount;
currentVersionOpCount = 0;
+ logger.warn(
+ "XXX DistributionAdvisor doPutProfile advisee="
+ + this.advisee.getFullPath() + "; membershipVersion=" + membershipVersion
+ + "; previousVersionOpCount=" + previousVersionOpCount
+ + "; currentVersionOpCount="
+ + currentVersionOpCount,
+ new Exception());
}
}
}
@@ -730,6 +740,12 @@ public class DistributionAdvisor {
synchronized (this.opCountLock) {
previousVersionOpCount += currentVersionOpCount;
currentVersionOpCount = 0;
+ logger.warn(
+ "XXX DistributionAdvisor forceNewMembershipVersion advisee="
+ + this.advisee.getFullPath() + "; membershipVersion=" + membershipVersion
+ + "; previousVersionOpCount=" + previousVersionOpCount + "; currentVersionOpCount="
+ + currentVersionOpCount,
+ new Exception());
if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_STATE_FLUSH_VERBOSE)) {
logger.trace(LogMarker.DISTRIBUTION_STATE_FLUSH_VERBOSE,
"advisor for {} forced new membership version to {} previousOpCount={}", getAdvisee(),
@@ -759,6 +775,12 @@ public class DistributionAdvisor {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "StateFlush current opcount incremented: {}",
currentVersionOpCount);
}
+ if (DISTRIBUTED_PUT_ALL_IN_PROGRESS.get()) {
+ logger.warn(
+ "XXX DistributionAdvisor startOperation advisee=" + this.advisee.getFullPath()
+ + "; membershipVersion=" + membershipVersion + "; currentVersionOpCount="
+ + currentVersionOpCount);
+ }
}
return membershipVersion;
}
@@ -776,13 +798,25 @@ public class DistributionAdvisor {
currentVersionOpCount--;
if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "StateFlush current opcount deccremented: {}", currentVersionOpCount);
+ "StateFlush current opcount decremented: {}", currentVersionOpCount);
+ }
+ if (DISTRIBUTED_PUT_ALL_IN_PROGRESS.get()) {
+ logger.warn(
+ "XXX DistributionAdvisor endOperation advisee=" + this.advisee.getFullPath()
+ + "; version=" + version + "; membershipVersion=" + membershipVersion
+ + "; currentVersionOpCount=" + currentVersionOpCount);
}
} else {
previousVersionOpCount--;
if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "StateFlush previous opcount incremented: {}", previousVersionOpCount);
+ "StateFlush previous opcount decremented: {}", previousVersionOpCount);
+ }
+ if (DISTRIBUTED_PUT_ALL_IN_PROGRESS.get()) {
+ logger.warn(
+ "XXX DistributionAdvisor endOperation advisee=" + this.advisee.getFullPath()
+ + "; version=" + version + "; membershipVersion=" + membershipVersion
+ + "; previousVersionOpCount=" + previousVersionOpCount);
}
}
}
@@ -818,6 +852,8 @@ public class DistributionAdvisor {
synchronized (this.opCountLock) {
opCount = this.previousVersionOpCount;
}
+ StateFlushOperation
+ .log("DistributionAdvisor.waitForCurrentOperations waiting opCount=" + opCount);
if (opCount <= 0) {
break;
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractUpdateOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractUpdateOperation.java
index 6ec4453..0da39a1 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractUpdateOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractUpdateOperation.java
@@ -42,7 +42,7 @@ import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
/**
- * Common code for both UpdateOperation and DistributedPutAllOperation.
+ * Common code for both UpdateOperation and DISTRIBUTED_PUT_ALL_OPERATION.
*
*/
public abstract class AbstractUpdateOperation extends DistributedCacheOperation {
@@ -262,7 +262,8 @@ public abstract class AbstractUpdateOperation extends DistributedCacheOperation
// @todo darrel: make this method static?
/**
* Do the actual update after operationOnRegion has confirmed work needs to be done Note this is
- * the default implementation used by UpdateOperation. DistributedPutAllOperation overrides and
+ * the default implementation used by UpdateOperation. DISTRIBUTED_PUT_ALL_OPERATION overrides
+ * and
* then calls back using super to this implementation. NOTE: be careful to not use methods like
* getEvent(); defer to the ev passed as a parameter instead.
*/
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 0fb745d..caeb95c 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
@@ -1844,7 +1844,7 @@ public class BucketRegion extends DistributedRegion implements Bucket {
* create a PutAllPRMessage for notify-only and send it to all adjunct nodes. return a set of
* members that should be attached to the operation's reply processor (if any)
*
- * @param dpao DistributedPutAllOperation object for PutAllMessage
+ * @param dpao DISTRIBUTED_PUT_ALL_OPERATION object for PutAllMessage
* @param cacheOpRecipients set of receiver which got cacheUpdateOperation.
* @param adjunctRecipients recipients that must unconditionally get the event
* @param filterRoutingInfo routing information for all members having the region
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
index 86ad5e0..cd8c41e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
@@ -254,7 +254,13 @@ public abstract class DistributedCacheOperation {
DistributedRegion region = getRegion();
long viewVersion = -1;
if (this.containsRegionContentChange()) {
+ if (this instanceof DistributedPutAllOperation) {
+ DISTRIBUTED_PUT_ALL_IN_PROGRESS.set(true);
+ logger.warn(
+ "XXX DistributedCacheOperation startOperation about to start operation msg=" + this);
+ }
viewVersion = region.getDistributionAdvisor().startOperation();
+ DISTRIBUTED_PUT_ALL_IN_PROGRESS.set(false);
}
if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "dispatching operation in view version {}",
@@ -276,6 +282,9 @@ public abstract class DistributedCacheOperation {
return viewVersion;
}
+ public static ThreadLocal<Boolean> DISTRIBUTED_PUT_ALL_IN_PROGRESS =
+ ThreadLocal.withInitial(() -> Boolean.FALSE);
+
/**
* region's distribution advisor marked that a distribution is ended. This method should pair with
* startOperation in try/finally block.
@@ -283,7 +292,13 @@ public abstract class DistributedCacheOperation {
public void endOperation(long viewVersion) {
DistributedRegion region = getRegion();
if (viewVersion != -1) {
+ if (this instanceof DistributedPutAllOperation) {
+ DISTRIBUTED_PUT_ALL_IN_PROGRESS.set(true);
+ }
region.getDistributionAdvisor().endOperation(viewVersion);
+ logger.warn(
+ "XXX DistributedCacheOperation endOperation operation msg=" + this);
+ DISTRIBUTED_PUT_ALL_IN_PROGRESS.set(false);
if (logger.isTraceEnabled()) {
logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
"done dispatching operation in view version {}", viewVersion);
@@ -455,7 +470,7 @@ public abstract class DistributedCacheOperation {
if (this instanceof DistributedPutAllOperation) {
// if (logger.isDebugEnabled()) {
logger.warn(
- "XXX DistributedPutAllOperation _distribute recipients for {}: {} with adjunct messages to: {}",
+ "XXX DISTRIBUTED_PUT_ALL_OPERATION _distribute recipients for {}: {} with adjunct messages to: {}",
this, recipients,
adjunctRecipients);
// }
@@ -610,7 +625,7 @@ public abstract class DistributedCacheOperation {
if (failures != null && !failures.isEmpty()) {
logger.warn(
- "XXX DistributedPutAllOperation _distribute Failed sending ({}) to {} while processing event:{}",
+ "XXX DISTRIBUTED_PUT_ALL_OPERATION _distribute Failed sending ({}) to {} while processing event:{}",
msg, failures, event);
}
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 1642a8a..61aabd1 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
@@ -10061,7 +10061,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
*
* @param key the cache key
* @param value the cache value
- * @param putallOp the DistributedPutAllOperation associated with the event
+ * @param putallOp the DISTRIBUTED_PUT_ALL_OPERATION associated with the event
* @param tagHolder holder for version tag
* @throws TimeoutException if the operation times out
* @throws CacheWriterException if a cache writer objects to the update
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 99aa3f0..59e762c 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
@@ -2211,7 +2211,7 @@ public class PartitionedRegion extends LocalRegion
/**
* Create PutAllPRMsgs for each bucket, and send them.
*
- * @param putAllOp DistributedPutAllOperation object.
+ * @param putAllOp DISTRIBUTED_PUT_ALL_OPERATION object.
* @param successfulPuts not used in PartitionedRegion.
*/
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java
index 11e2e2d..5877617 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java
@@ -98,7 +98,7 @@ public class StateFlushOperation {
private DistributionManager dm;
- private static void log(String message) {
+ public static void log(String message) {
// System.out.println(Thread.currentThread().getName() + ": " + message);
logger.warn("XXX " + message);
}
@@ -506,7 +506,8 @@ public class StateFlushOperation {
@Override
public String toString() {
return "StateMarkerMessage(requestingMember=" + this.getSender() + ",processorId="
- + processorId + ",target=" + relayRecipient + ",region=" + regionPath + ")";
+ + processorId + ",target=" + relayRecipient + ",region=" + regionPath + ",flushNewOps="
+ + flushNewOps + ")";
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java
index 9768fb8..ef1a6f2 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java
@@ -1928,7 +1928,7 @@ public class TXState implements TXStateInterface {
*
* @see
* org.apache.geode.internal.cache.InternalDataView#postPutAll(org.apache.geode.internal.cache.
- * DistributedPutAllOperation, java.util.Map, org.apache.geode.internal.cache.LocalRegion)
+ * DISTRIBUTED_PUT_ALL_OPERATION, java.util.Map, org.apache.geode.internal.cache.LocalRegion)
*/
@Override
public void postPutAll(final DistributedPutAllOperation putallOp,
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemotePutAllMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemotePutAllMessage.java
index f4a68b0..09f759a 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemotePutAllMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemotePutAllMessage.java
@@ -68,7 +68,7 @@ import org.apache.geode.internal.offheap.annotations.Released;
/**
* A Replicate Region putAll message. Meant to be sent only to the peer who hosts transactional
* data. It is also used to implement non-transactional putAlls, see:
- * DistributedPutAllOperation.initMessage
+ * DISTRIBUTED_PUT_ALL_OPERATION.initMessage
*
* @since GemFire 6.5
*/