You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by mh...@apache.org on 2021/02/26 00:21:49 UTC
[geode] branch bugfix/StatusReporter updated (5b7ae8c -> 47a9bcf)
This is an automated email from the ASF dual-hosted git repository.
mhanson pushed a change to branch bugfix/StatusReporter
in repository https://gitbox.apache.org/repos/asf/geode.git.
discard 5b7ae8c removing a print stack trace. (+11 squashed commits) Squashed commits: [2b22101819] remove some problem causing logging. [b0cdb82f85] process message debugging [b218425165] process message debugging [4161f89af4] process message debugging [2c9490accd] reply message debugging [f877930405] reply message debugging [490996a7b3] deeper debugging. [e18a539346] Status Report implementation... [f3739e1c68] Status Report implementation... [f4e25743a4] more logging. [27f6f03256] Ad [...]
discard 9db2a1d GEODE-8886: Turning up logging.
new 47a9bcf Adding GatewaySender logging.
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 (5b7ae8c)
\
N -- N -- N refs/heads/bugfix/StatusReporter (47a9bcf)
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:
[geode] 01/01: Adding GatewaySender logging.
Posted by mh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mhanson pushed a commit to branch bugfix/StatusReporter
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 47a9bcf3be7d7f0d70ebeb97b485a7b0ac868bed
Author: Mark Hanson <ha...@vmware.com>
AuthorDate: Thu Jan 28 15:31:14 2021 -0800
Adding GatewaySender logging.
---
.../java/org/apache/geode/ComponentStatus.java | 9 +
.../main/java/org/apache/geode/StatusReporter.java | 99 ++++++++++
.../apache/geode/distributed/ServerLauncher.java | 31 ++-
.../geode/distributed/internal/ReplyMessage.java | 4 +
.../distributed/internal/ReplyProcessor21.java | 21 +-
.../internal/cache/DistributedCacheOperation.java | 82 ++++++--
.../geode/internal/cache/GemFireCacheImpl.java | 90 +++------
.../internal/cache/wan/AbstractGatewaySender.java | 43 ++--
.../wan/AbstractGatewaySenderEventProcessor.java | 62 ++----
.../internal/cache/wan/GatewaySenderAdvisor.java | 89 +++++----
.../wan/GatewaySenderEventCallbackDispatcher.java | 8 +-
...aySenderQueueEntrySynchronizationOperation.java | 34 ++--
...currentParallelGatewaySenderEventProcessor.java | 40 ++--
.../ParallelGatewaySenderEventProcessor.java | 20 +-
.../wan/parallel/ParallelGatewaySenderQueue.java | 216 +++++++--------------
...tilParallelGatewaySenderFlushedCoordinator.java | 36 ++--
...oncurrentSerialGatewaySenderEventProcessor.java | 52 ++---
.../serial/SerialGatewaySenderEventProcessor.java | 95 ++++-----
.../cache/wan/serial/SerialGatewaySenderQueue.java | 86 +++-----
.../wan/serial/SerialSecondaryGatewayListener.java | 6 +-
.../java/org/apache/geode/StatusReporterTest.java | 105 ++++++++++
.../cache/wan/serial/SerialGatewaySenderImpl.java | 27 ++-
22 files changed, 687 insertions(+), 568 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/ComponentStatus.java b/geode-core/src/main/java/org/apache/geode/ComponentStatus.java
new file mode 100644
index 0000000..6018d0a
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/ComponentStatus.java
@@ -0,0 +1,9 @@
+package org.apache.geode;
+
+public interface ComponentStatus {
+ String name();
+
+ String getStatusString();
+
+ void print();
+}
diff --git a/geode-core/src/main/java/org/apache/geode/StatusReporter.java b/geode-core/src/main/java/org/apache/geode/StatusReporter.java
new file mode 100644
index 0000000..4a437fd
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/StatusReporter.java
@@ -0,0 +1,99 @@
+package org.apache.geode;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.logging.internal.log4j.api.LogService;
+
+public class StatusReporter {
+ protected static Logger logger = LogService.getLogger();
+
+ public static boolean isComponentRegistered(ComponentStatus componentStatus) {
+ synchronized (componentStatusSet) {
+ return componentStatusSet.contains(componentStatus);
+ }
+ }
+
+ enum State {
+ SR_RUNNING, SR_NOT_STARTED
+ }
+
+ private static final Set<ComponentStatus> componentStatusSet = new HashSet<>();
+ private static State status = State.SR_NOT_STARTED;
+ private static final Semaphore stopSem = new Semaphore(1);
+
+ static final Thread statusReporterThread = new Thread(() -> {
+ status = State.SR_RUNNING;
+ logger.info("StatusReporter Started thread.");
+ try {
+ stopSem.acquire();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ while (true) {
+ try {
+ if (stopSem.tryAcquire(5000, TimeUnit.MILLISECONDS)) {
+ break;
+ }
+ StatusReporter.getStatusReport();
+ } catch (Exception e) {
+ return;
+ }
+ }
+ logger.info("StatusReporter Stopping thread.");
+ status = State.SR_NOT_STARTED;
+ });
+
+ public static State getStatus() {
+ State tempState;
+ synchronized (statusReporterThread) {
+ if(status != State.SR_RUNNING) {
+ StatusReporter.start();
+ }
+ tempState = status;
+ }
+ return tempState;
+ }
+
+ public static void start() {
+ synchronized (statusReporterThread) {
+ statusReporterThread.start();
+ }
+ }
+
+ public static void stop() throws InterruptedException {
+ synchronized (statusReporterThread) {
+ stopSem.release();
+ }
+ statusReporterThread.join();
+ }
+
+ public static boolean registerComponent(ComponentStatus componentStatus) {
+ synchronized (componentStatusSet) {
+ return componentStatusSet.add(componentStatus);
+ }
+ }
+
+ public static boolean removeComponent(ComponentStatus componentStatus) {
+ synchronized (componentStatusSet) {
+ return componentStatusSet.remove(componentStatus);
+ }
+ }
+
+ public static boolean getStatusReport() {
+ synchronized (statusReporterThread) {
+ if (componentStatusSet.isEmpty()) {
+ return false;
+ }
+ for (ComponentStatus componentStatus : componentStatusSet) {
+ componentStatus.print();
+ }
+ }
+ return true;
+ }
+
+}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/ServerLauncher.java b/geode-core/src/main/java/org/apache/geode/distributed/ServerLauncher.java
index f10b053..b9320c1 100755
--- a/geode-core/src/main/java/org/apache/geode/distributed/ServerLauncher.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/ServerLauncher.java
@@ -37,10 +37,12 @@ import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.ServiceLoader;
+import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
@@ -60,6 +62,8 @@ import joptsimple.OptionSet;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.logging.log4j.Logger;
+import org.apache.geode.ComponentStatus;
+import org.apache.geode.StatusReporter;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
@@ -114,7 +118,7 @@ import org.apache.geode.util.internal.GeodeGlossary;
* @since GemFire 7.0
*/
@SuppressWarnings("unused")
-public class ServerLauncher extends AbstractLauncher<String> {
+public class ServerLauncher extends AbstractLauncher<String> implements ComponentStatus {
private static final Logger log = LogService.getLogger();
@@ -195,7 +199,7 @@ public class ServerLauncher extends AbstractLauncher<String> {
private final ControlNotificationHandler controlHandler;
private final AtomicBoolean starting = new AtomicBoolean(false);
-
+ private final Set<ComponentStatus> subComponentStatus = new HashSet<>();
private final boolean assignBuckets;
private final boolean deletePidFileOnStop;
private final boolean disableDefaultServer;
@@ -832,7 +836,8 @@ public class ServerLauncher extends AbstractLauncher<String> {
}
awaitStartupTasks(cache, startTime);
-
+ StatusReporter.start();
+ StatusReporter.registerComponent(this);
debug("Running Server on (%1$s) in (%2$s) as (%3$s)...", getId(), getWorkingDirectory(),
getMember());
running.set(true);
@@ -1369,6 +1374,26 @@ public class ServerLauncher extends AbstractLauncher<String> {
ProcessType.SERVER, isForcing());
}
+ @Override
+ public String name() {
+ return getId();
+ }
+
+ @Override
+ public String getStatusString() {
+ return isRunning() ? "running" : "not running";
+ }
+
+
+
+ @Override
+ public void print() {
+ logger.info(" Server Name: " + name() + " Component Status: " + getStatusString());
+ for (ComponentStatus componentStatus : subComponentStatus) {
+ componentStatus.print();
+ }
+ }
+
private class ServerControllerParameters implements ProcessControllerParameters {
@Override
public File getPidFile() {
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyMessage.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyMessage.java
index 2f59608..6d387a1 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/ReplyMessage.java
@@ -334,6 +334,10 @@ public class ReplyMessage extends HighPriorityDistributionMessage {
sb.append(ex);
}
}
+ sb.append(" ignored ? ");
+ sb.append(this.ignored);
+ sb.append(" closed ? ");
+ sb.append(this.closed);
return sb;
}
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 83d7884..20421b3 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
@@ -409,29 +409,38 @@ public class ReplyProcessor21 implements MembershipListener {
}
protected void process(DistributionMessage msg, boolean warn) {
- if (logger.isDebugEnabled()) {
- logger.debug("{} got process({}) from {}", this, msg, msg.getSender());
- }
+ logger.info("MLH {} got process({}) from {}", this, msg, msg.getSender());
+ logger.info("MLH Process 1 msg {}", msg);
if (msg instanceof ReplyMessage) {
ReplyException ex = ((ReplyMessage) msg).getException();
+ logger.info("MLH Process 2 msg {} with exception of {}", msg, ex);
+
if (ex != null) {
if (ex.getCause() instanceof DSFIDNotFoundException) {
processException(msg, (DSFIDNotFoundException) ex.getCause());
} else {
processException(msg, ex);
}
+ } else {
+ ReplyMessage replyMessage = (ReplyMessage) msg;
+ logger.info(
+ "MLH Process 3 msg {} ignored ? {} was closed ? {} ",replyMessage,
+ replyMessage.ignored, replyMessage.closed);
}
}
+ logger.info("MLH Process 4");
+
final InternalDistributedMember sender = msg.getSender();
if (!removeMember(sender, false) && warn) {
+ logger.info("MLH Process 5 ");
// if the member hasn't left the system, something is wrong
final DistributionManager dm = getDistributionManager(); // fix for bug 33253
Set ids = getDistributionManagerIds();
if (ids == null || ids.contains(sender)) {
List viewMembers = dm.getViewMembers();
if (system.getConfig().getMcastPort() == 0 // could be using multicast & will get responses
- // from everyone
+ // from everyone
&& (viewMembers == null || viewMembers.contains(sender))) {
logger.warn(
"Received reply from member {} but was not expecting one. More than one reply may have been received. The reply that was not expected is: {}",
@@ -439,6 +448,7 @@ public class ReplyProcessor21 implements MembershipListener {
}
}
}
+ logger.info("MLH Process 6 ");
checkIfDone();
}
@@ -761,7 +771,7 @@ public class ReplyProcessor21 implements MembershipListener {
* @param p_msecs the number of milliseconds to wait for replies, zero will be interpreted as
* Long.MAX_VALUE
*
- * @throws ReplyException an exception passed back in reply
+ * @throws ReplyException an exception passed back in reply
*
* @throws InternalGemFireException if ack-threshold was exceeded and system property
* "ack-threshold-exception" is set to true
@@ -964,6 +974,7 @@ public class ReplyProcessor21 implements MembershipListener {
@Override
public String toString() {
+
return "<" + shortName() + " " + this.getProcessorId() + " waiting for " + numMembers()
+ " replies" + (exception == null ? "" : (" exception: " + exception)) + " from "
+ membersToString() + ">";
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 3e55045..69658c7 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
@@ -331,6 +331,7 @@ public abstract class DistributedCacheOperation {
DistributedRegion region = getRegion();
DistributionManager mgr = region.getDistributionManager();
boolean reliableOp = isOperationReliable() && region.requiresReliabilityCheck();
+ boolean isDebugEnabled = logger.isDebugEnabled();
if (SLOW_DISTRIBUTION_MS > 0) { // test hook
try {
@@ -365,7 +366,7 @@ public abstract class DistributedCacheOperation {
routingComputed = true;
filterRouting = getRecipientFilterRouting(recipients);
if (filterRouting != null) {
- if (logger.isDebugEnabled()) {
+ if (isDebugEnabled) {
logger.debug("Computed this filter routing: {}", filterRouting);
}
}
@@ -472,7 +473,7 @@ public abstract class DistributedCacheOperation {
}
}
- if (logger.isDebugEnabled()) {
+ if (isDebugEnabled) {
logger.debug("recipients for {}: {} with adjunct messages to: {}", this, recipients,
adjunctRecipients);
}
@@ -485,10 +486,20 @@ public abstract class DistributedCacheOperation {
// common
// case
waitForMembers = recipients;
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 1 this : " + this + " recipients " + recipients);
+ }
+
} else if (!cachelessNodes.isEmpty()) {
waitForMembers = new HashSet(recipients);
waitForMembers.addAll(cachelessNodes);
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 2 this : " + this + " recipients " + recipients);
+ }
} else {
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 3 this : " + this + " recipients " + recipients);
+ }
// note that we use a Vector instead of a Set for the responders
// collection
// because partitioned regions sometimes send both a regular cache
@@ -504,7 +515,7 @@ public abstract class DistributedCacheOperation {
LOSS_SIMULATION_GENERATOR = new Random(this.hashCode());
}
if ((LOSS_SIMULATION_GENERATOR.nextInt(100) * 1.0 / 100.0) < LOSS_SIMULATION_RATIO) {
- if (logger.isDebugEnabled()) {
+ if (isDebugEnabled) {
logger.debug("loss simulation is inhibiting message transmission to {}",
recipients);
}
@@ -513,6 +524,9 @@ public abstract class DistributedCacheOperation {
}
}
if (reliableOp) {
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 4 reliableOp : " + reliableOp);
+ }
this.departedMembers = new HashSet();
this.processor = new ReliableCacheReplyProcessor(region.getSystem(), waitForMembers,
this.departedMembers);
@@ -520,7 +534,9 @@ public abstract class DistributedCacheOperation {
this.processor = new CacheOperationReplyProcessor(region.getSystem(), waitForMembers);
}
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 5 ");
+ }
Set failures = null;
CacheOperationMessage msg = createMessage();
initMessage(msg, this.processor);
@@ -528,7 +544,9 @@ public abstract class DistributedCacheOperation {
if (DistributedCacheOperation.internalBeforePutOutgoing != null) {
DistributedCacheOperation.internalBeforePutOutgoing.run();
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 6 ");
+ }
if (processor != null && msg.isSevereAlertCompatible()) {
this.processor.enableSevereAlertProcessing();
// if this message is distributing for a partitioned region message,
@@ -545,7 +563,9 @@ public abstract class DistributedCacheOperation {
}
}
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 7 ");
+ }
msg.setMulticast(useMulticast);
msg.directAck = directAck;
if (region.isUsedForPartitionedRegionBucket()) {
@@ -559,7 +579,9 @@ public abstract class DistributedCacheOperation {
} else if (!routingComputed) {
msg.needsRouting = true;
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 8 ");
+ }
initProcessor(processor, msg);
if (region.cache.isClosed() && !canBeSentDuringShutdown()) {
@@ -567,7 +589,9 @@ public abstract class DistributedCacheOperation {
"The cache has been closed",
null);
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 9 ");
+ }
msg.setRecipients(recipients);
failures = mgr.putOutgoing(msg);
@@ -578,7 +602,7 @@ public abstract class DistributedCacheOperation {
msg.setRecipients(needsOldValueInCacheOp);
Set newFailures = mgr.putOutgoing(msg);
if (newFailures != null) {
- if (logger.isDebugEnabled()) {
+ if (isDebugEnabled) {
logger.debug("Failed sending ({}) to {}", msg, newFailures);
}
if (failures != null && failures.size() > 0) {
@@ -588,7 +612,9 @@ public abstract class DistributedCacheOperation {
}
}
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 10 ");
+ }
if (cachelessNodes.size() > 0) {
cachelessNodes.removeAll(cachelessNodesWithNoCacheServer);
if (cachelessNodes.size() > 0) {
@@ -622,8 +648,10 @@ public abstract class DistributedCacheOperation {
cachelessNodes.addAll(cachelessNodesWithNoCacheServer);
}
}
-
- if (failures != null && !failures.isEmpty() && logger.isDebugEnabled()) {
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 11 ");
+ }
+ if (failures != null && !failures.isEmpty() && isDebugEnabled) {
logger.debug("Failed sending ({}) to {} while processing event:{}", msg, failures, event);
}
@@ -666,7 +694,9 @@ public abstract class DistributedCacheOperation {
}
}
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 12 ");
+ }
// compute local client routing before waiting for an ack only for a bucket
if (region.isUsedForPartitionedRegionBucket()) {
FilterInfo filterInfo = getLocalFilterRouting(filterRouting);
@@ -674,7 +704,9 @@ public abstract class DistributedCacheOperation {
}
waitForAckIfNeeded(msg, persistentIds);
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 13 ");
+ }
if (/* msg != null && */reliableOp) {
Set successfulRecips = new HashSet(recipients);
successfulRecips.addAll(cachelessNodes);
@@ -688,13 +720,15 @@ public abstract class DistributedCacheOperation {
region.handleReliableDistribution(successfulRecips);
}
}
-
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 14 ");
+ }
if (region.isUsedForPartitionedRegionBucket() && filterRouting != null) {
removeDestroyTokensFromCqResultKeys(filterRouting);
}
} catch (CancelException e) {
- if (logger.isDebugEnabled()) {
+ if (isDebugEnabled) {
logger.debug("distribution of message aborted by shutdown: {}", this);
}
throw e;
@@ -703,6 +737,9 @@ public abstract class DistributedCacheOperation {
e);
throw e;
} finally {
+ if (isDebugEnabled) {
+ logger.debug("MLH _distribute 15 ");
+ }
ReplyProcessor21.setShortSevereAlertProcessing(false);
}
}
@@ -1605,19 +1642,30 @@ public abstract class DistributedCacheOperation {
@Override
protected void process(final DistributionMessage dmsg, boolean warn) {
+ logger.info("MLH CacheOperationReplyProcessor.process 1 dmsg " + dmsg + " msg = " + msg);
if (dmsg instanceof ReplyMessage) {
+ logger.info("MLH CacheOperationReplyProcessor.process 2");
+
ReplyMessage replyMessage = (ReplyMessage) dmsg;
if (msg != null) {
+ logger.info("MLH CacheOperationReplyProcessor.process 3 replyMessage " + replyMessage);
+
boolean discard = !msg.processReply(replyMessage, this);
+ logger.info("MLH CacheOperationReplyProcessor.process 4 discard = " + discard);
if (discard) {
+ logger.info("MLH CacheOperationReplyProcessor.process 5 discarding replyMessage = " + replyMessage);
return;
}
}
+ logger.info("MLH CacheOperationReplyProcessor.process 5 ");
+
if (replyMessage.getClosed()) {
+ logger.info("MLH CacheOperationReplyProcessor.process 6 ");
+
closedMembers.add(replyMessage.getSender());
}
}
-
+ logger.info("MLH CacheOperationReplyProcessor.process 7 invoking super " + super.getClass().getName());
super.process(dmsg, warn);
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
index 1ec0ed2..ca27cdf 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
@@ -2166,7 +2166,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
if (!keepDS && systemFailureCause == null
&& (isReconnecting() || system.getReconnectedSystem() != null)) {
- logger.debug(
+ logger.info(
"Cache is shutting down distributed system connection. isReconnecting={} reconnectedSystem={} keepAlive={} keepDS={}",
isReconnecting(), system.getReconnectedSystem(), keepAlive, keepDS);
@@ -2206,28 +2206,34 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
this.keepAlive = keepAlive;
isClosing = true;
- logger.info("{}: Now closing.", this);
+ logger.info("MLH {}: Now closing.", this);
// we don't clear the prID map if there is a system failure. Other
// threads may be hung trying to communicate with the map locked
if (systemFailureCause == null) {
+ logger.info("Clearing the PartitionedRegion.clearPRIdMap");
PartitionedRegion.clearPRIdMap();
}
TXStateProxy tx = null;
try {
if (transactionManager != null) {
+ logger.info("Pausing transactions in transactionManager");
tx = transactionManager.pauseTransaction();
}
// do this before closing regions
+ logger.info("MLH Stopping the resourceManager");
+
resourceManager.close();
try {
+ logger.info("MLH Stopping the resourceAdvisor advisor");
resourceAdvisor.close();
} catch (CancelException ignore) {
}
try {
+ logger.info("MLH Stopping the jmxAdvisor advisor");
jmxAdvisor.close();
} catch (CancelException ignore) {
}
@@ -2237,21 +2243,17 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
sender.stop();
GatewaySenderAdvisor advisor = ((AbstractGatewaySender) sender).getSenderAdvisor();
if (advisor != null) {
- if (isDebugEnabled) {
- logger.debug("Stopping the GatewaySender advisor");
- }
+ logger.info("Stopping the GatewaySender advisor");
advisor.close();
}
} catch (CancelException ignore) {
}
}
-
+ logger.info("MLH destroyGatewaySenderLockService");
destroyGatewaySenderLockService();
if (eventThreadPool != null) {
- if (isDebugEnabled) {
- logger.debug("{}: stopping event thread pool...", this);
- }
+ logger.info("{}: stopping event thread pool...", this);
eventThreadPool.shutdown();
}
@@ -2265,9 +2267,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
stopServices();
// no need to track PR instances
- if (isDebugEnabled) {
- logger.debug("{}: clearing partitioned regions...", this);
- }
+ logger.info("{}: clearing partitioned regions...", this);
synchronized (partitionedRegions) {
int prSize = -partitionedRegions.size();
partitionedRegions.clear();
@@ -2288,9 +2288,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
InternalRegion prRoot = null;
for (InternalRegion lr : rootRegions.values()) {
- if (isDebugEnabled) {
- logger.debug("{}: processing region {}", this, lr.getFullPath());
- }
+ logger.info("{}: processing region {}", this, lr.getFullPath());
if (PartitionedRegionHelper.PR_ROOT_REGION_NAME.equals(lr.getName())) {
prRoot = lr;
} else {
@@ -2298,9 +2296,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
// this region will be closed internally by parent region
continue;
}
- if (isDebugEnabled) {
- logger.debug("{}: closing region {}...", this, lr.getFullPath());
- }
+ logger.info("{}: closing region {}...", this, lr.getFullPath());
try {
lr.handleCacheClose(op);
} catch (RuntimeException e) {
@@ -2314,9 +2310,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
}
try {
- if (isDebugEnabled) {
- logger.debug("{}: finishing partitioned region close...", this);
- }
+ logger.info("{}: finishing partitioned region close...", this);
PartitionedRegion.afterRegionsClosedByCacheClose(this);
if (prRoot != null) {
// do the PR meta root region last
@@ -2334,9 +2328,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
// Close the CqService Handle.
try {
- if (isDebugEnabled) {
- logger.debug("{}: closing CQ service...", this);
- }
+ logger.info("{}: closing CQ service...", this);
cqService.close();
} catch (RuntimeException ignore) {
logger.info("Failed to get the CqService, to close during cache close (1).");
@@ -2344,20 +2336,16 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
PoolManager.close(keepAlive);
- if (isDebugEnabled) {
- logger.debug("{}: notifying admins of close...", this);
- }
+ logger.info("{}: notifying admins of close...", this);
try {
SystemMemberCacheEventProcessor.send(this, Operation.CACHE_CLOSE);
} catch (CancelException ignore) {
if (logger.isDebugEnabled()) {
- logger.debug("Ignored cancellation while notifying admins");
+ logger.info("Ignored cancellation while notifying admins");
}
}
- if (isDebugEnabled) {
- logger.debug("{}: stopping destroyed entries processor...", this);
- }
+ logger.info("{}: stopping destroyed entries processor...", this);
tombstoneService.stop();
// NOTICE: the CloseCache message is the *last* message you can send!
@@ -2370,9 +2358,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
if (distributionManager != null) {
// Send CacheClosedMessage (and NOTHING ELSE) here
- if (isDebugEnabled) {
- logger.debug("{}: sending CloseCache to peers...", this);
- }
+ logger.info("{}: sending CloseCache to peers...", this);
Set<InternalDistributedMember> otherMembers =
distributionManager.getOtherDistributionManagerIds();
ReplyProcessor21 processor = replyProcessor21Factory.create(system, otherMembers);
@@ -2658,22 +2644,16 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
private void stopServers() {
boolean isDebugEnabled = logger.isDebugEnabled();
- if (isDebugEnabled) {
- logger.debug("{}: stopping cache servers...", this);
- }
+ logger.info("{}: stopping cache servers...", this);
boolean stoppedCacheServer = false;
for (InternalCacheServer cacheServer : allCacheServers) {
- if (isDebugEnabled) {
- logger.debug("stopping bridge {}", cacheServer);
- }
+ logger.info("stopping bridge {}", cacheServer);
try {
cacheServer.stop();
} catch (CancelException e) {
- if (isDebugEnabled) {
- logger.debug("Ignored cache closure while closing bridge {}", cacheServer, e);
- }
+ logger.info("Ignored cache closure while closing bridge {}", cacheServer, e);
}
allCacheServers.remove(cacheServer);
stoppedCacheServer = true;
@@ -2681,16 +2661,12 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
InternalCacheServer receiverServer = gatewayReceiverServer.getAndSet(null);
if (receiverServer != null) {
- if (isDebugEnabled) {
- logger.debug("stopping gateway receiver server {}", receiverServer);
- }
+ logger.info("stopping gateway receiver server {}", receiverServer);
try {
receiverServer.stop();
} catch (CancelException e) {
- if (isDebugEnabled) {
- logger.debug("Ignored cache closure while closing gateway receiver server {}",
+ logger.info("Ignored cache closure while closing gateway receiver server {}",
receiverServer, e);
- }
}
stoppedCacheServer = true;
}
@@ -2702,26 +2678,18 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
}
// stop HA services if they had been started
- if (isDebugEnabled) {
- logger.debug("{}: stopping HA services...", this);
- }
+ logger.info("{}: stopping HA services...", this);
try {
HARegionQueue.stopHAServices();
} catch (CancelException e) {
- if (isDebugEnabled) {
- logger.debug("Ignored cache closure while closing HA services", e);
- }
+ logger.info("Ignored cache closure while closing HA services", e);
}
- if (isDebugEnabled) {
- logger.debug("{}: stopping client health monitor...", this);
- }
+ logger.info("{}: stopping client health monitor...", this);
try {
ClientHealthMonitor.shutdownInstance();
} catch (CancelException e) {
- if (isDebugEnabled) {
- logger.debug("Ignored cache closure while closing client health monitor", e);
- }
+ logger.info("Ignored cache closure while closing client health monitor", e);
}
// Reset the unique id counter for durable clients. If a durable client stops/starts its cache,
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 4ea2c6d..959d2f6 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
@@ -613,9 +613,7 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
// close the GatewaySenderAdvisor
GatewaySenderAdvisor advisor = this.getSenderAdvisor();
if (advisor != null) {
- if (logger.isDebugEnabled()) {
- logger.debug("Stopping the GatewaySender advisor");
- }
+ logger.debug("Stopping the GatewaySender advisor");
advisor.close();
}
@@ -1036,14 +1034,22 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
seca.getRecipientDSIds().addAll(allRemoteDSIds);
}
} else {
+
GatewaySenderEventCallbackArgument geCallbackArg =
new GatewaySenderEventCallbackArgument(callbackArg, this.getMyDSId(), allRemoteDSIds);
clonedEvent.setCallbackArgument(geCallbackArg);
+
+ if (isDebugEnabled) {
+ logger.debug("MLH distribute: 1 creating object GatewaySenderEventCallbackArgument:" + clonedEvent);
+ }
}
// If this gateway is not running, return
if (!isRunning()) {
if (this.isPrimary()) {
+ if (isDebugEnabled) {
+ logger.debug("MLH distribute: 2 recordDroppedEvent:" + clonedEvent);
+ }
recordDroppedEvent(clonedEvent);
}
if (isDebugEnabled) {
@@ -1079,6 +1085,9 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
logger.debug("Returning back without putting into the gateway sender queue:" + event);
}
if (this.isPrimary()) {
+ if (isDebugEnabled) {
+ logger.debug("MLH distribute: 3 recordDroppedEvent:" + clonedEvent);
+ }
recordDroppedEvent(clonedEvent);
}
return;
@@ -1099,7 +1108,9 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
// Get substitution value to enqueue if necessary
Object substituteValue = getSubstituteValue(clonedEvent, operation);
-
+ if (isDebugEnabled) {
+ logger.debug("MLH distribute: 4 enqueueEvent:" + clonedEvent);
+ }
ev.enqueueEvent(operation, clonedEvent, substituteValue, isLastEventInTransaction);
} catch (CancelException e) {
logger.debug("caught cancel exception", e);
@@ -1128,11 +1139,11 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
private void recordDroppedEvent(EntryEventImpl event) {
if (this.eventProcessor != null) {
this.eventProcessor.registerEventDroppedInPrimaryQueue(event);
+ logger.debug("registerEventDroppedInPrimaryQueue event: {}", event);
+
} else {
tmpDroppedEvents.add(event);
- if (logger.isDebugEnabled()) {
- logger.debug("added to tmpDroppedEvents event: {}", event);
- }
+ logger.debug("added to tmpDroppedEvents event: {}", event);
}
}
@@ -1169,10 +1180,8 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
synchronized (this.queuedEventsSync) {
while ((nextEvent = tmpQueuedEvents.poll()) != null) {
try {
- if (logger.isDebugEnabled()) {
- logger.debug("Event :{} is enqueued to GatewaySenderQueue from TempQueue",
- nextEvent);
- }
+ logger.debug("Event :{} is enqueued to GatewaySenderQueue from TempQueue",
+ nextEvent);
stats.decTempQueueSize();
this.eventProcessor.enqueueEvent(nextEvent.getOperation(), nextEvent.getEvent(),
nextEvent.getSubstituteValue());
@@ -1204,11 +1213,9 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
while (itr.hasNext()) {
TmpQueueEvent event = itr.next();
if (tailKey.equals(event.getEvent().getTailKey())) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "shadowKey {} is found in tmpQueueEvents at AbstractGatewaySender level. Removing from there..",
- tailKey);
- }
+ logger.debug(
+ "shadowKey {} is found in tmpQueueEvents at AbstractGatewaySender level. Removing from there..",
+ tailKey);
event.release();
itr.remove();
return true;
@@ -1304,9 +1311,7 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
// Store the index locally
this.eventIdIndex = index;
- if (logger.isDebugEnabled()) {
- logger.debug("{}: {} event id index: {}", this, messagePrefix, this.eventIdIndex);
- }
+ logger.debug("{}: {} event id index: {}", this, messagePrefix, this.eventIdIndex);
}
} finally {
// Unlock the lock if necessary
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 294121a..a65e680 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
@@ -339,9 +339,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
this.isPaused = false;
// Notify thread to resume
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Resumed dispatching", this);
- }
+ logger.debug("{}: Resumed dispatching", this);
synchronized (this.pausedLock) {
this.pausedLock.notifyAll();
}
@@ -726,9 +724,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
}
}
} // unsuccessful batch
- if (logger.isDebugEnabled()) {
- logger.debug("Finished processing events (batch #{})", (getBatchId() - 1));
- }
+ logger.debug("Finished processing events (batch #{})", (getBatchId() - 1));
} finally {
afterExecute();
}
@@ -739,9 +735,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
this.resetLastPeekedEvents = true;
// most possible case is ParallelWan when user PR is locally destroyed
// shadow PR is also locally destroyed
- if (logger.isDebugEnabled()) {
- logger.debug("Observed RegionDestroyedException on Queue's region.");
- }
+ logger.debug("Observed RegionDestroyedException on Queue's region.");
} catch (CancelException e) {
logger.debug("Caught cancel exception", e);
setIsStopped(true);
@@ -791,9 +785,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
List<GatewaySenderEventImpl> conflatedEvents = null;
// Conflate the batch if necessary
if (this.sender.isBatchConflationEnabled() && events.size() > 1) {
- if (logger.isDebugEnabled()) {
- logEvents("original", events);
- }
+ logEvents("original", events);
Map<ConflationKey, GatewaySenderEventImpl> conflatedEventsMap =
new LinkedHashMap<ConflationKey, GatewaySenderEventImpl>();
conflatedEvents = new ArrayList<GatewaySenderEventImpl>();
@@ -832,9 +824,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
// Increment the events conflated from batches statistic
this.sender.getStatistics()
.incEventsConflatedFromBatches(events.size() - conflatedEvents.size());
- if (logger.isDebugEnabled()) {
- logEvents("conflated", conflatedEvents);
- }
+ logEvents("conflated", conflatedEvents);
} else {
conflatedEvents = events;
}
@@ -1043,9 +1033,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
}
}
List<GatewaySenderEventImpl> events = eventsArr[0];
- if (logger.isDebugEnabled()) {
- logger.debug("Removing events from the queue {}", events.size());
- }
+ logger.debug("Removing events from the queue {}", events.size());
eventQueueRemove(events.size());
logThresholdExceededAlerts(events);
@@ -1097,9 +1085,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
if (!this.isPaused) {
return;
}
- if (logger.isDebugEnabled()) {
- logger.debug("GatewaySenderEventProcessor is paused. Waiting for Resumption");
- }
+ logger.debug("GatewaySenderEventProcessor is paused. Waiting for Resumption");
this.isDispatcherWaiting = true;
this.pausedLock.notifyAll();
while (this.isPaused) {
@@ -1191,9 +1177,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
}
resumeDispatching();
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Notifying the dispatcher to terminate", this);
- }
+ logger.debug("{}: Notifying the dispatcher to terminate", this);
// If this is the primary, stay alive for a predefined time
// OR until the queue becomes empty
@@ -1202,9 +1186,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
try {
while (!(this.queue.size() == 0)) {
Thread.sleep(5000);
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Waiting for the queue to get empty.", this);
- }
+ logger.debug("{}: Waiting for the queue to get empty.", this);
}
} catch (InterruptedException e) {
// interrupted
@@ -1219,16 +1201,16 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
}
}
} else {
+ logger.debug("{}: notifyPrimaryLock.", this);
this.sender.getSenderAdvisor().notifyPrimaryLock();
}
+ logger.debug("{}: setIsStopped to true.", this);
setIsStopped(true);
dispatcher.stop();
if (this.isAlive()) {
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Joining with the dispatcher thread upto limit of 5 seconds", this);
- }
+ logger.debug("{}: Joining with the dispatcher thread upto limit of 5 seconds", this);
try {
this.join(5000); // wait for our thread to stop
if (this.isAlive()) {
@@ -1250,15 +1232,11 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
closeProcessor();
- if (logger.isDebugEnabled()) {
- logger.debug("Stopped dispatching: {}", this);
- }
+ logger.debug("Stopped dispatching: {}", this);
}
public void closeProcessor() {
- if (logger.isDebugEnabled()) {
- logger.debug("Closing dispatcher");
- }
+ logger.debug("Closing dispatcher");
try {
if (this.sender.isPrimary() && this.queue.size() > 0) {
logger.warn("Destroying GatewayEventDispatcher with actively queued data.");
@@ -1273,16 +1251,12 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
// just checking in case we should log a warning
} finally {
this.queue.close();
- if (logger.isDebugEnabled()) {
- logger.debug("Closed dispatcher");
- }
+ logger.debug("Closed dispatcher");
}
}
protected void destroyProcessor() {
- if (logger.isDebugEnabled()) {
- logger.debug("Destroying dispatcher");
- }
+ logger.debug("Destroying dispatcher");
try {
try {
if (this.queue.peek() != null) {
@@ -1299,9 +1273,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends LoggingThread
// just checking in case we should log a warning
} finally {
this.queue.getRegion().localDestroyRegion();
- if (logger.isDebugEnabled()) {
- logger.debug("Destroyed dispatcher");
- }
+ logger.debug("Destroyed dispatcher");
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAdvisor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAdvisor.java
index 6af0866..088998f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAdvisor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderAdvisor.java
@@ -27,7 +27,9 @@ import java.util.TreeSet;
import org.apache.logging.log4j.Logger;
import org.apache.geode.CancelException;
+import org.apache.geode.ComponentStatus;
import org.apache.geode.DataSerializer;
+import org.apache.geode.StatusReporter;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.util.Gateway;
@@ -52,7 +54,7 @@ import org.apache.geode.internal.serialization.StaticSerialization;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
-public class GatewaySenderAdvisor extends DistributionAdvisor {
+public class GatewaySenderAdvisor extends DistributionAdvisor implements ComponentStatus {
private static final Logger logger = LogService.getLogger();
private DistributedLockService lockService;
@@ -75,6 +77,8 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
public static GatewaySenderAdvisor createGatewaySenderAdvisor(DistributionAdvisee sender) {
GatewaySenderAdvisor advisor = new GatewaySenderAdvisor(sender);
+ StatusReporter.getStatus();
+ StatusReporter.registerComponent(advisor);
advisor.initialize();
return advisor;
}
@@ -252,6 +256,8 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
@Override
public void profileUpdated(Profile profile) {
if (profile instanceof GatewaySenderProfile) {
+ logger.info("MLH profileUpdated " + profile.getDistributedMember().getName());
+
GatewaySenderProfile sp = (GatewaySenderProfile) profile;
if (!sp.isParallel) { // SerialGatewaySender
if (!sp.isRunning) {
@@ -262,11 +268,9 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
if (!this.sender.isPrimary()) {
if (!adviseEldestGatewaySender()) {// AND this is not the eldest
// sender
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Sender {} is not the eldest in the system. Giving preference to eldest sender to become primary...",
- this.sender);
- }
+ logger.debug(
+ "Sender {} is not the eldest in the system. Giving preference to eldest sender to become primary...",
+ this.sender);
return;
}
launchLockObtainingVolunteerThread();
@@ -287,6 +291,8 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
@Override
protected void profileRemoved(Profile profile) {
if (profile instanceof GatewaySenderProfile) {
+ logger.info("MLH profileRemoved " + profile.getDistributedMember().getName());
+
GatewaySenderProfile sp = (GatewaySenderProfile) profile;
if (!sp.isParallel) {// SerialGatewaySender
// if there is a primary sender, then don't volunteer for primary
@@ -295,11 +301,9 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
}
if (!this.sender.isPrimary()) {// IF this sender is not primary
if (!adviseEldestGatewaySender()) {// AND this is not the eldest sender
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Sender {} is not the eldest in the system. Giving preference to eldest sender to become primary...",
- this.sender);
- }
+ logger.debug(
+ "Sender {} is not the eldest in the system. Giving preference to eldest sender to become primary...",
+ this.sender);
return;
}
launchLockObtainingVolunteerThread();
@@ -320,29 +324,21 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
this.lockService = DLockService.create(dlsName, ds, true, true, true);
}
Assert.assertTrue(this.lockService != null);
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Obtained DistributedLockService: {}", this, this.lockService);
- }
+ logger.debug("{}: Obtained DistributedLockService: {}", this, this.lockService);
}
public boolean volunteerForPrimary() {
- if (logger.isDebugEnabled()) {
- logger.debug("Sender : {} is volunteering for Primary ", this.sender.getId());
- }
+ logger.debug("Sender : {} is volunteering for Primary ", this.sender.getId());
if (advisePrimaryGatewaySender() == null) {
if (!adviseEldestGatewaySender()) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Sender {} is not the eldest in the system. Giving preference to eldest sender to become primary...",
- this.sender);
- }
- return false;
- }
- if (logger.isDebugEnabled()) {
- logger.debug("Sender : {} no Primary available. So going to acquire distributed lock",
+ logger.debug(
+ "Sender {} is not the eldest in the system. Giving preference to eldest sender to become primary...",
this.sender);
+ return false;
}
+ logger.debug("Sender : {} no Primary available. So going to acquire distributed lock",
+ this.sender);
this.lockService.lock(this.lockToken, 10000, -1);
return this.lockService.isHeldByCurrentThread(this.lockToken);
}
@@ -404,13 +400,13 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
}
public void makePrimary() {
- logger.info("{} : Starting as primary", this.sender);
+ logger.info("{} : Starting as primary", this.sender.id);
AbstractGatewaySenderEventProcessor eventProcessor = this.sender.getEventProcessor();
if (eventProcessor != null) {
eventProcessor.removeCacheListener();
}
- logger.info("{} : Becoming primary gateway sender", this.sender);
+ logger.info("{} : Becoming primary gateway sender", this.sender.id);
notifyAndBecomePrimary();
new UpdateAttributesProcessor(this.sender).distribute(false);
}
@@ -429,10 +425,8 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
}
public void makeSecondary() {
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Did not obtain the lock on {}. Starting as secondary gateway sender.",
- this.sender, this.lockToken);
- }
+ logger.debug("{}: Did not obtain the lock on {}. Starting as secondary gateway sender.",
+ this.sender, this.lockToken);
// Set primary flag to false
logger.info(
@@ -451,15 +445,11 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
if (!(GatewaySenderAdvisor.this.sender.isRunning())) {
return;
}
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Obtaining the lock on {}", this, GatewaySenderAdvisor.this.lockToken);
- }
+ logger.debug("{}: Obtaining the lock on {}", this, GatewaySenderAdvisor.this.lockToken);
if (volunteerForPrimary()) {
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Obtained the lock on {}", this,
- GatewaySenderAdvisor.this.lockToken);
- }
+ logger.debug("{}: Obtained the lock on {}", this,
+ GatewaySenderAdvisor.this.lockToken);
logger.info("{} is becoming primary gateway Sender.",
GatewaySenderAdvisor.this);
@@ -484,6 +474,8 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
public void waitToBecomePrimary(AbstractGatewaySenderEventProcessor callingProcessor)
throws InterruptedException {
+ logger.info("MLH {} : waitToBecomePrimary", this.sender.getId());
+
if (isPrimary()) {
return;
}
@@ -499,6 +491,24 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
}
}
+ @Override
+ public String name() {
+ return "Gateway Sender Advisor";
+ }
+
+ @Override
+ public String getStatusString() {
+ return "Running";
+ }
+
+ @Override
+ public void print() {
+ logger.info(" Server Name: " + name() + " Component Status: " + getStatusString()
+ + " isPrimary: " + isPrimary() + " Advisee: " + getAdvisee() + " Sender: " + sender
+ + " Profiles: " + profiles);
+
+ }
+
/**
* Profile information for a remote counterpart.
*/
@@ -758,6 +768,7 @@ public class GatewaySenderAdvisor extends DistributionAdvisor {
@Override
public void close() {
+ logger.info("MLH close Removing profile " + this.sender.locName);
new UpdateAttributesProcessor(this.getAdvisee(), true).distribute(false);
super.close();
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventCallbackDispatcher.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventCallbackDispatcher.java
index e5683e0..589a7ff 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventCallbackDispatcher.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventCallbackDispatcher.java
@@ -70,16 +70,12 @@ public class GatewaySenderEventCallbackDispatcher implements GatewaySenderEventD
GatewaySenderStats statistics = this.eventProcessor.sender.getStatistics();
boolean success = false;
try {
- if (logger.isDebugEnabled()) {
- logger.debug("About to dispatch batch");
- }
+ logger.debug("About to dispatch batch");
long start = statistics.startTime();
// Send the batch to the corresponding GatewaySender
success = dispatchBatch(events);
statistics.endBatch(start, events.size());
- if (logger.isDebugEnabled()) {
- logger.debug("Done dispatching the batch");
- }
+ logger.debug("Done dispatching the batch");
} catch (GatewaySenderException e) {
// Do nothing in this case. The exception has already been logged.
} catch (CancelException e) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderQueueEntrySynchronizationOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderQueueEntrySynchronizationOperation.java
index 00a1196..287cdc7 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderQueueEntrySynchronizationOperation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderQueueEntrySynchronizationOperation.java
@@ -68,12 +68,10 @@ public class GatewaySenderQueueEntrySynchronizationOperation {
}
protected void synchronizeEntries() {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: Requesting synchronization from member={}; regionPath={}; entriesToSynchronize={}",
- getClass().getSimpleName(), this.recipient, this.region.getFullPath(),
- this.entriesToSynchronize);
- }
+ logger.debug(
+ "{}: Requesting synchronization from member={}; regionPath={}; entriesToSynchronize={}",
+ getClass().getSimpleName(), this.recipient, this.region.getFullPath(),
+ this.entriesToSynchronize);
// Create and send message
DistributionManager dm = this.region.getDistributionManager();
GatewaySenderQueueEntrySynchronizationReplyProcessor processor =
@@ -125,13 +123,11 @@ public class GatewaySenderQueueEntrySynchronizationOperation {
if (msg instanceof ReplyMessage) {
ReplyMessage reply = (ReplyMessage) msg;
if (reply.getException() == null) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: Processing reply from member={}; regionPath={}; key={}; entriesToSynchronize={}",
- getClass().getSimpleName(), reply.getSender(),
- this.operation.region.getFullPath(), this.operation.entriesToSynchronize,
- reply.getReturnValue());
- }
+ logger.debug(
+ "{}: Processing reply from member={}; regionPath={}; key={}; entriesToSynchronize={}",
+ getClass().getSimpleName(), reply.getSender(),
+ this.operation.region.getFullPath(), this.operation.entriesToSynchronize,
+ reply.getReturnValue());
List<Map<String, GatewayQueueEvent>> events =
(List<Map<String, GatewayQueueEvent>>) reply.getReturnValue();
for (int i = 0; i < events.size(); i++) {
@@ -194,10 +190,8 @@ public class GatewaySenderQueueEntrySynchronizationOperation {
Object result = null;
ReplyException replyException = null;
try {
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Providing synchronization region={}; entriesToSynchronize={}",
- getClass().getSimpleName(), this.regionPath, this.entriesToSynchronize);
- }
+ logger.debug("{}: Providing synchronization region={}; entriesToSynchronize={}",
+ getClass().getSimpleName(), this.regionPath, this.entriesToSynchronize);
result = getSynchronizationEvents(dm.getCache());
} catch (Throwable t) {
replyException = new ReplyException(t);
@@ -210,10 +204,8 @@ public class GatewaySenderQueueEntrySynchronizationOperation {
} else {
replyMsg.setException(replyException);
}
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Sending synchronization reply returnValue={}; exception={}",
- getClass().getSimpleName(), replyMsg.getReturnValue(), replyMsg.getException());
- }
+ logger.debug("{}: Sending synchronization reply returnValue={}; exception={}",
+ getClass().getSimpleName(), replyMsg.getReturnValue(), replyMsg.getException());
dm.putOutgoing(replyMsg);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java
index 0ef1fd3..1b18258 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java
@@ -98,9 +98,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor
targetRs.add(pr);
}
}
- if (logger.isDebugEnabled()) {
- logger.debug("The target PRs are {} Dispatchers: {}", targetRs, nDispatcher);
- }
+ logger.debug("The target PRs are {} Dispatchers: {}", targetRs, nDispatcher);
createProcessors(sender.getDispatcherThreads(), targetRs, cleanQueues);
@@ -111,9 +109,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor
protected void createProcessors(int dispatcherThreads, Set<Region> targetRs,
boolean cleanQueues) {
processors = new ParallelGatewaySenderEventProcessor[sender.getDispatcherThreads()];
- if (logger.isDebugEnabled()) {
- logger.debug("Creating AsyncEventProcessor");
- }
+ logger.debug("Creating AsyncEventProcessor");
for (int i = 0; i < sender.getDispatcherThreads(); i++) {
processors[i] = new ParallelGatewaySenderEventProcessor(sender, targetRs, i,
sender.getDispatcherThreads(), getThreadMonitorObj(), cleanQueues);
@@ -158,10 +154,8 @@ public class ConcurrentParallelGatewaySenderEventProcessor
ConcurrentParallelGatewaySenderQueue cpgsq = (ConcurrentParallelGatewaySenderQueue) queue;
PartitionedRegion prQ = cpgsq.getRegion(droppedEvent.getRegion().getFullPath());
if (prQ == null) {
- if (logger.isDebugEnabled()) {
- logger.debug("shadow partitioned region " + droppedEvent.getRegion().getFullPath()
- + " is not created yet.");
- }
+ logger.debug("shadow partitioned region " + droppedEvent.getRegion().getFullPath()
+ + " is not created yet.");
return;
}
int bucketId = PartitionedRegionHelper.getHashKey((EntryOperation) droppedEvent);
@@ -172,10 +166,8 @@ public class ConcurrentParallelGatewaySenderEventProcessor
if (isPrimary) {
pgsq.sendQueueRemovalMesssageForDroppedEvent(prQ, bucketId, shadowKey);
this.sender.getStatistics().incEventsDroppedDueToPrimarySenderNotRunning();
- if (logger.isDebugEnabled()) {
- logger.debug("register dropped event for primary queue. BucketId is " + bucketId
- + ", shadowKey is " + shadowKey + ", prQ is " + prQ.getFullPath());
- }
+ logger.debug("register dropped event for primary queue. BucketId is " + bucketId
+ + ", shadowKey is " + shadowKey + ", prQ is " + prQ.getFullPath());
}
}
@@ -259,11 +251,9 @@ public class ConcurrentParallelGatewaySenderEventProcessor
for (Future<Boolean> f : futures) {
try {
Boolean b = f.get();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "ConcurrentParallelGatewaySenderEventProcessor: {} stopped dispatching: {}",
- (b ? "Successfully" : "Unsuccessfully"), this);
- }
+ logger.debug(
+ "ConcurrentParallelGatewaySenderEventProcessor: {} stopped dispatching: {}",
+ (b ? "Successfully" : "Unsuccessfully"), this);
} catch (ExecutionException e) {
// we don't expect any exception but if caught then eat it and log warning
logger.warn(String.format("GatewaySender %s caught exception while stopping: %s", sender,
@@ -276,9 +266,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor
stopperService.shutdown();
closeProcessor();
- if (logger.isDebugEnabled()) {
- logger.debug("ConcurrentParallelGatewaySenderEventProcessor: Stopped dispatching: {}", this);
- }
+ logger.debug("ConcurrentParallelGatewaySenderEventProcessor: Stopped dispatching: {}", this);
}
@Override
@@ -294,9 +282,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor
parallelProcessor.pauseDispatching();
}
super.pauseDispatching();
- if (logger.isDebugEnabled()) {
- logger.debug("ConcurrentParallelGatewaySenderEventProcessor: Paused dispatching: {}", this);
- }
+ logger.debug("ConcurrentParallelGatewaySenderEventProcessor: Paused dispatching: {}", this);
}
@Override
@@ -312,9 +298,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor
parallelProcessor.resumeDispatching();
}
super.resumeDispatching();
- if (logger.isDebugEnabled()) {
- logger.debug("ConcurrentParallelGatewaySenderEventProcessor: Resumed dispatching: {}", this);
- }
+ logger.debug("ConcurrentParallelGatewaySenderEventProcessor: Resumed dispatching: {}", this);
}
@Override
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java
index 65567f0..8c5efa0 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java
@@ -74,9 +74,7 @@ public class ParallelGatewaySenderEventProcessor extends AbstractGatewaySenderEv
targetRs.add(region);
}
}
- if (logger.isDebugEnabled()) {
- logger.debug("The target Regions are(PGSEP) {}", targetRs);
- }
+ logger.debug("The target Regions are(PGSEP) {}", targetRs);
ParallelGatewaySenderQueue queue;
queue = new ParallelGatewaySenderQueue(this.sender, targetRs, this.index, this.nDispatcher,
@@ -108,11 +106,9 @@ public class ParallelGatewaySenderEventProcessor extends AbstractGatewaySenderEv
// If it is the case then the event must be coming from notificationOnly message.
// Don't enqueue the event and return from here only.
// Fix for #49081 and EntryDestroyedException in #49367.
- if (logger.isDebugEnabled()) {
- logger.debug(
- "ParallelGatewaySenderEventProcessor not enqueing the following event since tailKey is not set. {}",
- event);
- }
+ logger.debug(
+ "ParallelGatewaySenderEventProcessor not enqueing the following event since tailKey is not set. {}",
+ event);
return;
}
@@ -141,9 +137,7 @@ public class ParallelGatewaySenderEventProcessor extends AbstractGatewaySenderEv
}
getSender().getStatistics().endPut(start);
} else {
- if (logger.isDebugEnabled()) {
- logger.debug("The Event {} is filtered.", gatewayQueueEvent);
- }
+ logger.debug("The Event {} is filtered.", gatewayQueueEvent);
getSender().getStatistics().incEventsFiltered();
}
} finally {
@@ -207,9 +201,7 @@ public class ParallelGatewaySenderEventProcessor extends AbstractGatewaySenderEv
@Override
public void initializeEventDispatcher() {
- if (logger.isDebugEnabled()) {
- logger.debug(" Creating the GatewayEventCallbackDispatcher");
- }
+ logger.debug(" Creating the GatewayEventCallbackDispatcher");
this.dispatcher = new GatewaySenderEventCallbackDispatcher(this);
}
}
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 797494f..d07620f 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
@@ -211,16 +211,12 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
try {
destroyEventFromQueue(prQ, bucketId, previousTailKeyTobeRemoved);
} catch (EntryNotFoundException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Not conflating {} due to EntryNotFoundException", this,
- conflatableObject.getKeyToConflate());
- }
- }
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Conflated {} for key={} in queue for region={}", this,
- conflatableObject.getValueToConflate(), conflatableObject.getKeyToConflate(),
- prQ.getName());
+ logger.debug("{}: Not conflating {} due to EntryNotFoundException", this,
+ conflatableObject.getKeyToConflate());
}
+ logger.debug("{}: Conflated {} for key={} in queue for region={}", this,
+ conflatableObject.getValueToConflate(), conflatableObject.getKeyToConflate(),
+ prQ.getName());
}
private Object deserialize(Object serializedBytes) {
@@ -325,11 +321,9 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
this.sender.getLifeCycleLock().writeLock().lock();
PartitionedRegion prQ = null;
- if (logger.isDebugEnabled()) {
- logger.debug(
- "addShadowPartitionedRegionForUserRR: Going to create shadowpr for userRegion {}",
- userRegion.getFullPath());
- }
+ logger.debug(
+ "addShadowPartitionedRegionForUserRR: Going to create shadowpr for userRegion {}",
+ userRegion.getFullPath());
try {
String regionName = userRegion.getFullPath();
@@ -377,9 +371,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
final RegionAttributes ra = fact.getCreateAttributes();
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Attempting to create queue region: {}", this, prQName);
- }
+ logger.debug("{}: Attempting to create queue region: {}", this, prQName);
ParallelGatewaySenderQueueMetaRegion meta =
new ParallelGatewaySenderQueueMetaRegion(prQName, ra, null, cache, sender,
@@ -389,10 +381,8 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
.setSnapshotInputStream(null).setImageTarget(null);
prQ = (PartitionedRegion) fact.create(prQName);
- if (logger.isDebugEnabled()) {
- logger.debug("Region created : {} partition Attributes : {}", prQ,
- prQ.getPartitionAttributes());
- }
+ logger.debug("Region created : {} partition Attributes : {}", prQ,
+ prQ.getPartitionAttributes());
// TODO This should not be set on the PR but on the GatewaySender
prQ.enableConflation(sender.isBatchConflationEnabled());
@@ -409,9 +399,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
}
// In case of Replicated Region it may not be necessary.
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Created queue region: {}", this, prQ);
- }
+ logger.debug("{}: Created queue region: {}", this, prQ);
} else {
// in case shadowPR exists already (can be possible when sender is
// started from stop operation)
@@ -445,11 +433,9 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
PartitionedRegion childPR) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{} addShadowPartitionedRegionForUserPR: Attempting to create queue region: {}; child region: {}",
- this, userPR.getDisplayName(), childPR == null ? "null" : childPR.getDisplayName());
- }
+ logger.debug(
+ "{} addShadowPartitionedRegionForUserPR: Attempting to create queue region: {}; child region: {}",
+ this, userPR.getDisplayName(), childPR == null ? "null" : childPR.getDisplayName());
this.sender.getLifeCycleLock().writeLock().lock();
PartitionedRegion prQ = null;
@@ -534,9 +520,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
final RegionAttributes ra = fact.getCreateAttributes();
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Attempting to create queue region: {}", this, prQName);
- }
+ logger.debug("{}: Attempting to create queue region: {}", this, prQName);
ParallelGatewaySenderQueueMetaRegion meta =
metaRegionFactory.newMetataRegion(cache, prQName, ra, sender);
@@ -564,9 +548,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
prQ.shadowPRWaitForBucketRecovery();
}
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Created queue region: {}", this, prQ);
- }
+ logger.debug("{}: Created queue region: {}", this, prQ);
} else {
if (isAccessor)
@@ -642,10 +624,8 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
*/
private void handleShadowPRExistsScenario(Cache cache, PartitionedRegion prQ) {
// Note: The region will not be null if the sender is started again after stop operation
- if (logger.isDebugEnabled()) {
- logger.debug("{}: No need to create the region as the region has been retrieved: {}", this,
- prQ);
- }
+ logger.debug("{}: No need to create the region as the region has been retrieved: {}", this,
+ prQ);
}
protected void afterRegionAdd(PartitionedRegion userPR) {
@@ -875,17 +855,13 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
if (isQueueEmpty) {
queueEmptyLock.lock();
try {
- if (logger.isDebugEnabled()) {
- logger.debug("Going to notify, isQueueEmpty {}", isQueueEmpty);
- }
+ logger.debug("Going to notify, isQueueEmpty {}", isQueueEmpty);
if (isQueueEmpty) {
isQueueEmpty = false;
queueEmptyCondition.signal();
}
} finally {
- if (logger.isDebugEnabled()) {
- logger.debug("Notified!, isQueueEmpty {}", isQueueEmpty);
- }
+ logger.debug("Notified!, isQueueEmpty {}", isQueueEmpty);
queueEmptyLock.unlock();
}
}
@@ -902,17 +878,13 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
// need to find out from hcih revision this code came
}
} catch (BucketNotFoundException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("For bucket {} the current bucket redundancy is {}", brq.getId(),
- brq.getPartitionedRegion().getRegionAdvisor().getBucketAdvisor(brq.getId())
- .getBucketRedundancy());
- }
+ logger.debug("For bucket {} the current bucket redundancy is {}", brq.getId(),
+ brq.getPartitionedRegion().getRegionAdvisor().getBucketAdvisor(brq.getId())
+ .getBucketRedundancy());
} catch (ForceReattemptException e) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "getInitializedBucketForId: Got ForceReattemptException for {} for bucket = {}", this,
- brq.getId());
- }
+ logger.debug(
+ "getInitializedBucketForId: Got ForceReattemptException for {} for bucket = {}", this,
+ brq.getId());
} finally {
if (!addedValueToQueue) {
value.release();
@@ -1006,10 +978,8 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
}
}
- if (logger.isDebugEnabled()) {
- logger.debug("getRandomPrimaryBucket: total {} for this processor: {}", allBuckets.size(),
- thisProcessorBuckets.size());
- }
+ logger.debug("getRandomPrimaryBucket: total {} for this processor: {}", allBuckets.size(),
+ thisProcessorBuckets.size());
int nTry = thisProcessorBuckets.size();
@@ -1103,19 +1073,13 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
key, this, bucketId, this.sender);
}
} catch (ForceReattemptException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("Bucket :{} moved to other member", bucketId);
- }
+ logger.debug("Bucket :{} moved to other member", bucketId);
} catch (PrimaryBucketException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("Primary bucket :{} moved to other member", bucketId);
- }
+ logger.debug("Primary bucket :{} moved to other member", bucketId);
} catch (RegionDestroyedException e) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Caught RegionDestroyedException attempting to remove key {} from bucket {} in {}", key,
- bucketId, prQ.getFullPath());
- }
+ logger.debug(
+ "Caught RegionDestroyedException attempting to remove key {} from bucket {} in {}", key,
+ bucketId, prQ.getFullPath());
}
addRemovedEvent(prQ, bucketId, key);
}
@@ -1144,9 +1108,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
} catch (BucketRegionQueueUnavailableException e) {
return object;// since this is not set, it would be null
} catch (ForceReattemptException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("Remove: Got ForceReattemptException for {} for bucke = {}", this, bucketId);
- }
+ logger.debug("Remove: Got ForceReattemptException for {} for bucke = {}", this, bucketId);
}
}
return object; // OFFHEAP: ok since only callers uses it to check for empty queue
@@ -1385,11 +1347,9 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
peekedEvents.add(event);
areAllEventsForTransactionInBatch = event.isLastEventInTransaction();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Peeking extra event: {}, bucketId: {}, isLastEventInTransaction: {}, batch size: {}",
- event.getKey(), bucketId, event.isLastEventInTransaction(), batch.size());
- }
+ logger.debug(
+ "Peeking extra event: {}, bucketId: {}, isLastEventInTransaction: {}, batch size: {}",
+ event.getKey(), bucketId, event.isLastEventInTransaction(), batch.size());
}
}
if (!areAllEventsForTransactionInBatch) {
@@ -1481,14 +1441,12 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
this.peekedEventsProcessingInProgress = true;
addPreviouslyPeekedEvents(batch, batchSize);
}
- if (logger.isDebugEnabled()) {
- StringBuffer buffer = new StringBuffer();
- for (Object ge : batch) {
- buffer.append("event :");
- buffer.append(ge);
- }
- logger.debug("Adding already peeked events to the batch {}", buffer);
+ StringBuffer buffer = new StringBuffer();
+ for (Object ge : batch) {
+ buffer.append("event :");
+ buffer.append(ge);
}
+ logger.debug("Adding already peeked events to the batch {}", buffer);
}
}
@@ -1518,9 +1476,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
queueEmptyLock.lock();
try {
if (isQueueEmpty) { // merge44610: this if condition came from cheetah 44610
- if (logger.isDebugEnabled()) {
- logger.debug("Going to wait, till notified.");
- }
+ logger.debug("Going to wait, till notified.");
// merge44610: this time waiting came from cheetah 44610. In cedar 1000
// is assumed as milliseconds. In cheetah TimeUnitParamter Millisecond
// is used. In cheetah stoppable has method to consider timeunit
@@ -1531,9 +1487,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
// update the flag so that next time when we come we will block.
isQueueEmpty = this.localSizeForProcessor() == 0;
} finally {
- if (logger.isDebugEnabled()) {
- logger.debug("Going to unblock. isQueueEmpty {}", isQueueEmpty);
- }
+ logger.debug("Going to unblock. isQueueEmpty {}", isQueueEmpty);
queueEmptyLock.unlock();
}
}
@@ -1542,18 +1496,14 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
Object object = null;
BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId);
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Peekahead for the bucket {}", this, bucketId);
- }
+ logger.debug("{}: Peekahead for the bucket {}", this, bucketId);
try {
object = brq.peek();
} catch (BucketRegionQueueUnavailableException e) {
// BucketRegionQueue unavailable. Can be due to the BucketRegionQueue being destroyed.
return object;// this will be null
}
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Peeked object from bucket {} object: {}", this, bucketId, object);
- }
+ logger.debug("{}: Peeked object from bucket {} object: {}", this, bucketId, object);
if (object == null) {
if (this.stats != null) {
@@ -1630,10 +1580,8 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
size += prQ.getDataStore().getSizeOfLocalPrimaryBuckets();
}
}
- if (logger.isDebugEnabled()) {
- logger.debug("The name of the queue region is {} and the size is {}", prQ.getFullPath(),
- size);
- }
+ logger.debug("The name of the queue region is {} and the size is {}", prQ.getFullPath(),
+ size);
}
return size /* + sender.getTmpQueuedEventSize() */;
}
@@ -1650,10 +1598,8 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
size += br.size();
}
}
- if (logger.isDebugEnabled()) {
- logger.debug("The name of the queue region is {} and the size is {}", prQ.getFullPath(),
- size);
- }
+ logger.debug("The name of the queue region is {} and the size is {}", prQ.getFullPath(),
+ size);
}
return size /* + sender.getTmpQueuedEventSize() */;
}
@@ -1662,10 +1608,8 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
public int size() {
int size = 0;
for (PartitionedRegion prQ : this.userRegionNameToShadowPRMap.values()) {
- if (logger.isDebugEnabled()) {
- logger.debug("The name of the queue region is {} and the size is {}. keyset size is {}",
- prQ.getName(), prQ.size(), prQ.keys().size());
- }
+ logger.debug("The name of the queue region is {} and the size is {}. keyset size is {}",
+ prQ.getName(), prQ.size(), prQ.keys().size());
size += prQ.size();
}
@@ -1703,18 +1647,14 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
for (PartitionedRegion prQ : this.userRegionNameToShadowPRMap.values()) {
DiskRegionStats diskStats = prQ.getDiskRegionStats();
if (diskStats == null) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: DiskRegionStats for shadow PR is null. Returning the numEntriesOverflowOnDisk as 0",
- this);
- }
- return 0;
- }
- if (logger.isDebugEnabled()) {
logger.debug(
- "{}: DiskRegionStats for shadow PR is NOT null. Returning the numEntriesOverflowOnDisk obtained from DiskRegionStats",
+ "{}: DiskRegionStats for shadow PR is null. Returning the numEntriesOverflowOnDisk as 0",
this);
+ return 0;
}
+ logger.debug(
+ "{}: DiskRegionStats for shadow PR is NOT null. Returning the numEntriesOverflowOnDisk obtained from DiskRegionStats",
+ this);
numEntriesOnDisk += diskStats.getNumOverflowOnDisk();
}
return numEntriesOnDisk;
@@ -1725,17 +1665,13 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
for (PartitionedRegion prQ : this.userRegionNameToShadowPRMap.values()) {
DiskRegionStats diskStats = prQ.getDiskRegionStats();
if (diskStats == null) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: DiskRegionStats for shadow PR is null. Returning the numEntriesInVM as 0", this);
- }
- return 0;
- }
- if (logger.isDebugEnabled()) {
logger.debug(
- "{}: DiskRegionStats for shadow PR is NOT null. Returning the numEntriesInVM obtained from DiskRegionStats",
- this);
+ "{}: DiskRegionStats for shadow PR is null. Returning the numEntriesInVM as 0", this);
+ return 0;
}
+ logger.debug(
+ "{}: DiskRegionStats for shadow PR is NOT null. Returning the numEntriesInVM obtained from DiskRegionStats",
+ this);
numEntriesInVM += diskStats.getNumEntriesInVM();
}
return numEntriesInVM;
@@ -1858,14 +1794,12 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
Thread.currentThread().interrupt();
}
- if (logger.isDebugEnabled()) {
- buckToDispatchLock.lock();
- try {
- logger.debug("BatchRemovalThread about to query the batch removal map {}",
- regionToDispatchedKeysMap);
- } finally {
- buckToDispatchLock.unlock();
- }
+ buckToDispatchLock.lock();
+ try {
+ logger.debug("BatchRemovalThread about to query the batch removal map {}",
+ regionToDispatchedKeysMap);
+ } finally {
+ buckToDispatchLock.unlock();
}
final HashMap<String, Map<Integer, List>> temp;
@@ -1895,9 +1829,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
} // be somewhat tolerant of failures
catch (CancelException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread is exiting due to cancellation");
- }
+ logger.debug("BatchRemovalThread is exiting due to cancellation");
break;
} catch (VirtualMachineError err) {
SystemFailure.initiateFailure(err);
@@ -1921,16 +1853,12 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
if (checkCancelled()) {
break;
}
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread: ignoring exception", t);
- }
+ logger.debug("BatchRemovalThread: ignoring exception", t);
}
} // for
} // ensure exit message is printed
catch (CancelException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread exiting due to cancellation: " + e);
- }
+ logger.debug("BatchRemovalThread exiting due to cancellation: " + e);
} finally {
logger.info("The QueueRemovalThread is done.");
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java
index c2a3dbd..0304660c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/WaitUntilParallelGatewaySenderFlushedCoordinator.java
@@ -75,11 +75,9 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
// create and submit callable with updated timeout
Callable<Boolean> callable = createWaitUntilBucketRegionQueueFlushedCallable(
(BucketRegionQueue) br, nanosRemaining, TimeUnit.NANOSECONDS);
- if (logger.isDebugEnabled()) {
- logger.debug(
- "WaitUntilParallelGatewaySenderFlushedCoordinator: Submitting callable for bucket "
- + br.getId() + " callable=" + callable + " nanosRemaining=" + nanosRemaining);
- }
+ logger.debug(
+ "WaitUntilParallelGatewaySenderFlushedCoordinator: Submitting callable for bucket "
+ + br.getId() + " callable=" + callable + " nanosRemaining=" + nanosRemaining);
callableFutures.add(service.submit(callable));
callableCount++;
if ((callableCount % CALLABLES_CHUNK_SIZE) == 0
@@ -88,10 +86,8 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
new CallablesChunkResults(localResult, exceptionToThrow, callableFutures).invoke();
localResult = callablesChunkResults.getLocalResult();
exceptionToThrow = callablesChunkResults.getExceptionToThrow();
- if (logger.isDebugEnabled()) {
- logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Processed local result= "
- + localResult + "; exceptionToThrow= " + exceptionToThrow);
- }
+ logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Processed local result= "
+ + localResult + "; exceptionToThrow= " + exceptionToThrow);
if (exceptionToThrow != null) {
throw exceptionToThrow;
}
@@ -100,10 +96,8 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
}
// Return the full result
- if (logger.isDebugEnabled()) {
- logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Returning full result="
- + (localResult));
- }
+ logger.debug("WaitUntilParallelGatewaySenderFlushedCoordinator: Returning full result="
+ + (localResult));
return localResult;
}
@@ -172,12 +166,10 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
try {
if (msg instanceof ReplyMessage) {
ReplyMessage reply = (ReplyMessage) msg;
- if (logger.isDebugEnabled()) {
- logger
- .debug("WaitUntilGatewaySenderFlushedReplyProcessor: Processing reply from sender="
- + reply.getSender() + "; returnValue=" + reply.getReturnValue() + "; exception="
- + reply.getException());
- }
+ logger
+ .debug("WaitUntilGatewaySenderFlushedReplyProcessor: Processing reply from sender="
+ + reply.getSender() + "; returnValue=" + reply.getReturnValue() + "; exception="
+ + reply.getException());
if (reply.getException() == null) {
this.responses.put(reply.getSender(), (Boolean) reply.getReturnValue());
} else {
@@ -194,10 +186,8 @@ public class WaitUntilParallelGatewaySenderFlushedCoordinator
for (boolean singleMemberResult : this.responses.values()) {
combinedResult = combinedResult && singleMemberResult;
}
- if (logger.isDebugEnabled()) {
- logger.debug("WaitUntilGatewaySenderFlushedReplyProcessor: Returning combinedResult="
- + combinedResult);
- }
+ logger.debug("WaitUntilGatewaySenderFlushedReplyProcessor: Returning combinedResult="
+ + combinedResult);
return combinedResult;
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
index 7adf996..242f72d 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/ConcurrentSerialGatewaySenderEventProcessor.java
@@ -92,9 +92,7 @@ public class ConcurrentSerialGatewaySenderEventProcessor
processors.add(
new SerialGatewaySenderEventProcessor(this.sender, id + "." + i, getThreadMonitorObj(),
cleanQueues));
- if (logger.isDebugEnabled()) {
- logger.debug("Created the SerialGatewayEventProcessor_{}->{}", i, processors.get(i));
- }
+ logger.debug("Created the SerialGatewayEventProcessor_{}->{}", i, processors.get(i));
}
}
@@ -126,9 +124,7 @@ public class ConcurrentSerialGatewaySenderEventProcessor
public void setModifiedEventId(EntryEventImpl clonedEvent, int index) {
EventID originalEventId = clonedEvent.getEventId();
- if (logger.isDebugEnabled()) {
- logger.debug("The original EventId is {}", originalEventId);
- }
+ logger.debug("The original EventId is {}", originalEventId);
// PARALLEL_THREAD_BUFFER * (index +1) + originalEventId.getThreadID();
// generating threadId by the algorithm explained above used to clash with
// fakeThreadId generated by putAll
@@ -142,14 +138,12 @@ public class ConcurrentSerialGatewaySenderEventProcessor
*/);
EventID newEventId = new EventID(originalEventId.getMembershipID(), newThreadId,
originalEventId.getSequenceID());
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: Generated event id for event with key={}, index={}, original event id={}, threadId={}, new event id={}, newThreadId={}"
- + ":index=" + this.sender.getEventIdIndex(),
- this, clonedEvent.getKey(), index, originalEventId,
- ThreadIdentifier.toDisplayString(originalEventId.getThreadID()), newEventId,
- ThreadIdentifier.toDisplayString(newThreadId));
- }
+ logger.debug(
+ "{}: Generated event id for event with key={}, index={}, original event id={}, threadId={}, new event id={}, newThreadId={}"
+ + ":index=" + this.sender.getEventIdIndex(),
+ this, clonedEvent.getKey(), index, originalEventId,
+ ThreadIdentifier.toDisplayString(originalEventId.getThreadID()), newEventId,
+ ThreadIdentifier.toDisplayString(newThreadId));
clonedEvent.setEventId(newEventId);
}
@@ -276,10 +270,8 @@ public class ConcurrentSerialGatewaySenderEventProcessor
int memberIdHashCode = Arrays.hashCode(memberId);
int threadIdHashCode = (int) (threadId ^ (threadId >>> 32));
eventHashCode = memberIdHashCode + threadIdHashCode;
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Generated hashcode for event with key={}, memberId={}, threadId={}: {}",
- this, event.getKey(), Arrays.toString(memberId), threadId, eventHashCode);
- }
+ logger.debug("{}: Generated hashcode for event with key={}, memberId={}, threadId={}: {}",
+ this, event.getKey(), Arrays.toString(memberId), threadId, eventHashCode);
break;
case PARTITION:
eventHashCode = PartitionRegionHelper.isPartitionedRegion(event.getRegion())
@@ -287,10 +279,8 @@ public class ConcurrentSerialGatewaySenderEventProcessor
// Get the partition for the event
: event.getKey().hashCode();
// Fall back to key ordering if the region is not partitioned
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Generated partition hashcode for event with key={}: {}", this,
- event.getKey(), eventHashCode);
- }
+ logger.debug("{}: Generated partition hashcode for event with key={}: {}", this,
+ event.getKey(), eventHashCode);
break;
}
@@ -318,10 +308,8 @@ public class ConcurrentSerialGatewaySenderEventProcessor
for (Future<Boolean> f : futures) {
try {
boolean b = f.get();
- if (logger.isDebugEnabled()) {
- logger.debug("ConcurrentSerialGatewaySenderEventProcessor: {} stopped dispatching: {}",
- (b ? "Successfully" : "Unsuccessfully"), this);
- }
+ logger.debug("ConcurrentSerialGatewaySenderEventProcessor: {} stopped dispatching: {}",
+ (b ? "Successfully" : "Unsuccessfully"), this);
} catch (ExecutionException e) {
// we don't expect any exception but if caught then eat it and log
// warning
@@ -337,9 +325,7 @@ public class ConcurrentSerialGatewaySenderEventProcessor
closeProcessor();
- if (logger.isDebugEnabled()) {
- logger.debug("ConcurrentSerialGatewaySenderEventProcessor: Stopped dispatching: {}", this);
- }
+ logger.debug("ConcurrentSerialGatewaySenderEventProcessor: Stopped dispatching: {}", this);
}
@Override
@@ -355,9 +341,7 @@ public class ConcurrentSerialGatewaySenderEventProcessor
serialProcessor.pauseDispatching();
}
super.pauseDispatching();
- if (logger.isDebugEnabled()) {
- logger.debug("ConcurrentSerialGatewaySenderEventProcessor: Paused dispatching: {}", this);
- }
+ logger.debug("ConcurrentSerialGatewaySenderEventProcessor: Paused dispatching: {}", this);
}
@Override
@@ -366,9 +350,7 @@ public class ConcurrentSerialGatewaySenderEventProcessor
serialProcessor.resumeDispatching();
}
super.resumeDispatching();
- if (logger.isDebugEnabled()) {
- logger.debug("ConcurrentSerialGatewaySenderEventProcessor: Resumed dispatching: {}", this);
- }
+ logger.debug("ConcurrentSerialGatewaySenderEventProcessor: Resumed dispatching: {}", this);
}
public List<SerialGatewaySenderEventProcessor> getProcessors() {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
index 6efac4d..e22883f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
@@ -132,9 +132,7 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
// Create the region queue
this.queue = new SerialGatewaySenderQueue(sender, regionName, listener, cleanQueues);
- if (logger.isDebugEnabled()) {
- logger.debug("Created queue: {}", this.queue);
- }
+ logger.debug("Created queue: {}", this.queue);
}
/**
@@ -202,9 +200,7 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
completeFailover();
}
// Begin to process the message queue after becoming primary
- if (logger.isDebugEnabled()) {
- logger.debug("Beginning to process the message queue");
- }
+ logger.debug("Beginning to process the message queue");
if (!sender.isPrimary()) {
logger.warn("About to process the message queue but not the primary.");
@@ -371,9 +367,7 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
if (m != null) {
for (EventWrapper ew : m.values()) {
GatewaySenderEventImpl gatewayEvent = ew.event;
- if (logger.isDebugEnabled()) {
- logger.debug("releaseUnprocessedEvents:" + gatewayEvent);
- }
+ logger.debug("releaseUnprocessedEvents:" + gatewayEvent);
gatewayEvent.release();
}
this.unprocessedEvents = null;
@@ -405,7 +399,10 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
// used in the sendBatch method, and it can't be null. See EntryEventImpl
// for details.
GatewaySenderEventImpl senderEvent = null;
-
+ boolean isDebugEnabled = logger.isDebugEnabled();
+ if (isDebugEnabled) {
+ logger.debug("MLH enqueueEvent: 1 event = " + event);
+ }
boolean isPrimary = sender.isPrimary();
if (!isPrimary) {
// Fix for #40615. We need to check if we've now become the primary
@@ -428,6 +425,9 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
senderEvent =
new GatewaySenderEventImpl(operation, event, substituteValue, false,
isLastEventInTransaction);
+ if (isDebugEnabled) {
+ logger.debug("MLH enqueueEvent: 2 handleSecondaryEvent senderEvent = " + senderEvent);
+ }
handleSecondaryEvent(senderEvent);
}
}
@@ -447,6 +447,9 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
boolean queuedEvent = false;
try {
+ if (isDebugEnabled) {
+ logger.debug("MLH enqueueEvent: 3 queuePrimaryEvent senderEvent = " + senderEvent);
+ }
queuedEvent = queuePrimaryEvent(senderEvent);
} finally {
// When queuePrimaryEvent() failed with some exception, it could
@@ -455,6 +458,9 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
// and IllegalStateException could be thrown if getDeserializedValue is called
// when the event is accessed through the region queue.
if (!queuedEvent) {
+ if (isDebugEnabled) {
+ logger.debug("MLH enqueueEvent: 3 release senderEvent = " + senderEvent);
+ }
GatewaySenderEventImpl.release(senderEvent);
}
}
@@ -465,14 +471,10 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
throws IOException, CacheException {
// Queue the event
GatewaySenderStats statistics = this.sender.getStatistics();
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Queueing event ({}): {}", sender.getId(),
- (statistics.getEventsQueued() + 1), gatewayEvent);
- }
+ logger.debug("{}: Queueing event ({}): {}", sender.getId(),
+ (statistics.getEventsQueued() + 1), gatewayEvent);
if (!sender.beforeEnqueue(gatewayEvent)) {
- if (logger.isDebugEnabled()) {
- logger.debug("Event {} is not added to queue.", gatewayEvent);
- }
+ logger.debug("Event {} is not added to queue.", gatewayEvent);
statistics.incEventsFiltered();
return false;
}
@@ -481,22 +483,16 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
try {
putDone = this.queue.put(gatewayEvent);
} catch (InterruptedException e) {
- // Asif Not expected from SingleWriteSingleReadRegionQueue as it does not
- // throw
- // InterruptedException. But since both HARegionQueue and
- // SingleReadSingleWriteRegionQueue
+ // Asif Not expected from SingleWriteSingleReadRegionQueue as it does not throw
+ // InterruptedException. But since both HARegionQueue and SingleReadSingleWriteRegionQueue
// extend RegionQueue , it has to handle InterruptedException
Thread.currentThread().interrupt();
getSender().getCancelCriterion().checkCancelInProgress(e);
}
statistics.endPut(start);
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Queued event ({}): {}", sender.getId(), (statistics.getEventsQueued()),
- gatewayEvent);
- }
- // this._logger.warning(getGateway() + ": Queued event (" +
- // (statistics.getEventsQueued()) + "): " + gatewayEvent + " queue size: "
- // + this._eventQueue.size());
+ logger.debug("{}: Queued event ({}): {}", sender.getId(), (statistics.getEventsQueued()),
+ gatewayEvent);
+
/*
* FAILOVER TESTING CODE System.out.println(getName() + ": Queued event (" +
* (statistics.getEventsQueued()) + "): " + gatewayEvent.getId());
@@ -599,6 +595,11 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
*/
protected void handlePrimaryDestroy(final GatewaySenderEventImpl gatewayEvent) {
Executor my_executor = this.executor;
+
+ boolean isDebugEnabled = logger.isDebugEnabled();
+ if (isDebugEnabled) {
+ logger.debug("MLH handlePrimaryDestroy: 1 gatewayEvent = " + gatewayEvent);
+ }
synchronized (listenerObjectLock) {
if (my_executor == null) {
// should mean we are now primary
@@ -619,6 +620,12 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
*/
protected boolean basicHandlePrimaryDestroy(final EventID eventId,
boolean addToUnprocessedTokens) {
+
+ boolean isDebugEnabled = logger.isDebugEnabled();
+ if (isDebugEnabled) {
+ logger.debug("MLH basicHandlePrimaryDestroy: 1 eventId = " + eventId);
+ }
+
if (this.sender.isPrimary()) {
// no need to do anything if we have become the primary
return false;
@@ -657,6 +664,10 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
}
protected void basicHandlePrimaryEvent(final GatewaySenderEventImpl gatewayEvent) {
+ boolean isDebugEnabled = logger.isDebugEnabled();
+ if (isDebugEnabled) {
+ logger.debug("MLH basicHandlePrimaryEvent: 1 gatewayEvent = " + gatewayEvent);
+ }
if (this.sender.isPrimary()) {
// no need to do anything if we have become the primary
return;
@@ -708,6 +719,10 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
}
private void basicHandleSecondaryEvent(final GatewaySenderEventImpl gatewayEvent) {
+ boolean isDebugEnabled = logger.isDebugEnabled();
+ if (isDebugEnabled) {
+ logger.debug("MLH basicHandleSecondaryEvent: 1 gatewayEvent = " + gatewayEvent);
+ }
boolean freeGatewayEvent = true;
try {
GatewaySenderStats statistics = this.sender.getStatistics();
@@ -861,9 +876,7 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
@Override
public void initializeEventDispatcher() {
- if (logger.isDebugEnabled()) {
- logger.debug(" Creating the GatewayEventCallbackDispatcher");
- }
+ logger.debug(" Creating the GatewayEventCallbackDispatcher");
this.dispatcher = new GatewaySenderEventCallbackDispatcher(this);
}
@@ -881,24 +894,18 @@ public class SerialGatewaySenderEventProcessor extends AbstractGatewaySenderEven
destroyEvent.setEventId(dropEvent.getEventId());
destroyEvent.disallowOffHeapValues();
destroyEvent.setTailKey(-1L);
- if (logger.isDebugEnabled()) {
- logger.debug(
- "SerialGatewaySenderEventProcessor sends BatchDestroyOperation to secondary for event {}",
- destroyEvent);
- }
+ logger.debug(
+ "SerialGatewaySenderEventProcessor sends BatchDestroyOperation to secondary for event {}",
+ destroyEvent);
try {
BatchDestroyOperation op = new BatchDestroyOperation(destroyEvent);
op.distribute();
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread completed destroy of dropped event {}", dropEvent);
- }
+ logger.debug("BatchRemovalThread completed destroy of dropped event {}", dropEvent);
} catch (Exception ignore) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Exception in sending dropped event could be ignored in order not to interrupt sender starting",
- ignore);
- }
+ logger.debug(
+ "Exception in sending dropped event could be ignored in order not to interrupt sender starting",
+ ignore);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
index 45f4f84..77f7368 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderQueue.java
@@ -232,9 +232,7 @@ public class SerialGatewaySenderQueue implements RegionQueue {
this.removalThread = new BatchRemovalThread(abstractSender.getCache());
this.removalThread.start();
this.sender = abstractSender;
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Contains {} elements", this, size());
- }
+ logger.debug("{}: Contains {} elements", this, size());
}
@Override
@@ -273,9 +271,7 @@ public class SerialGatewaySenderQueue implements RegionQueue {
// signal that a new object is available.
incrementTailKey();
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Inserted {} -> {}", this, key, object);
- }
+ logger.debug("{}: Inserted {} -> {}", this, key, object);
if (object instanceof Conflatable) {
removeOldEntry((Conflatable) object, key);
}
@@ -326,11 +322,9 @@ public class SerialGatewaySenderQueue implements RegionQueue {
} catch (EntryNotFoundException ok) {
// this is acceptable because the conflation can remove entries
// out from underneath us.
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: Did not destroy entry at {} it was not there. It should have been removed by conflation.",
- this, key);
- }
+ logger.debug(
+ "{}: Did not destroy entry at {} it was not there. It should have been removed by conflation.",
+ this, key);
}
boolean wasEmpty = this.lastDispatchedKey == this.lastDestroyedKey;
@@ -360,11 +354,9 @@ public class SerialGatewaySenderQueue implements RegionQueue {
}
}
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: Destroyed entry at key {} setting the lastDispatched Key to {}. The last destroyed entry was {}",
- this, key, this.lastDispatchedKey, this.lastDestroyedKey);
- }
+ logger.debug(
+ "{}: Destroyed entry at key {} setting the lastDispatched Key to {}. The last destroyed entry was {}",
+ this, key, this.lastDispatchedKey, this.lastDestroyedKey);
}
/**
@@ -484,11 +476,9 @@ public class SerialGatewaySenderQueue implements RegionQueue {
batch.add(event);
areAllEventsForTransactionInBatch = event.isLastEventInTransaction();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Peeking extra event: {}, isLastEventInTransaction: {}, batch size: {}",
- event.getKey(), event.isLastEventInTransaction(), batch.size());
- }
+ logger.debug(
+ "Peeking extra event: {}, isLastEventInTransaction: {}, batch size: {}",
+ event.getKey(), event.isLastEventInTransaction(), batch.size());
}
lastKeyForTransaction = eventsAndKey.lastKey;
}
@@ -689,10 +679,8 @@ public class SerialGatewaySenderQueue implements RegionQueue {
if (index != null) {
this.stats.decConflationIndexesMapSize();
}
- if (logger.isDebugEnabled()) {
- if (index != null) {
- logger.debug("{}: Removed index {} for {}", this, index, object);
- }
+ if (index != null) {
+ logger.debug("{}: Removed index {} for {}", this, index, object);
}
}
}
@@ -815,9 +803,7 @@ public class SerialGatewaySenderQueue implements RegionQueue {
}
}
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Peeked {}->{}", this, currentKey, object);
- }
+ logger.debug("{}: Peeked {}->{}", this, currentKey, object);
if (object != null) {
peekedIds.add(currentKey);
@@ -975,10 +961,8 @@ public class SerialGatewaySenderQueue implements RegionQueue {
this.tailKey.set(inc(largestKey));
}
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Initialized tail key to: {}, head key to: {}", this, this.tailKey,
- this.headKey);
- }
+ logger.debug("{}: Initialized tail key to: {}, head key to: {}", this, this.tailKey,
+ this.headKey);
}
}
@@ -1044,10 +1028,8 @@ public class SerialGatewaySenderQueue implements RegionQueue {
if (NO_ACK) {
factory.setScope(Scope.DISTRIBUTED_NO_ACK);
}
- if (logger.isDebugEnabled()) {
- logger.debug("The policy of region is {}",
- (this.enablePersistence ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.REPLICATE));
- }
+ logger.debug("The policy of region is {}",
+ (this.enablePersistence ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.REPLICATE));
// Set listener if it is not null. The listener will be non-null
// when the user of this queue is a secondary VM.
if (listener != null) {
@@ -1066,9 +1048,7 @@ public class SerialGatewaySenderQueue implements RegionQueue {
factory.setDiskSynchronous(this.isDiskSynchronous);
// Create the region
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Attempting to create queue region: {}", this, this.regionName);
- }
+ logger.debug("{}: Attempting to create queue region: {}", this, this.regionName);
final RegionAttributes<Long, AsyncEvent> ra = factory.getCreateAttributes();
try {
SerialGatewaySenderQueueMetaRegion meta =
@@ -1085,9 +1065,7 @@ public class SerialGatewaySenderQueue implements RegionQueue {
// Add overflow statistics to the mbean
addOverflowStatisticsToMBean(gemCache, sender);
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Created queue region: {}", this, this.region);
- }
+ logger.debug("{}: Created queue region: {}", this, this.region);
} catch (CacheException e) {
logger.fatal(String.format("%s: The queue region named %s could not be created",
new Object[] {this, this.regionName}),
@@ -1217,10 +1195,8 @@ public class SerialGatewaySenderQueue implements RegionQueue {
Thread.currentThread().interrupt();
}
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread about to send the last Dispatched key {}",
- lastDispatchedKey);
- }
+ logger.debug("BatchRemovalThread about to send the last Dispatched key {}",
+ lastDispatchedKey);
long temp;
synchronized (SerialGatewaySenderQueue.this) {
@@ -1241,17 +1217,13 @@ public class SerialGatewaySenderQueue implements RegionQueue {
BatchDestroyOperation op = new BatchDestroyOperation(event);
op.distribute();
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread completed destroy of keys from {} to {}",
- lastDestroyedKey, temp);
- }
+ logger.debug("BatchRemovalThread completed destroy of keys from {} to {}",
+ lastDestroyedKey, temp);
lastDestroyedKey = temp;
} // be somewhat tolerant of failures
catch (CancelException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread is exiting due to cancellation");
- }
+ logger.debug("BatchRemovalThread is exiting due to cancellation");
break;
} catch (VirtualMachineError err) {
SystemFailure.initiateFailure(err);
@@ -1268,16 +1240,12 @@ public class SerialGatewaySenderQueue implements RegionQueue {
if (checkCancelled()) {
break;
}
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread: ignoring exception", t);
- }
+ logger.debug("BatchRemovalThread: ignoring exception", t);
}
} // for
} // ensure exit message is printed
catch (CancelException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("BatchRemovalThread exiting due to cancellation: " + e);
- }
+ logger.debug("BatchRemovalThread exiting due to cancellation: " + e);
} finally {
logger.info("The QueueRemovalThread is done.");
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialSecondaryGatewayListener.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialSecondaryGatewayListener.java
index 2c6a2aa..4ede414 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialSecondaryGatewayListener.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialSecondaryGatewayListener.java
@@ -84,10 +84,8 @@ public class SerialSecondaryGatewayListener extends CacheListenerAdapter {
Object oldValue = event.getOldValue();
if (oldValue instanceof GatewaySenderEventImpl) {
GatewaySenderEventImpl senderEvent = (GatewaySenderEventImpl) oldValue;
- if (logger.isDebugEnabled()) {
- logger.debug("Received after Destroy for Secondary event {} the key was {}", senderEvent,
- event.getKey());
- }
+ logger.debug("Received after Destroy for Secondary event {} the key was {}", senderEvent,
+ event.getKey());
this.processor.handlePrimaryDestroy(senderEvent);
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/StatusReporterTest.java b/geode-core/src/test/java/org/apache/geode/StatusReporterTest.java
new file mode 100644
index 0000000..e6dca2a
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/StatusReporterTest.java
@@ -0,0 +1,105 @@
+package org.apache.geode;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.PrintStream;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class StatusReporterTest {
+
+ @Before
+ public void setUp() throws Exception {}
+
+ @After
+ public void tearDown() throws Exception {}
+
+ @Test
+ public void testStart() throws InterruptedException {
+ StatusReporter.start();
+ while (StatusReporter.getStatus() != StatusReporter.State.SR_RUNNING) {
+ Thread.sleep(1000);
+ }
+ assertThat(StatusReporter.getStatus()).isEqualTo(StatusReporter.State.SR_RUNNING);
+ }
+
+ @Test
+ public void testRegisterComponent() {
+ PrintStream printStream = Mockito.mock(PrintStream.class);
+ ComponentStatus componentStatus = new ComponentStatus() {
+ @Override
+ public String name() {
+ return "Test Component";
+ }
+
+ @Override
+ public String getStatusString() {
+ return "Everything is fine and happy.";
+ }
+
+ @Override
+ public void print() {
+ printStream.println(getStatusString());
+ }
+ };
+
+ StatusReporter.registerComponent(componentStatus);
+ assertThat(StatusReporter.isComponentRegistered(componentStatus)).isTrue();
+ StatusReporter.removeComponent(componentStatus);
+ }
+
+ @Test
+ public void testGetStatusReport() {
+ PrintStream printStream = Mockito.mock(PrintStream.class);
+ ComponentStatus componentStatus = new ComponentStatus() {
+ @Override
+ public String name() {
+ return "Test Component";
+ }
+
+ @Override
+ public String getStatusString() {
+ return "Everything is fine and happy.";
+ }
+
+ @Override
+ public void print() {
+ printStream.println(getStatusString());
+ }
+ };
+
+ StatusReporter.registerComponent(componentStatus);
+ assertThat(StatusReporter.getStatusReport()).isTrue();
+ Mockito.verify(printStream).println(componentStatus.getStatusString());
+ StatusReporter.removeComponent(componentStatus);
+ }
+
+ @Test
+ public void testPeriodicReports() throws InterruptedException {
+ PrintStream printStream = Mockito.mock(PrintStream.class);
+ ComponentStatus componentStatus = new ComponentStatus() {
+ @Override
+ public String name() {
+ return "Test Component";
+ }
+
+ @Override
+ public String getStatusString() {
+ return "Everything is fine and happy.";
+ }
+
+ @Override
+ public void print() {
+ printStream.println(getStatusString());
+ }
+ };
+
+ StatusReporter.registerComponent(componentStatus);
+ Thread.sleep(5000);
+ Mockito.verify(printStream).println(componentStatus.getStatusString());
+ StatusReporter.removeComponent(componentStatus);
+ }
+}
diff --git a/geode-wan/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderImpl.java b/geode-wan/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderImpl.java
index 97436a2..c7f934e 100644
--- a/geode-wan/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderImpl.java
+++ b/geode-wan/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderImpl.java
@@ -18,6 +18,8 @@ import java.util.Set;
import org.apache.logging.log4j.Logger;
+import org.apache.geode.ComponentStatus;
+import org.apache.geode.StatusReporter;
import org.apache.geode.cache.asyncqueue.AsyncEventListener;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.apache.geode.distributed.DistributedLockService;
@@ -43,7 +45,8 @@ import org.apache.geode.logging.internal.log4j.api.LogService;
/**
* @since GemFire 7.0
*/
-public class SerialGatewaySenderImpl extends AbstractRemoteGatewaySender {
+public class SerialGatewaySenderImpl extends AbstractRemoteGatewaySender implements
+ ComponentStatus {
private static final Logger logger = LogService.getLogger();
@@ -63,6 +66,7 @@ public class SerialGatewaySenderImpl extends AbstractRemoteGatewaySender {
}
private void start(boolean cleanQueues) {
+ StatusReporter.registerComponent(this);
if (logger.isDebugEnabled()) {
logger.debug("Starting gatewaySender : {}", this);
}
@@ -110,6 +114,7 @@ public class SerialGatewaySenderImpl extends AbstractRemoteGatewaySender {
logger
.info("Started {}", this);
+ status = "Started";
enqueueTempEvents();
} finally {
this.getLifeCycleLock().writeLock().unlock();
@@ -262,4 +267,24 @@ public class SerialGatewaySenderImpl extends AbstractRemoteGatewaySender {
return null;
}
}
+
+ @Override
+ public String name() {
+ return "SerialGatewaySender";
+ }
+
+ private String status = "Started";
+
+ @Override
+ public String getStatusString() {
+ synchronized (this) {
+ return status;
+ }
+ }
+
+ @Override
+ public void print() {
+ String status = getStatusString();
+ System.out.println("Component name: " + name() + " status: " + status + " " + toString());
+ }
}