Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Update the consumer / connection abort process such that when the
strategy is configured to abort the connection is only attempt to do so
once instead of once for every subscription in the map.  Also improve
logging to better indicate the subscription being aborted and the
destination that the subscription was on.
This commit is contained in:
Timothy Bish 2013-09-12 15:07:13 -04:00
parent 272b846b0c
commit cdb7bb11ff
1 changed files with 64 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/AbortSlowConsumerStrategy.java
View File

@ -16,7 +16,9 @@
*/
package org.apache.activemq.broker.region.policy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
@ -106,38 +108,73 @@ public class AbortSlowConsumerStrategy implements SlowConsumerStrategy, Runnable
}
protected void abortSubscription(Map<Subscription, SlowConsumerEntry> toAbort, boolean abortSubscriberConnection) {
Map<Connection, List<Subscription>> abortMap = new HashMap<Connection, List<Subscription>>();
for (final Entry<Subscription, SlowConsumerEntry> entry : toAbort.entrySet()) {
ConnectionContext connectionContext = entry.getValue().context;
if (connectionContext!= null) {
try {
LOG.info("aborting "
+ (abortSubscriberConnection ? "connection" : "consumer")
+ ", slow consumer: " + entry.getKey());
if (connectionContext == null) {
continue;
}
final Connection connection = connectionContext.getConnection();
if (connection != null) {
if (abortSubscriberConnection) {
scheduler.executeAfterDelay(new Runnable() {
@Override
public void run() {
connection.serviceException(new InactivityIOException("Consumer was slow too often (>"
+ maxSlowCount + ") or too long (>"
+ maxSlowDuration + "): " + entry.getKey().getConsumerInfo().getConsumerId()));
}}, 0l);
} else {
// just abort the consumer by telling it to stop
ConsumerControl stopConsumer = new ConsumerControl();
stopConsumer.setConsumerId(entry.getKey().getConsumerInfo().getConsumerId());
stopConsumer.setClose(true);
connection.dispatchAsync(stopConsumer);
}
} else {
LOG.debug("slowConsumer abort ignored, no connection in context:" + connectionContext);
Connection connection = connectionContext.getConnection();
if (connection == null) {
LOG.debug("slowConsumer abort ignored, no connection in context:" + connectionContext);
}
if (!abortMap.containsKey(connection)) {
abortMap.put(connection, new ArrayList<Subscription>());
}
abortMap.get(connection).add(entry.getKey());
}
for (Entry<Connection, List<Subscription>> entry : abortMap.entrySet()) {
final Connection connection = entry.getKey();
final List<Subscription> subscriptions = entry.getValue();
if (abortSubscriberConnection) {
LOG.info("aborting connection:{} with {} slow consumers",
connection.getConnectionId(), subscriptions.size());
if (LOG.isTraceEnabled()) {
for (Subscription subscription : subscriptions) {
LOG.trace("Connection {} being aborted because of slow consumer: {} on destination: {}",
new Object[] { connection.getConnectionId(),
subscription.getConsumerInfo().getConsumerId(),
subscription.getActiveMQDestination() });
}
}
try {
scheduler.executeAfterDelay(new Runnable() {
@Override
public void run() {
connection.serviceException(new InactivityIOException(
subscriptions.size() + " Consumers was slow too often (>"
+ maxSlowCount + ") or too long (>"
+ maxSlowDuration + "): "));
}}, 0l);
} catch (Exception e) {
LOG.info("exception on stopping "
+ (abortSubscriberConnection ? "connection" : "consumer")
+ " to abort slow consumer: " + entry.getKey(), e);
LOG.info("exception on aborting connection {} with {} slow consumers",
connection.getConnectionId(), subscriptions.size());
}
} else {
// just abort each consumer by telling it to stop
for (Subscription subscription : subscriptions) {
LOG.info("aborting slow consumer: {} for destination:{}",
subscription.getConsumerInfo().getConsumerId(),
subscription.getActiveMQDestination());
try {
ConsumerControl stopConsumer = new ConsumerControl();
stopConsumer.setConsumerId(subscription.getConsumerInfo().getConsumerId());
stopConsumer.setClose(true);
connection.dispatchAsync(stopConsumer);
} catch (Exception e) {
LOG.info("exception on aborting slow consumer: {}", subscription.getConsumerInfo().getConsumerId());
}
}
}
}