Apache/activemq
1
0
Fork 0
mirror of https://github.com/apache/activemq.git synced 2025-02-09 03:25:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-4621

Browse Source 
some minor tweaks to the slow ack abort strategy, wasn't always kicking
in when it should have.
This commit is contained in:
Timothy Bish 2013-09-09 17:48:12 -04:00
parent 8276676469
commit 272de3a146
1 changed files with 22 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/AbortSlowAckConsumerStrategy.java
View File

@ -107,6 +107,10 @@ public class AbortSlowAckConsumerStrategy extends AbortSlowConsumerStrategy {
// Not considered Idle so ensure its cleared from the list // Not considered Idle so ensure its cleared from the list
if (slowConsumers.remove(subscriber) != null) { if (slowConsumers.remove(subscriber) != null) {
LOG.info("sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId()); LOG.info("sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId());
} else {
if (LOG.isTraceEnabled()) {
LOG.trace("Not ignoring idle Consumer {}", subscriber.getConsumerInfo().getConsumerId());
}
} }
} }
@ -133,13 +137,25 @@ public class AbortSlowAckConsumerStrategy extends AbortSlowConsumerStrategy {
private void abortAllQualifiedSlowConsumers() { private void abortAllQualifiedSlowConsumers() {
HashMap<Subscription, SlowConsumerEntry> toAbort = new HashMap<Subscription, SlowConsumerEntry>(); HashMap<Subscription, SlowConsumerEntry> toAbort = new HashMap<Subscription, SlowConsumerEntry>();
for (Entry<Subscription, SlowConsumerEntry> entry : slowConsumers.entrySet()) { for (Entry<Subscription, SlowConsumerEntry> entry : slowConsumers.entrySet()) {
if (entry.getKey().isSlowConsumer()) {
if (getMaxSlowDuration() > 0 && if (getMaxSlowDuration() > 0 &&
(entry.getValue().markCount * getCheckPeriod() > getMaxSlowDuration()) || (entry.getValue().markCount * getCheckPeriod() > getMaxSlowDuration()) ||
getMaxSlowCount() > 0 && entry.getValue().slowCount > getMaxSlowCount()) { getMaxSlowCount() > 0 && entry.getValue().slowCount > getMaxSlowCount()) {
if (LOG.isTraceEnabled()) {
LOG.trace("Remove consumer {} from slow list: " +
"slow duration = " + entry.getValue().markCount * getCheckPeriod() + ", " +
"slow count = " + entry.getValue().slowCount,
entry.getKey().getConsumerInfo().getConsumerId());
}
toAbort.put(entry.getKey(), entry.getValue()); toAbort.put(entry.getKey(), entry.getValue());
slowConsumers.remove(entry.getKey()); slowConsumers.remove(entry.getKey());
} else {
if (LOG.isTraceEnabled()) {
LOG.trace("Not yet time to abot consumer {}: " +
"slow duration = " + entry.getValue().markCount * getCheckPeriod() + ", " +
"slow count = " + entry.getValue().slowCount,
entry.getKey().getConsumerInfo().getConsumerId());
} }
} }
} }