Page History

...

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.

...

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

...

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.

...

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. This will allow slow consumers to be checked and disconnected on a periodic basis.

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 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 delete 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.'

...

Error Processing
Currently a 0-8/9/91 Session will propogate propagate 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.

...

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

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

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

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

This <consumer> element will be added to the existing queue configuration to allow specific durable subscriptions to be identified and processed. In addition a new <topic> element will be added to allow configuration for topics. The resulting section of xml would look like this:

    <topic key="stocks.us.*">
        <consumer>
            <!-- The depth before which the policy will be applied-->
            <depth>4235264</depth>

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

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

            <!-- Policies configuration -->
            <policy name="Delete">
                <options>
                    <option name="includedelete-persistent" value="true"/>
                </options>
            </policy>
        </consumer>

    </topic>

Testing Spec

Testing for this new feature will mainly rely on system testing.

...

Testing should be completed at a minimum on the two sets of protocol 0-8/0-9/0-91 and 0-10. Ideally the test would be run on each protocol version to verify the protocol exception is correctly propogatedpropagated.

Client Ack Mode

The tests should be run against each client ack mode to validate if there is any difference in the exception handling. Transacted for instance should fail to commit by throwing the expected exception as well as having the exception appear on the ExceptionListener.

...