The Qpid 0.6 release introduced a simplistic producer-side flow control mechanism into the Java Messaging Broker, causing producers to be flow-controlled when they attempt to send messages to an overfull queue. Qpid 0.18 introduced a similar mechanism triggered by an overfull persistent message store on a virtual host.
Flow control is enabled on a producer when it sends a message to a Queue which is "overfull". The producer flow control will be rescinded when all Queues on which a producer is blocking become "underfull". A Queue is defined as overfull when the size (in bytes) of the messages on the queue exceeds the "capacity" of the Queue. A Queue becomes "underfull" when its size becomes less than the "flowResumeCapacity".
<queue>
<name>test</name>
<test>
<exchange>amq.direct</exchange>
<capacity>10485760</capacity> <!-- set the queue capacity to 10Mb -->
<flowResumeCapacity>8388608</flowResumeCapacity> <!-- set the resume capacity to 8Mb -->
</test>
</queue>
The default for all queues on a virtual host can also be set
<virtualhosts>
<virtualhost>
<name>localhost</name>
<localhost>
<capacity>10485760</capacity> <!-- set the queue capacity to 10Mb -->
<flowResumeCapacity>8388608</flowResumeCapacity> <!-- set the resume capacity to 8Mb -->
</localhost>
</virtualhost>
</virtualhosts>
Where no flowResumeCapacity is set, the flowResumeCapacity is set to be equal to the capacity. Where no capacity is set, capacity is defaulted to 0 meaning there is no capacity limit.
There are four new Broker log messages that may occur if flow control through queue capacity limits is enabled. Firstly, when a capacity limited queue becomes overfull, a log message similar to the following is produced
MESSAGE [vh(/test)/qu(MyQueue)] [vh(/test)/qu(MyQueue)] QUE-1003 : Overfull : Size : 1,200 bytes, Capacity : 1,000
Then for each channel which becomes blocked upon the overful queue a log message similar to the following is produced:
MESSAGE [con:2(guest@anonymous(713889609)/test)/ch:1] [con:2(guest@anonymous(713889609)/test)/ch:1] CHN-1005 : Flow Control Enforced (Queue MyQueue)
When enough messages have been consumed from the queue that it becomes underfull, then the following log is generated:
MESSAGE [vh(/test)/qu(MyQueue)] [vh(/test)/qu(MyQueue)] QUE-1004 : Underfull : Size : 600 bytes, Resume Capacity : 800
And for every channel which becomes unblocked you will see a message similar to:
MESSAGE [con:2(guest@anonymous(713889609)/test)/ch:1] [con:2(guest@anonymous(713889609)/test)/ch:1] CHN-1006 : Flow Control Removed
Obviously the details of connection, virtual host, queue, size, capacity, etc would depend on the configuration in use.
Since version 0.18 of Qpid Broker, flow control can be triggered when a configured disk quota is exceeded. This is supported by the BDB and Derby message stores.
This functionality blocks all producers on reaching the disk overflow limit. When consumers consume the messages, causing disk space usage to falls below the underflow limit, the producers are unblocked and continue working as normal.
Two limits can be configured:
overfull limit - the maximum space on disk (in bytes) which can be used by store.
underfull limit - when the space on disk drops below this limit, producers are allowed to resume publishing.
An example of quota configuration for the BDB message store is provided below.
<store>
<class>org.apache.qpid.server.store.berkeleydb.BDBMessageStore</class>
<environment-path>${work}/bdbstore/test</environment-path>
<overfull-size>50000000</overfull-size>
<underfull-size>45000000</underfull-size>
</store>
The disk quota functionality is based on "best effort" principle. This means the broker cannot guarantee that the disk space limit will not be exceeded. If several concurrent transactions are started before the limit is reached, which collectively cause the limit to be exceeded, the broker may allow all of them to be committed.
There are 2 new broker log messages that may occur if flow control through disk quota limits is enabled. When the virtual host is blocked due to exceeding of the disk quota limit the following message appears in the broker log
[vh(/test)/ms(BDBMessageStore)] MST-1008 : Store overfull, flow control will be enforced
When virtual host is unblocked after cleaning the disk space the following message appears in the broker log
[vh(/test)/ms(BDBMessageStore)] MST-1009 : Store overfull condition cleared
If a producer sends to a queue which is overfull, the broker will respond by instructing the client not to send any more messages. The impact of this is that any future attempts to send will block until the broker rescinds the flow control order.
While blocking the client will periodically log the fact that it is blocked waiting on flow control.
WARN Message send delayed by 5s due to broker enforced flow control
WARN Message send delayed by 10s due to broker enforced flow control
After a set period the send will timeout and throw a JMSException to the calling code.
If such a JMSException is thrown, the message will not be sent to the broker, however the underlying Session may still be active - in particular if the Session is transactional then the current transaction will not be automatically rolled back. Users may choose to either attempt to resend the message, or to roll back any transactional work and close the Session.
Both the timeout delay and the periodicity of the warning messages can be set using Java system properties.
The amount of time (in milliseconds) to wait before timing out is controlled by the property qpid.flow_control_wait_failure.
The frequency at which the log message informing that the producer is flow controlled is sent is controlled by the system property qpid.flow_control_wait_notify_period.
Adding the following to the command line to start the client would result in a timeout of one minute, with warning messages every ten seconds:
-Dqpid.flow_control_wait_failure=60000
-Dqpid.flow_control_wait_notify_period=10000
The flow control feature was first added to the Java broker/client in the 0.6 release. If an older client connects to the broker then the flow control commands will be ignored by it and it will not be blocked. So to fully benefit from this feature both Client and Broker need to be at least version 0.6.
New in 0.8 is the ability to define configuration for topics. Currently this is limited to configuration for slow consumer detection. This configuration is based on the work designed on the design wiki.
A configuration section has two entries that can be used to identify how the configuration will be applied: 'name' and 'subscriptionName'.
<topic>
<name>stocks.us</name>
<topic>
<subscriptionName>clientid:mysubscription</subscriptionName>
It is also possible to combine these two identifiers to specify a unique subscription to a given topic.
<topic>
<name>stocks.us</name>
<subscriptionName>clientid:mysubscription</subscriptionName>
Currently only one element of the designed configuration is processed, that of the slow consumer detection. This is setup as below using the 'slow-consumer-detection' element. There are two required types of tag, first the trigger, which is one of 'depth', 'messageAge' or 'messageCount' and secondly the 'policy'.
<slow-consumer-detection>
<!-- The maximum depth before which the policy will be applied-->
<depth>4235264</depth>
<!-- The maximum message age before which the policy will be applied-->
<messageAge>600000</messageAge>
<!-- The maximum number of message before which the policy will be applied-->
<messageCount>50</messageCount>
<!-- Policy Selection -->
<policy name="TopicDelete"/>
</slow-consumer-detection>
The trigger is used to determine when the policy should be applied. Currently we have a simple policy 'topicdelete', this will disconnect consumers of topics where their consumption rate falls sufficiently to hit one of the trigger values.
As of 0.8 the topic configuration is limited to straight string matching. This means that given the following two topic configuring sections for 'stocks.us' and 'stocks.*' a subscription for 'stocks.uk' will not match the expected 'stocks.*'. Nor will any additional configuration listed in 'stocks.*' affect any 'stocks.us' subscriptions.
<topics>
<topic>
<name>stocks.us</name>
...
</topic>
<topic>
<name>stocks.*</name>
...
</topic>
</topics>
A subscription for 'stocks.us' will only receive configuration settings that are defined in the 'stocks.us' section.