[jira] [Updated] (AMQ-4121) Expose Destination.setMemoryUsage so that custom policies can override default MemoryUsage

["Stirling Chow (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/AMQ-4121?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Stirling Chow updated AMQ-4121:
-------------------------------

    Attachment: AMQ4121.patch

Patch file to expose Destination.setMemoryUsage
                
> Expose Destination.setMemoryUsage so that custom policies can override default MemoryUsage
> ------------------------------------------------------------------------------------------
>
>                 Key: AMQ-4121
>                 URL: https://issues.apache.org/jira/browse/AMQ-4121
>             Project: ActiveMQ
>          Issue Type: New Feature
>    Affects Versions: 5.7.0
>            Reporter: Stirling Chow
>         Attachments: AMQ4121.patch
>
>
> Using queues as an example, but this also applies to topics...
> When a queue is created, it inherits the System-wide (broker) usage manager and its limits.  A policy can be applied to the queue in order to specify a destination-specific memory usage limit:
> {code}
> <policyEntry queue=">" producerFlowControl="true" memoryLimit="1mb"> 
> ...
> {code}
> This policy entry is still dependent on the existing org.apache.activemq.usage.MemoryUsage class:
> {code:title=org.apache.activemq.broker.region.policy.PolicyEntry}
> public void configure(Broker broker,Queue queue) {
>     baseConfiguration(broker,queue);
>     if (dispatchPolicy != null) {
>         queue.setDispatchPolicy(dispatchPolicy);
>     }
>     queue.setDeadLetterStrategy(getDeadLetterStrategy());
>     queue.setMessageGroupMapFactory(getMessageGroupMapFactory());
>     if (memoryLimit > 0) {
>         queue.getMemoryUsage().setLimit(memoryLimit);
>     }
> ...
> {code}
> We wanted to create a usage policy that would limit the number of messages in a queue by count (i.e., max 1000 messages) rather than by memory usage.
> The existing usage strategies are memory-centric.  It would be nice if a destination had a generic Usage property rather than a specific MemoryUsage property, but barring refactoring the interfaces, it would be possible to achieve our goal if we could *set* the MemoryUsage property of a queue; this is as simple as adding a Destination.setMemoryUsage(MemoryUsage) property go with the existing Destination.getMemoryUsage(MemoryUsage) property.
> The attached patch contains the trivial changes to expose Destination.setMemoryUsage.  The patch also includes an example of an extended policy that allows a discrete (i.e., count based) usage policy on queues.  Rather than enforcing a specific limit to the number of messages in a queue, this policy determines the limit as a multiple of the number of consumers.  It would be even easier to do the former, but our use case required the latter.  Also note that this consumer-ratio policy is orthogonal to the consumer prefetch limit --- it's a policy that we needed to prevent a distributed queue with many messages from being completely drained to a remote broker during startup (i.e., before additional brokers had started).
> {code:Example of ExtendedPolicy configuration}
> <bean class="com.invoqsystems.foundation.activemq.ExtendedPolicyEntry">
>     <property name="queue" value="shared.notification.*" />
>     <property name="messageToConsumerRatioLimit" value="99" />
> </bean>
> {code}
> {code:com.invoqsystems.foundation.activemq.ExtendedPolicyEntry}
> public class ExtendedPolicyEntry extends PolicyEntry {
>     private long messageToConsumerRatioLimit;
>     /**
>      * This is called by AMQ when a queue is first created. If a message-to-consumer ratio is specified, a
>      * {@link DiscreteMemoryUsage} class with a {@link ConsumerRatioUsageCapacity} limiter replaces the queue's
>      * byte-based MemoryUsage class. The original parent of the queue's byte-based MemoryUsage becomes the parent of the
>      * {@link DiscreteMemoryUsage}, so it will also receive updates and can signal when the queue is full (e.g., because
>      * of queue or system memory limits).
>      */
>     @Override
>     public void configure(Broker broker, Queue queue) {
>         super.configure(broker, queue);
>         if (messageToConsumerRatioLimit > 0) {
>             DiscreteMemoryUsage ratioUsage = new DiscreteMemoryUsage(queue.getMemoryUsage().getParent(), ":ratio");
>             ratioUsage.setLimiter(new ConsumerRatioUsageCapacity(queue));
>             ratioUsage.setLimit(messageToConsumerRatioLimit);
>             ratioUsage.setParent(queue.getMemoryUsage());
>             ratioUsage.setExecutor(queue.getMemoryUsage().getExecutor());
>             queue.setMemoryUsage(ratioUsage);
>         }
>     }
> ...
> {code}

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira