You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2018/07/09 22:46:25 UTC
[geode] branch develop updated: GEODE-5393: StateFlushOperation
hangs waiting for non-existant operation to complete
This is an automated email from the ASF dual-hosted git repository.
bschuchardt pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 9d87117 GEODE-5393: StateFlushOperation hangs waiting for non-existant operation to complete
9d87117 is described below
commit 9d87117b131172e05ba28fc6154aca3370115aa1
Author: Bruce Schuchardt <bs...@pivotal.io>
AuthorDate: Mon Jul 9 15:45:29 2018 -0700
GEODE-5393: StateFlushOperation hangs waiting for non-existant operation to complete
I've added additional debugging to DistributionAdvisor so that it knows
which threads are performing operations and can log them at debug level.
This let me determine that a putAll operation was the source of the hang
due to an exception being thrown during message distribution in
DistributedCacheOperation.startOperation(). The exception resulted in
DistributionAdvisor.endOperation() not being invoked correctly.
---
.../distributed/internal/DistributionAdvisor.java | 427 +++++++++++++--------
.../internal/admin/remote/ShutdownAllRequest.java | 19 +-
.../internal/cache/DistributedCacheOperation.java | 37 +-
.../geode/internal/cache/LocalRegionDataView.java | 4 +-
.../geode/internal/cache/TXCommitMessage.java | 2 +-
.../cache/DistributedCacheOperationTest.java | 61 +++
6 files changed, 367 insertions(+), 183 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java
index 9ac4563..71d64a4 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/DistributionAdvisor.java
@@ -17,6 +17,8 @@ package org.apache.geode.distributed.internal;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import java.io.OutputStream;
+import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -122,11 +124,12 @@ public class DistributionAdvisor {
private final AtomicInteger profileVersionSequencer = new AtomicInteger(START_VERSION_NUMBER);
/**
- * This system property is not supported and disabling intelligent messaging is currently
- * problematic
+ * The operationMonitor tracks in-progress cache operations and holds the profile set
+ * version number
*/
- protected static final boolean disabled = Boolean.getBoolean("disable-intelligent-msg");
-
+ private final OperationMonitor operationMonitor =
+ logger.isDebugEnabled() ? new ThreadTrackingOperationMonitor(this)
+ : new OperationMonitor(this);
/**
@@ -143,36 +146,12 @@ public class DistributionAdvisor {
private final Object initializeLock = new Object();
/**
- * the version of the profile set
- *
- * @since GemFire 5.1
- */
- private long membershipVersion;
-
- /**
* whether membership ops are closed (because the DA's been closed). Access under synchronization
* on (this)
*/
private boolean membershipClosed;
/**
- * opCountLock guards access to previousVersionOpCount and currentVersionOpCount
- */
- private final Object opCountLock = new Object();
-
- /**
- * the number of operations in-progress for previous versions of the profile set. Guarded by
- * opCountLock
- */
- private long previousVersionOpCount;
-
- /**
- * the number of operations in-progress for the current version of the profile set. Guarded by
- * opCountLock
- */
- private long currentVersionOpCount;
-
- /**
* Hold onto removed profiles to compare to late-processed profiles. Fix for bug 36881. Protected
* by synchronizing on this DistributionAdvisor. guarded.By this DistributionAdvisor
*/
@@ -399,10 +378,7 @@ public class DistributionAdvisor {
try {
synchronized (this) {
this.membershipClosed = true;
- synchronized (this.opCountLock) {
- this.previousVersionOpCount = 0;
- this.currentVersionOpCount = 0;
- }
+ operationMonitor.close();
}
getDistributionManager().removeMembershipListener(this.ml);
} catch (CancelException e) {
@@ -412,6 +388,7 @@ public class DistributionAdvisor {
}
}
+
/**
* Atomically add listener to the list to receive notification when a *new* profile is added or a
* profile is removed, and return adviseGeneric(). This ensures that no membership listener calls
@@ -616,20 +593,11 @@ public class DistributionAdvisor {
newProfile.initialMembershipVersion = oldProfile.initialMembershipVersion;
} else {
if (!membershipClosed) {
- membershipVersion++;
- if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
- logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "StateFlush incremented membership version: {}", membershipVersion);
- }
- newProfile.initialMembershipVersion = membershipVersion;
- synchronized (this.opCountLock) {
- previousVersionOpCount += currentVersionOpCount;
- currentVersionOpCount = 0;
- }
+ operationMonitor.initNewProfile(newProfile);
}
}
} else {
- forceNewMembershipVersion();
+ operationMonitor.forceNewMembershipVersion();
}
if (isTraceEnabled_DistributionAdvisor) {
@@ -719,23 +687,8 @@ public class DistributionAdvisor {
*
* @since GemFire 5.1
*/
- public synchronized void forceNewMembershipVersion() {
- if (!membershipClosed) {
- membershipVersion++;
- if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
- logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "StateFlush forced new membership version: {}", membershipVersion);
- }
- synchronized (this.opCountLock) {
- previousVersionOpCount += currentVersionOpCount;
- currentVersionOpCount = 0;
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_STATE_FLUSH_VERBOSE)) {
- logger.trace(LogMarker.DISTRIBUTION_STATE_FLUSH_VERBOSE,
- "advisor for {} forced new membership version to {} previousOpCount={}", getAdvisee(),
- membershipVersion, previousVersionOpCount);
- }
- }
- }
+ public void forceNewMembershipVersion() {
+ operationMonitor.forceNewMembershipVersion();
}
/**
@@ -746,20 +699,8 @@ public class DistributionAdvisor {
* @return the current membership version for this advisor
* @since GemFire 5.1
*/
- public synchronized long startOperation() {
- if (logger.isTraceEnabled(LogMarker.DISTRIBUTION_STATE_FLUSH_VERBOSE)) {
- logger.trace(LogMarker.DISTRIBUTION_STATE_FLUSH_VERBOSE,
- "startOperation() op count is now {} in view version {}", currentVersionOpCount + 1,
- membershipVersion);
- }
- synchronized (this.opCountLock) {
- currentVersionOpCount++;
- if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
- logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "StateFlush current opcount incremented: {}",
- currentVersionOpCount);
- }
- }
- return membershipVersion;
+ public long startOperation() {
+ return operationMonitor.startOperation();
}
/**
@@ -769,23 +710,8 @@ public class DistributionAdvisor {
* @param version The membership version returned by startOperation
* @since GemFire 5.1
*/
- public synchronized long endOperation(long version) {
- synchronized (this.opCountLock) {
- if (version == membershipVersion) {
- currentVersionOpCount--;
- if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
- logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "StateFlush current opcount deccremented: {}", currentVersionOpCount);
- }
- } else {
- previousVersionOpCount--;
- if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
- logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "StateFlush previous opcount incremented: {}", previousVersionOpCount);
- }
- }
- }
- return membershipVersion;
+ public void endOperation(long version) {
+ operationMonitor.endOperation(version);
}
/**
@@ -795,63 +721,13 @@ public class DistributionAdvisor {
* @since GemFire 5.1
*/
public void waitForCurrentOperations() {
- long timeout =
- 1000L * this.getDistributionManager().getSystem().getConfig().getAckWaitThreshold();
- waitForCurrentOperations(logger, timeout, timeout * 2L);
+ operationMonitor.waitForCurrentOperations();
}
public void waitForCurrentOperations(Logger alertLogger, long warnMS, long severeAlertMS) {
// this may wait longer than it should if the membership version changes, dumping
// more operations into the previousVersionOpCount
- final long startTime = System.currentTimeMillis();
- final long warnTime = startTime + warnMS;
- final long severeAlertTime = startTime + severeAlertMS;
- boolean warned = false;
- boolean severeAlertIssued = false;
- final boolean isDebugEnabled_STATE_FLUSH_OP =
- DistributionAdvisor.logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE);
- while (true) {
- long opCount;
- synchronized (this.opCountLock) {
- opCount = this.previousVersionOpCount;
- }
- if (opCount <= 0) {
- if (warned) {
- alertLogger.info("Wait for current operations completed");
- }
- break;
- }
- // The advisor's close() method will set the pVOC to zero. This loop
- // must not terminate due to cache closure until that happens.
- // See bug 34361 comment 79
- if (isDebugEnabled_STATE_FLUSH_OP) {
- DistributionAdvisor.logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "Waiting for current operations to finish({})", opCount);
- }
- try {
- Thread.sleep(50);
- } catch (InterruptedException e) {
- throw new GemFireIOException("State flush interrupted");
- }
- long now = System.currentTimeMillis();
- if ((!warned) && System.currentTimeMillis() >= warnTime) {
- warned = true;
- alertLogger.warn("This operation has been stalled for {} milliseconds waiting for "
- + "current operations to complete.", warnMS);
- } else if (warned && !severeAlertIssued && (now >= severeAlertTime)) {
- // OSProcess.printStacks(0);
- alertLogger.fatal("This operation has been stalled for {} milliseconds "
- + "waiting for current operations to complete. Something may be blocking operations.",
- severeAlertMS);
- severeAlertIssued = true;
- }
- }
- if (this.membershipClosed) {
- if (isDebugEnabled_STATE_FLUSH_OP) {
- DistributionAdvisor.logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
- "State Flush stopped waiting for operations to distribute because advisor has been closed");
- }
- }
+ operationMonitor.waitForCurrentOperations(alertLogger, warnMS, severeAlertMS);
}
/**
@@ -1279,12 +1155,6 @@ public class DistributionAdvisor {
/** All advise methods go through this method */
protected Set<InternalDistributedMember> adviseFilter(Filter f) {
initializationGate();
- if (disabled) {
- if (logger.isDebugEnabled()) {
- logger.debug("Intelligent Messaging Disabled");
- }
- return getDefaultDistributionMembers();
- }
Set<InternalDistributedMember> recipients = null;
Profile[] locProfiles = this.profiles; // grab current profiles
for (int i = 0; i < locProfiles.length; i++) {
@@ -1310,12 +1180,6 @@ public class DistributionAdvisor {
**/
protected boolean satisfiesFilter(Filter f) {
initializationGate();
- if (disabled) {
- if (logger.isDebugEnabled()) {
- logger.debug("Intelligent Messaging Disabled");
- }
- return !getDefaultDistributionMembers().isEmpty();
- }
Profile[] locProfiles = this.profiles; // grab current profiles
for (Profile p : locProfiles) {
if (f.include(p)) {
@@ -1680,4 +1544,257 @@ public class DistributionAdvisor {
return null;
}
}
+
+
+
+ private static class OperationMonitor {
+ private final DistributionAdvisor distributionAdvisor;
+
+ /**
+ * the version of the profile set
+ */
+ private long membershipVersion;
+
+ /**
+ * the number of operations in-progress for previous versions of the profile set
+ */
+ private long previousVersionOpCount;
+ /**
+ * the number of operations in-progress for the current version of the profile set
+ */
+ private long currentVersionOpCount;
+
+ /**
+ * for debugging stalled state-flush operations we track threads performing operations
+ * and capture the state when startOperatiopn is invoked
+ */
+ private boolean closed;
+
+ private OperationMonitor(DistributionAdvisor distributionAdvisor) {
+ this.distributionAdvisor = distributionAdvisor;
+ }
+
+ private synchronized void incrementMembershipVersion() {
+ membershipVersion++;
+ }
+
+ /**
+ * Create a new version of the membership profile set. This is used in flushing state out of the
+ * VM for previous versions of the set.
+ *
+ * @since GemFire 5.1
+ */
+ synchronized void forceNewMembershipVersion() {
+ if (!closed) {
+ incrementMembershipVersion();
+ previousVersionOpCount += currentVersionOpCount;
+ currentVersionOpCount = 0;
+ membershipVersionChanged();
+ }
+ }
+
+ /**
+ * this method must be invoked at the start of every operation that can modify the state of
+ * resource. The return value must be recorded and sent to the advisor in an endOperation
+ * message when messages for the operation have been put in the DistributionManager's outgoing
+ * "queue".
+ *
+ * @return the current membership version for this advisor
+ * @since GemFire 5.1
+ */
+ synchronized long startOperation() {
+ logNewOperation();
+ currentVersionOpCount++;
+ return membershipVersion;
+ }
+
+ /**
+ * This method must be invoked when messages for an operation have been put in the
+ * DistributionManager's outgoing queue.
+ *
+ * @param version The membership version returned by startOperation
+ * @since GemFire 5.1
+ */
+ synchronized void endOperation(long version) {
+ if (version == membershipVersion) {
+ currentVersionOpCount--;
+ logEndOperation(true);
+ } else {
+ previousVersionOpCount--;
+ logEndOperation(false);
+ }
+ }
+
+ /**
+ * wait for the current operations being sent on views prior to the joining of the given member
+ * to be placed on communication channels before returning
+ *
+ * @since GemFire 5.1
+ */
+ void waitForCurrentOperations() {
+ long timeout =
+ 1000L * distributionAdvisor.getDistributionManager().getSystem().getConfig()
+ .getAckWaitThreshold();
+ waitForCurrentOperations(logger, timeout, timeout * 2L);
+ }
+
+ void waitForCurrentOperations(Logger alertLogger, long warnMS, long severeAlertMS) {
+ // this may wait longer than it should if the membership version changes, dumping
+ // more operations into the previousVersionOpCount
+ final long startTime = System.currentTimeMillis();
+ final long warnTime = startTime + warnMS;
+ final long severeAlertTime = startTime + severeAlertMS;
+ boolean warned = false;
+ boolean severeAlertIssued = false;
+ while (operationsAreInProgress()) {
+ // The advisor's close() method will set the pVOC to zero. This loop
+ // must not terminate due to cache closure until that happens.
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ throw new GemFireIOException("State flush interrupted");
+ }
+ long now = System.currentTimeMillis();
+ if ((!warned) && System.currentTimeMillis() >= warnTime) {
+ warned = true;
+ logWaitOnOperationsWarning(alertLogger, warnMS);
+ } else if (warned && !severeAlertIssued && (now >= severeAlertTime)) {
+ logWaitOnOperationsSevere(alertLogger, severeAlertMS);
+ severeAlertIssued = true;
+ }
+ }
+ if (warned) {
+ alertLogger.info("Wait for current operations completed");
+ }
+ }
+
+ synchronized boolean operationsAreInProgress() {
+ return previousVersionOpCount > 0;
+ }
+
+ synchronized void initNewProfile(Profile newProfile) {
+ membershipVersion++;
+ newProfile.initialMembershipVersion = membershipVersion;
+ previousVersionOpCount =
+ previousVersionOpCount + currentVersionOpCount;
+ currentVersionOpCount = 0;
+ membershipVersionChanged();
+ }
+
+ synchronized void close() {
+ previousVersionOpCount = 0;
+ currentVersionOpCount = 0;
+ closed = true;
+ }
+
+ void logNewOperation() {}
+
+ void logEndOperation(boolean newOperation) {}
+
+ void logWaitOnOperationsSevere(Logger alertLogger, long severeAlertMS) {
+ // OSProcess.printStacks(0);
+ alertLogger.fatal("This thread has been stalled for {} milliseconds "
+ + "waiting for current operations to complete. Something may be blocking operations.",
+ severeAlertMS);
+ }
+
+ void logWaitOnOperationsWarning(Logger alertLogger, long warnMS) {
+ alertLogger.warn("This thread has been stalled for {} milliseconds waiting for "
+ + "current operations to complete.", warnMS);
+ }
+
+ void membershipVersionChanged() {}
+
+ }
+
+ private static class ThreadTrackingOperationMonitor extends OperationMonitor {
+
+ /**
+ * for debugging stalled state-flush operations we track threads performing operations
+ * and capture the state when startOperatiopn is invoked
+ */
+ private final Map<Thread, ExceptionWrapper> currentVersionOperationThreads;
+ private final Map<Thread, ExceptionWrapper> previousVersionOperationThreads;
+
+ private ThreadTrackingOperationMonitor(
+ DistributionAdvisor distributionAdvisor) {
+ super(distributionAdvisor);
+ this.currentVersionOperationThreads = new HashMap<>();
+ this.previousVersionOperationThreads = new HashMap<>();
+ }
+
+ @Override
+ void logNewOperation() {
+ currentVersionOperationThreads.put(Thread.currentThread(),
+ new ExceptionWrapper(new Exception("stack trace")));
+ }
+
+ @Override
+ void logEndOperation(boolean newOp) {
+ if (newOp) {
+ currentVersionOperationThreads.remove(Thread.currentThread());
+ } else {
+ previousVersionOperationThreads.remove(Thread.currentThread());
+ }
+ }
+
+ @Override
+ void logWaitOnOperationsWarning(Logger alertLogger, long warnMS) {
+ super.logWaitOnOperationsWarning(alertLogger, warnMS);
+ synchronized (this) {
+ logger
+ .debug("Waiting for these threads: {}", previousVersionOperationThreads);
+ logger
+ .debug("New version threads are {}", currentVersionOperationThreads);
+ }
+ }
+
+ @Override
+ void logWaitOnOperationsSevere(Logger alertLogger, long severeAlertMS) {
+ super.logWaitOnOperationsSevere(alertLogger, severeAlertMS);
+ synchronized (this) {
+ logger
+ .debug("Waiting for these threads: {}", previousVersionOperationThreads);
+ logger
+ .debug("New version threads are {}", currentVersionOperationThreads);
+ }
+ }
+
+ @Override
+ void membershipVersionChanged() {
+ super.membershipVersionChanged();
+ previousVersionOperationThreads
+ .putAll(currentVersionOperationThreads);
+ currentVersionOperationThreads.clear();
+ }
+
+
+ /**
+ * ExceptionWrapper is used in debugging hangs in waitForCurrentOperations(). It
+ * captures the call stack of a thread invoking startOperation().
+ */
+ private static class ExceptionWrapper {
+ private Exception exception;
+
+ ExceptionWrapper(Exception exception) {
+ this.exception = exception;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder(500);
+ OutputStream os = new OutputStream() {
+ @Override
+ public void write(int i) {
+ builder.append((char) i);
+ }
+ };
+ PrintStream stream = new PrintStream(os);
+ exception.printStackTrace(stream);
+ return builder.toString();
+ }
+ }
+
+ }
+
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/ShutdownAllRequest.java b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/ShutdownAllRequest.java
index 4579fe1..1d074fc 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/admin/remote/ShutdownAllRequest.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/admin/remote/ShutdownAllRequest.java
@@ -20,6 +20,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Set;
import java.util.TreeSet;
@@ -260,13 +261,11 @@ public class ShutdownAllRequest extends AdminRequest {
}
if (msg instanceof ShutdownAllResponse) {
if (((ShutdownAllResponse) msg).isToShutDown()) {
- if (logger.isDebugEnabled()) {
- synchronized (results) {
- logger.debug("{} adding {} to result set {}", this, msg.getSender(),
- results);
- }
+ synchronized (results) {
+ logger.debug("{} adding {} to result set {}", this, msg.getSender(),
+ results);
+ this.results.add(msg.getSender());
}
- this.results.add(msg.getSender());
} else {
// for member without cache, we will not wait for its result
// so no need to wait its DS to close either
@@ -291,9 +290,11 @@ public class ShutdownAllRequest extends AdminRequest {
}
public Set getResults() {
- logger.debug("{} shutdownAll returning {}", this,
- results/* , new Exception("stack trace") */);
- return results;
+ synchronized (results) {
+ logger.debug("{} shutdownAll returning {}", this,
+ results);
+ return new HashSet(results);
+ }
}
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
index 4fee0a0..a7c3533 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
@@ -253,25 +253,30 @@ public abstract class DistributedCacheOperation {
public long startOperation() {
DistributedRegion region = getRegion();
long viewVersion = -1;
- if (this.containsRegionContentChange()) {
- viewVersion = region.getDistributionAdvisor().startOperation();
- }
- if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
- logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "dispatching operation in view version {}",
- viewVersion);
- }
try {
- _distribute();
- } catch (InvalidVersionException e) {
- if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
- logger.trace(LogMarker.DM_VERBOSE,
- "PutAll failed since versions were missing; retrying again", e);
+ if (this.containsRegionContentChange()) {
+ viewVersion = region.getDistributionAdvisor().startOperation();
}
+ if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
+ logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "dispatching operation in view version {}",
+ viewVersion);
+ }
+ try {
+ _distribute();
+ } catch (InvalidVersionException e) {
+ if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
+ logger.trace(LogMarker.DM_VERBOSE,
+ "PutAll failed since versions were missing; retrying", e);
+ }
- if (test_InvalidVersionAction != null) {
- test_InvalidVersionAction.run();
+ if (test_InvalidVersionAction != null) {
+ test_InvalidVersionAction.run();
+ }
+ _distribute();
}
- _distribute();
+ } catch (RuntimeException | Error e) {
+ endOperation(viewVersion);
+ throw e;
}
return viewVersion;
}
@@ -310,7 +315,7 @@ public abstract class DistributedCacheOperation {
* members. This method should wrapped by startOperation() and endOperation() in try/finally
* block.
*/
- private void _distribute() {
+ protected void _distribute() {
DistributedRegion region = getRegion();
DistributionManager mgr = region.getDistributionManager();
boolean reliableOp = isOperationReliable() && region.requiresReliabilityCheck();
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
index c17a25c..6779574 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
@@ -326,7 +326,7 @@ public class LocalRegionDataView implements InternalDataView {
token = reg.postPutAllSend(putallOp, successfulPuts);
reg.postPutAllFireEvents(putallOp, successfulPuts);
} finally {
- if (reg instanceof DistributedRegion) {
+ if (token != -1 && reg instanceof DistributedRegion) {
putallOp.endOperation(token);
}
}
@@ -348,7 +348,7 @@ public class LocalRegionDataView implements InternalDataView {
token = reg.postRemoveAllSend(op, successfulOps);
reg.postRemoveAllFireEvents(op, successfulOps);
} finally {
- if (reg instanceof DistributedRegion) {
+ if (token != -1 && reg instanceof DistributedRegion) {
op.endOperation(token);
}
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java
index 722130e..13f15ea 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java
@@ -300,7 +300,7 @@ public class TXCommitMessage extends PooledDistributionMessage
// need to continue the iteration if one of the regions is destroyed
// since others may still be okay
try {
- long newv = dr.getDistributionAdvisor().endOperation(viewVersion);
+ dr.getDistributionAdvisor().endOperation(viewVersion);
} catch (RuntimeException ex) {
rte = ex;
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java
index c0eb997..bde43fb 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/DistributedCacheOperationTest.java
@@ -15,6 +15,8 @@
package org.apache.geode.internal.cache;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -26,6 +28,7 @@ import java.util.Map;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.apache.geode.cache.CacheEvent;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.DistributedCacheOperation.CacheOperationMessage;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
@@ -48,4 +51,62 @@ public class DistributedCacheOperationTest {
assertThat(mockDistributedCacheOperation.supportsDirectAck()).isFalse();
}
+
+ /**
+ * The startOperation and endOperation methods of DistributedCacheOperation record the
+ * beginning and end of distribution of an operation. If startOperation is invoked it
+ * is essential that endOperation be invoked or the state-flush operation will hang.<br>
+ * This test ensures that if distribution of the operation throws an exception then
+ * endOperation is correctly invoked before allowing the exception to escape the startOperation
+ * method.
+ */
+ @Test
+ public void endOperationIsInvokedOnDistributionError() {
+ DistributedRegion region = mock(DistributedRegion.class);
+ CacheDistributionAdvisor advisor = mock(CacheDistributionAdvisor.class);
+ when(region.getDistributionAdvisor()).thenReturn(advisor);
+ TestOperation operation = new TestOperation(null);
+ operation.region = region;
+ try {
+ operation.startOperation();
+ } catch (RuntimeException e) {
+ assertEquals("boom", e.getMessage());
+ }
+ assertTrue(operation.endOperationInvoked);
+ }
+
+ static class TestOperation extends DistributedCacheOperation {
+ boolean endOperationInvoked;
+ DistributedRegion region;
+
+ public TestOperation(CacheEvent event) {
+ super(event);
+ }
+
+ @Override
+ public DistributedRegion getRegion() {
+ return region;
+ }
+
+ @Override
+ public boolean containsRegionContentChange() {
+ return true;
+ }
+
+ @Override
+ public void endOperation(long viewVersion) {
+ endOperationInvoked = true;
+ super.endOperation(viewVersion);
+ }
+
+ @Override
+ protected CacheOperationMessage createMessage() {
+ return null;
+ }
+
+ @Override
+ protected void _distribute() {
+ throw new RuntimeException("boom");
+ }
+ }
}