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