Page History

Slow Consumer Problem Statement

The problem with slow consumers is that the broker must act as a buffer until they can catch up. However the broker does not have infinite resouces resources so it will fail if the consumer does not catch up.Work has already been done to help protect the broker from these situations by limiting producers on queues: Producer flow control.

Consumers in the Java Broker

There are three types of consumers that we need to investigate. Each one corresponds to a different use of a Queue. In addition to these three types we also have a number of Acknowledge modes that change the behaviour of the client and so also need special consideration.

The tree types of queue we have to consider are:

• JMS Durable Subscriptions: Durable queues bound to amq.topic
• JMS Queues: Queues bound to the amq.direct.
• JMS Topics: Temporary queues bound to amq.topic

JMS Durable Subscriptions

In the case of durable subscriptions there is nothing simple we can do. The JMS spec requires the broker to keep all the messages. If a consumer here is disconnected and we are unable to hold all their messages then we need alternative strategies similar to the C++ broker: hard limit queue then drop/dead-letter, cyclic queue oldest are deleted.

JMS Queues

In the case of durable queues we are constrained just as we are for durable subscriptions. In the transient case we could perform the same limiting strategies as for the durable case however as the queue is transient we could be more aggressive in our cleanup. For example if we disconnected the slow consumer and had no consumers on the queue we could purge all the messages. This however would result in unexpected behaviour for our users who expect defined queue to hold their data.

JMS Topics / Temporary queues

The types of queues where a slow consumer can occur boils down to two properties: durability and bound exchange.

Queues that are bound to the amq.direct exchange, i.e. JMS Queues, are not going to be included in this work.
Queues bound to other exchanges such as amq.match, the Headers exchange will also not be included in this work.

This reduces the queues to consider to just queues bound to the topic exchange.

Topics

In AMQP consumption is always from an AMQP Queue to avoid confusion with JMS Queues in the following discussion the term topic is defined to mean an AMQP Queue bound to the amq.topic exchange.

When a topic reaches a set threshold for message count, size or age the attached consumer session we have three options.

• Flow the producer.
• Disconnect the slow consumer.
• Discard messages for the slow consumer.

Work has already been done to flow producers on queues: Producer flow control.

This leaves us with two options.

Disconnect the Slow Consumer

For non-durable topics this means that it will be deleted so potentially freeing up the memory used by the messages. Remember the messages are shared across all topics so the memory will only be freed up when all the topics no longer require the message.

For durable topics (JMS Durable Subscriptions) disconnecting the consumer will leave the queue bound and receiving messages. This will only make the memory situation worse as we now have a queue with no consumer rather than just a slow consumer. If all the messages on the topic are persistent then they can be evicted from memory if required but there is no guarantee that all the messages will have been sent persistently.

On disconnection the consumer would receive an AMQP error, 506 Resource Limit Exceeded/Resource Error. For non-durable consumers this will always work. However, for a durable subscription it is possible that the consumer has disconnected when the limit is reached. So whilst deleting their subscription would be in line with the configuration it would not be expected by the user. The configuration for enabling slow consumer disconnection should allow for durable subscriptions to be maintained, targeting only transient subscriptions for disconnection.

Discard messages

The C++ broker implements a type of queue called a Ring Queue that will delete the oldest data to make room for new messages. This approach could be applied to both durable and non-durable topics and would bound the memory used. No notification to the client would be possible however as messages were deleted logging should be performed so that broker administrators can tune the topic size or inform the client that they need to consume faster.

Design Specification

This work is mainly focused on the broker however the the client may also require changes to ensure that the error is correctly reported.

Broker Changes

Extension Point
To enable the broker to monitor the queues and perform the appropriate action we can extend a existing mechanism. That of the VirtualHost housekeeping thread. This is a thread that checks all the queues for alerting purposes. Currently this is done via a single TimerTask however by updating this to utilise a ScheduledThreadPoolExecutor we can run arbitrary processes in the pool and ensure that any error in their operation does not prevent them from running on their defined schedule.

Queue Detection
The target queues can easily be identified by checking their bindings. Topics are all bound to the TopicExchange. Once we have identified a topic exchange we can use the queue assigned configuration to determine if we are checking depth, messageCount or messageAge as a means of selecting the subscription for processing.

Processing
When we have our identified queue/subscription we have two policy options. In both situations we will identify the session/channel that the subscription is on and close it with the appropriate error code. The first policy is 'Delete' which will then ensure that the queue is deleted and all messages released after the session/channel has been closed. The second policy is 'Cycle' which will limit the queue to the given size. This means that the oldest messages, i.e. the ones at the front, will be purged as new messages arrive.

Error Code
The AMQP error code 506 will be used to communicate the failure to the client. This is defined as a Resource Error or Resource Limit Exceeded in 0-8/9/91 and 0-10 respectively. In addition the protocol allows for a textual description to be sent back to the client. In this field we will send 'Consuming too slow.'

Client Changes

Error Processing
Currently a 0-8/9/91 Session will propogate a ChannelCloseException to the client via the JMS ExceptionListener we need to ensure that the 0-10 code path will create the same Exception type and present it to the JMS ExceptionListener.

After Effects
The Exception that is thrown should not be classed as a 'HardError' which would result in Failover starting. After the exception has been received the Consumer and the associated Session should be closed however the Connection will still be operational. This will allow the client to perform recovery without having to reestablish its Connection.

Configuration

Picking up on the Topic Configuration Design the addition of slow consumer configuration would be done using a 'consumer' element.

The topic currently exposes three properties that we can use to control the client, depth, oldest message, and count. The configuration will provide the option to one or all of these values to apply to the specified topic. In the situation where more than one value is specified they will all be used to trigger the policy. e.g. setting count to 10 and depth to 1024 would allow the 10 messages to exist as long as their total size was not more than 1024.

One additional property that would be of use here would be the consumption rate. If the topic reported the consumption rate this property could be used to define a threshold that the consumer must stay above.

        <consumer>
            <!-- The maximum depth before which the policy will be applied-->
            <maximumDepth>4235264</maximumDepth>

            <!-- The maximum message age before which the policy will be applied-->
            <maximumMessageAge>600000</maximumMessageAge>

            <!-- The maximum number of message before which the policy will be applied-->
            <maximumMessageCount>50</maximumMessageCount>

            <!-- Available Policies : Delete | Cycle -->
            <policy name="Delete">
                <options>
                    <option name="include-persistent" value="true"/>
                </options>
            </policy>
        </consumer>

Testing Spec

– Draft –

Protocol Version : 0-8/0-9/0-91/0-10
Client AckMode : Auto/Client/Dups/Transacted, NoAck - NoAck case has addition issue in that it can overwhelm IO layer in presence of a slow consumer.
Client Consume Mode : Async/Sync
Topic Type : Durable/Non-DurableThe topic case is a simpler case to tackle at a first pass. The queue that is used is an exlusive, auto-delete transient queue. This means that the user is expecting the messages to be lost if they disconnect, so by detecting these queues that have reached a set threshold for message count, size or age the attached consumer session can be closed.