001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.broker.region.policy;
018
019import java.util.*;
020import java.util.Map.Entry;
021import java.util.concurrent.ConcurrentHashMap;
022
023import org.apache.activemq.broker.Broker;
024import org.apache.activemq.broker.ConnectionContext;
025import org.apache.activemq.broker.region.AbstractSubscription;
026import org.apache.activemq.broker.region.Destination;
027import org.apache.activemq.broker.region.Subscription;
028import org.slf4j.Logger;
029import org.slf4j.LoggerFactory;
030
031/**
032 * Abort slow consumers when they reach the configured threshold of slowness,
033 *
034 * default is that a consumer that has not Ack'd a message for 30 seconds is slow.
035 *
036 * @org.apache.xbean.XBean
037 */
038public class AbortSlowAckConsumerStrategy extends AbortSlowConsumerStrategy {
039
040    private static final Logger LOG = LoggerFactory.getLogger(AbortSlowAckConsumerStrategy.class);
041
042    private final Map<String, Destination> destinations = new ConcurrentHashMap<String, Destination>();
043    private long maxTimeSinceLastAck = 30*1000;
044    private boolean ignoreIdleConsumers = true;
045
046    public AbortSlowAckConsumerStrategy() {
047        this.name = "AbortSlowAckConsumerStrategy@" + hashCode();
048    }
049
050    @Override
051    public void setBrokerService(Broker broker) {
052        super.setBrokerService(broker);
053
054        // Task starts right away since we may not receive any slow consumer events.
055        if (taskStarted.compareAndSet(false, true)) {
056            scheduler.executePeriodically(this, getCheckPeriod());
057        }
058    }
059
060    @Override
061    public void slowConsumer(ConnectionContext context, Subscription subs) {
062        // Ignore these events, we just look at time since last Ack.
063    }
064
065    @Override
066    public void run() {
067
068        if (maxTimeSinceLastAck < 0) {
069            // nothing to do
070            LOG.info("no limit set, slowConsumer strategy has nothing to do");
071            return;
072        }
073
074
075        List<Subscription> subscribersDestroyed  = new LinkedList<Subscription>();
076        // check for removed consumers also
077        for (Map.Entry<Subscription, SlowConsumerEntry> entry : slowConsumers.entrySet()) {
078            if (getMaxSlowDuration() > 0) {
079                // For subscriptions that are already slow we mark them again and check below if
080                // they've exceeded their configured lifetime.
081                entry.getValue().mark();
082            }
083            if (!entry.getKey().isSlowConsumer()) {
084                subscribersDestroyed.add(entry.getKey());
085            }
086        }
087
088        for (Subscription subscription: subscribersDestroyed) {
089            slowConsumers.remove(subscription);
090        }
091
092        List<Destination> disposed = new ArrayList<Destination>();
093
094        for (Destination destination : destinations.values()) {
095            if (destination.isDisposed()) {
096                disposed.add(destination);
097                continue;
098            }
099
100            // Not explicitly documented but this returns a stable copy.
101            List<Subscription> subscribers = destination.getConsumers();
102
103            updateSlowConsumersList(subscribers);
104        }
105
106        // Clean up an disposed destinations to save space.
107        for (Destination destination : disposed) {
108            destinations.remove(destination.getName());
109        }
110
111        abortAllQualifiedSlowConsumers();
112    }
113
114    private void updateSlowConsumersList(List<Subscription> subscribers) {
115        for (Subscription subscriber : subscribers) {
116            if (isIgnoreNetworkSubscriptions() && subscriber.getConsumerInfo().isNetworkSubscription()) {
117                if (slowConsumers.remove(subscriber) != null) {
118                    LOG.info("network sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId());
119                }
120                continue;
121            }
122
123            if (isIgnoreIdleConsumers() && subscriber.getDispatchedQueueSize() == 0) {
124                // Not considered Idle so ensure its cleared from the list
125                if (slowConsumers.remove(subscriber) != null) {
126                    LOG.info("idle sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId());
127                }
128                continue;
129            }
130
131            long lastAckTime = subscriber.getTimeOfLastMessageAck();
132            long timeDelta = System.currentTimeMillis() - lastAckTime;
133
134            if (timeDelta > maxTimeSinceLastAck) {
135                if (!slowConsumers.containsKey(subscriber)) {
136                    LOG.debug("sub: {} is now slow", subscriber.getConsumerInfo().getConsumerId());
137                    SlowConsumerEntry entry = new SlowConsumerEntry(subscriber.getContext());
138                    entry.mark(); // mark consumer on first run
139                    if (subscriber instanceof AbstractSubscription) {
140                        AbstractSubscription abstractSubscription = (AbstractSubscription) subscriber;
141                        if (!abstractSubscription.isSlowConsumer()) {
142                            abstractSubscription.setSlowConsumer(true);
143                            for (Destination destination: abstractSubscription.getDestinations()) {
144                               destination.slowConsumer(broker.getAdminConnectionContext(), abstractSubscription);
145                            }
146                        }
147                    }
148                    slowConsumers.put(subscriber, entry);
149                } else if (getMaxSlowCount() > 0) {
150                    slowConsumers.get(subscriber).slow();
151                }
152            } else {
153                if (slowConsumers.remove(subscriber) != null) {
154                    LOG.info("sub: {} is no longer slow", subscriber.getConsumerInfo().getConsumerId());
155                }
156            }
157        }
158    }
159
160    private void abortAllQualifiedSlowConsumers() {
161        HashMap<Subscription, SlowConsumerEntry> toAbort = new HashMap<Subscription, SlowConsumerEntry>();
162        for (Entry<Subscription, SlowConsumerEntry> entry : slowConsumers.entrySet()) {
163            if (getMaxSlowDuration() > 0 && (entry.getValue().markCount * getCheckPeriod() >= getMaxSlowDuration()) ||
164                getMaxSlowCount() > 0 && entry.getValue().slowCount >= getMaxSlowCount()) {
165
166                LOG.trace("Transferring consumer{} to the abort list: {} slow duration = {}, slow count = {}",
167                        new Object[]{ entry.getKey().getConsumerInfo().getConsumerId(),
168                        entry.getValue().markCount * getCheckPeriod(),
169                        entry.getValue().getSlowCount() });
170
171                toAbort.put(entry.getKey(), entry.getValue());
172                slowConsumers.remove(entry.getKey());
173            } else {
174
175                LOG.trace("Not yet time to abort consumer {}: slow duration = {}, slow count = {}", new Object[]{ entry.getKey().getConsumerInfo().getConsumerId(), entry.getValue().markCount * getCheckPeriod(), entry.getValue().slowCount });
176
177            }
178        }
179
180        // Now if any subscriptions made it into the aborts list we can kick them.
181        abortSubscription(toAbort, isAbortConnection());
182    }
183
184    @Override
185    public void addDestination(Destination destination) {
186        this.destinations.put(destination.getName(), destination);
187    }
188
189    /**
190     * Gets the maximum time since last Ack before a subscription is considered to be slow.
191     *
192     * @return the maximum time since last Ack before the consumer is considered to be slow.
193     */
194    public long getMaxTimeSinceLastAck() {
195        return maxTimeSinceLastAck;
196    }
197
198    /**
199     * Sets the maximum time since last Ack before a subscription is considered to be slow.
200     *
201     * @param maxTimeSinceLastAck
202     *      the maximum time since last Ack (mills) before the consumer is considered to be slow.
203     */
204    public void setMaxTimeSinceLastAck(long maxTimeSinceLastAck) {
205        this.maxTimeSinceLastAck = maxTimeSinceLastAck;
206    }
207
208    /**
209     * Returns whether the strategy is configured to ignore consumers that are simply idle, i.e
210     * consumers that have no pending acks (dispatch queue is empty).
211     *
212     * @return true if the strategy will ignore idle consumer when looking for slow consumers.
213     */
214    public boolean isIgnoreIdleConsumers() {
215        return ignoreIdleConsumers;
216    }
217
218    /**
219     * Sets whether the strategy is configured to ignore consumers that are simply idle, i.e
220     * consumers that have no pending acks (dispatch queue is empty).
221     *
222     * When configured to not ignore idle consumers this strategy acks not only on consumers
223     * that are actually slow but also on any consumer that has not received any messages for
224     * the maxTimeSinceLastAck.  This allows for a way to evict idle consumers while also
225     * aborting slow consumers.
226     *
227     * @param ignoreIdleConsumers
228     *      Should this strategy ignore idle consumers or consider all consumers when checking
229     *      the last ack time verses the maxTimeSinceLastAck value.
230     */
231    public void setIgnoreIdleConsumers(boolean ignoreIdleConsumers) {
232        this.ignoreIdleConsumers = ignoreIdleConsumers;
233    }
234}