You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by da...@apache.org on 2016/11/29 18:00:47 UTC
activemq git commit: AMQ-5659: Add safety measure against infinite
loop when store exception prevents message removal. Thanks to metatechbe for
the patch. This fixes #72.
Repository: activemq
Updated Branches:
refs/heads/master b98811358 -> 78492febc
AMQ-5659: Add safety measure against infinite loop when store exception prevents message removal. Thanks to metatechbe for the patch. This fixes #72.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/78492feb
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/78492feb
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/78492feb
Branch: refs/heads/master
Commit: 78492febc858ff06c1ef42e49cdfefc39a6855fb
Parents: b988113
Author: Claus Ibsen <cl...@gmail.com>
Authored: Tue Nov 29 18:58:17 2016 +0100
Committer: Claus Ibsen <cl...@gmail.com>
Committed: Tue Nov 29 18:58:17 2016 +0100
----------------------------------------------------------------------
.../org/apache/activemq/broker/region/Queue.java | 15 +++++++++++++++
1 file changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/78492feb/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
index 6a42ebc..409c978 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java
@@ -1239,6 +1239,8 @@ public class Queue extends BaseDestination implements Task, UsageListener, Index
public void purge() throws Exception {
ConnectionContext c = createConnectionContext();
List<MessageReference> list = null;
+ long previousDequeueCount = -1;
+ long previousDequeueCountRepeated = 1L;
long originalMessageCount = this.destinationStatistics.getMessages().getCount();
do {
doPageIn(true, false, getMaxPageSize()); // signal no expiry processing needed.
@@ -1250,6 +1252,19 @@ public class Queue extends BaseDestination implements Task, UsageListener, Index
}
for (MessageReference ref : list) {
+ long currentDequeueCount = this.destinationStatistics.getDequeues().getCount();
+ if (previousDequeueCount == currentDequeueCount) {
+ previousDequeueCountRepeated++;
+ if (previousDequeueCountRepeated >= 3) {
+ // Break the infinite loop in case the removal fails
+ // 3 times in a row -> error is fatal and not transient.
+ LOG.error("Aborted purge operation after attempting to delete messages failed 3 times in a row (to avoid endless looping)");
+ throw new RuntimeException("Purge operation failed to delete messages failed 3 times in a row (to avoid endless looping)");
+ }
+ } else {
+ previousDequeueCount = currentDequeueCount;
+ previousDequeueCountRepeated = 0L;
+ }
try {
QueueMessageReference r = (QueueMessageReference) ref;
removeMessage(c, r);