I went back to 5.1 and this behavior seems to work well by default with the KahaPersistenceAdapter. I assume it is using the Store Based Cursor. I didn't configure a PolicyEntry. Memory usage is capped at 64MB, which is the default cap. I will use 5.1 as my work around for now.
scot.hale wrote: > > Should this work in 5.3.0, or do we need to wait for 5.4.0 to be released? > > > > > > Gary Tully wrote: >> >> Ok. there is a problem and a workaround. see: >> https://issues.apache.org/activemq/browse/AMQ-2610 >> The test case[1] attached to the jira works as expected with the >> workaround. >> The tests is based on the code posted by Scot. >> >> As the file cursor and the queue share the same usage, (they are split >> for >> vmcursors), the key to ensuring the cursor limit kicks in before the >> queue >> memory limit (and producer flow controll) is to configure a cursor, >> policy.setCursorMemoryHighWaterMark(50); which is less than the 70% value >> used by the Queue. This ensures that it will spool messages to disk once >> 50% >> of the system usage is reached. >> >> Have a peek at the test case, [1] >> https://issues.apache.org/activemq/secure/attachment/18932/UnlimitedEnqueueTest.java >> >> >> On 16 February 2010 11:38, Gary Tully <gary.tu...@gmail.com> wrote: >> >>> There is something not right here. let me build a test case to >>> investigate >>> a bit. >>> >>> >>> On 16 February 2010 01:19, scot.hale <scot.h...@gmail.com> wrote: >>> >>>> >>>> A.) I tried using the FilePendingQueueMessageStoragePolicy. I assume >>>> that >>>> this needs to be added to the queue destination policy specifically. >>>> However I added it to default and Topic just to be sure (not shown >>>> here). >>>> >>>> I turned on flow control, but was unable to figure out what memory >>>> settings >>>> are needed. What I gathered from your post is that I need to set the >>>> queue >>>> destination memory limit higher than the default SystemUsage memory >>>> limit. >>>> Is that right? For example: >>>> >>>> >>>> >>>> >>>> brokerService.getSystemUsage().getMemoryUsage().setLimit(32*1024*1024); >>>> >>>> brokerService.getSystemUsage().getTempUsage().setLimit(128*1024*1024); >>>> >>>> PolicyMap policyMap = new PolicyMap(); >>>> >>>> List<PolicyEntry> entries = new ArrayList<PolicyEntry>(); >>>> >>>> PolicyEntry policy = new PolicyEntry(); >>>> policy.setProducerFlowControl(true); >>>> policy.setPendingQueuePolicy(new >>>> FilePendingQueueMessageStoragePolicy()); >>>> policy.setQueue(">"); >>>> policy.setMemoryLimit(64*1024*1024); >>>> entries.add(policy); >>>> policyMap.setPolicyEntries(entries); >>>> >>>> brokerService.setDestinationPolicy(policyMap); >>>> >>>> >>>> I tried it the other way around as well and it still stops (meaning >>>> producers are blocked or resourceAllocationExceptions are thrown from >>>> Queue.send()) when it gets to the lower of the two memory limits. I am >>>> definitely missing something. >>>> >>>> B.) Would using the StorePendingQueueMessageStoragePolicy provide the >>>> same >>>> behavior I am looking for? >>>> >>>> C.) I didn't understand the last sentence in your post. Does this mean >>>> that >>>> when the brokerService.getSystemUsage().getTempUsage() is the disk >>>> usage >>>> limit that should generate ResourceAllocationExceptions (assuming >>>> SendFailIfNoSpace is set to true)? In my configureation, it would mean >>>> when >>>> the 128MB is used up by the temp cursor references on disk, then no >>>> more >>>> resources are available? >>>> >>>> >>>> >>>> >>>> >>>> >>>> Gary Tully wrote: >>>> > >>>> > First thing is you need to use the FilePendingMessageStoragePolicy() >>>> as >>>> > that >>>> > will off load message references to the file system when >>>> > SystemUsage.MemoryUsage limit is reached. >>>> > >>>> > So 1) add the following to the broker policy entry >>>> > PendingQueueMessageStoragePolicy pendingQueuePolicy = new >>>> > FilePendingQueueMessageStoragePolicy(); >>>> > policy.setPendingQueuePolicy(pendingQueuePolicy); >>>> > >>>> > With flow control on, you need to configure a lower SystemUsage as >>>> the >>>> use >>>> > of disk space by the file based cursors is determined by the shared >>>> > SystemUsage.memoryLimit, which by default is the same value as the >>>> memory >>>> > limit for a destination. With a single destination, the flowcontroll >>>> kicks >>>> > in before the system usage so no spooling to disk occurs. >>>> > >>>> > 2) Configure a SystemUsage.MemoryLimit that is less than the default >>>> > destination memory limit of 64M >>>> > brokerService.getSystemUsage().getMemoryUsage().setLimit(1024 * >>>> 1024 >>>> * >>>> > 63); >>>> > >>>> > This should do it once you add a TempStore() limit to implement 5. >>>> > >>>> > >>>> > On 15 February 2010 17:22, scot.hale <scot.h...@gmail.com> wrote: >>>> > >>>> >> >>>> >> I am trying to setup a queue with the following requirements: >>>> >> >>>> >> >>>> >> ActiveMQ 5.1 or 5.3 ( I have been testing with 5.3 ) >>>> >> 1. ) VM Transport >>>> >> 2. ) Persistent with KahaPersistenceAdaptor >>>> >> 4. ) JVM Memory usage is capped at something like 64MB >>>> >> - When this limit is reached the producers should continue to >>>> >> store >>>> >> incoming messages to disk (StoreBasedCursor or FileBasedCursor will >>>> >> work, >>>> >> since the former is the default that is the one I have been using.) >>>> >> 5. ) File System usage is capped at something like 10GB >>>> >> - When this limit is reached the producers should start >>>> throwing >>>> >> javax.jms.ResourceAllocationExceptions to the Producers >>>> >> >>>> >> Number 5 is the least important, as it will be difficult to fill up >>>> disk >>>> >> space in production. My current setup configures ActiveMQ >>>> >> programatically. >>>> >> I don't think this is introducing problems, let me know if there are >>>> >> issues >>>> >> with programatic configuration. >>>> >> >>>> >> >>>> >> Default settings: >>>> >> If I do not configure the SystemUsage or the Flow control, >>>> then >>>> >> 64MB >>>> >> default memory usage is reached and the producers are halted even >>>> though >>>> >> the >>>> >> queues are persistent and have much more space. Should the default >>>> >> StoreBasedCursor behave this way? >>>> >> >>>> >> >>>> >> Turn off Flow Control: >>>> >> When I turn off Flow Control with default SystemUseage >>>> settings, >>>> >> then the >>>> >> JVM memory is not capped. After about 5 million messages with no >>>> >> consumers >>>> >> the JVM assigned 1GB of memory starts returning OutOfMemoryErrors. >>>> >> >>>> >> >>>> >> So what setting do I need to cap the memory and allow the messages >>>> to >>>> be >>>> >> stored to disk even when the cap is reached? >>>> >> >>>> >> >>>> >> >>>> >> This is how I programtically configure my BrokerService >>>> >> >>>> >> System.setProperty("defaultBinSize", "16384");//Only way to >>>> set >>>> >> HashIndex bin size for KahaPersistenceAdapter >>>> >> try { >>>> >> uri = new URI("vm://"+brokerName); >>>> >> } catch (URISyntaxException e) { >>>> >> throw new RuntimeException(e); >>>> >> } >>>> >> brokerService = new BrokerService(); >>>> >> brokerService.setBrokerName(brokerName); >>>> >> brokerService.setUseJmx(true); >>>> >> brokerService.setUseLoggingForShutdownErrors(true); >>>> >> >>>> >> >>>> >> PolicyMap policyMap = new PolicyMap(); >>>> >> List<PolicyEntry> entries = new ArrayList<PolicyEntry>(); >>>> >> PolicyEntry policy = new PolicyEntry(); >>>> >> policy.setProducerFlowControl(true); >>>> >> policy.setQueue(">"); >>>> >> entries.add(policy); >>>> >> policyMap.setPolicyEntries(entries); >>>> >> brokerService.setDestinationPolicy(policyMap); >>>> >> >>>> >> //PERSISTENCE >>>> >> brokerService.setPersistent(true); >>>> >> KahaPersistenceAdapter persistenceAdapter = new >>>> >> KahaPersistenceAdapter(); >>>> >> persistenceAdapter.setDirectory(new >>>> >> File("/tmp/activemq-"+brokerName+"/kaha")); >>>> >> brokerService.setDataDirectoryFile(new >>>> >> File("/tmp/activemq-"+brokerName+"/data")); >>>> >> brokerService.setTmpDataDirectory(new >>>> >> File("/tmp/activemq-"+brokerName+"/temp")); >>>> >> persistenceAdapter.setMaxDataFileLength(500L*1024*1024); >>>> >> >>>> >> try { >>>> >> brokerService.setPersistenceAdapter(persistenceAdapter); >>>> >> } catch (IOException e) { >>>> >> throw new RuntimeException(e); >>>> >> } >>>> >> try { >>>> >> >>>> brokerService.getSystemUsage().setSendFailIfNoSpace(true); >>>> >> brokerService.addConnector(uri); >>>> >> brokerService.start(); >>>> >> } catch (Exception e) { >>>> >> throw new RuntimeException(e); >>>> >> } >>>> >> >>>> >> >>>> >> >>>> >> Here is a Producer: >>>> >> >>>> >> public class Producer implements Runnable{ >>>> >> >>>> >> private BrokerService brokerService; >>>> >> private long numberOfMessages; >>>> >> >>>> >> public Producer(BrokerService brokerService, long n){ >>>> >> this.brokerService = brokerService; >>>> >> this.numberOfMessages = n; >>>> >> } >>>> >> >>>> >> public void run(){ >>>> >> ActiveMQConnectionFactory factory = new >>>> >> ActiveMQConnectionFactory(brokerService.getVmConnectorURI()); >>>> >> try { >>>> >> Connection conn = factory.createConnection(); >>>> >> conn.start(); >>>> >> for (int i = 0; i < numberOfMessages; i++) { >>>> >> Session session = conn.createSession(false, >>>> >> Session.AUTO_ACKNOWLEDGE); >>>> >> Destination destination = >>>> >> session.createQueue("test-queue"); >>>> >> MessageProducer producer = >>>> >> session.createProducer(destination); >>>> >> producer.setDeliveryMode(DeliveryMode.PERSISTENT); >>>> >> BytesMessage message = session.createBytesMessage(); >>>> >> message.writeBytes(new >>>> >> >>>> >> >>>> byte[]{0,0,0,66,0,0,0,5,0,0,0,0,0,0,0,3,0,0,0,49,51,49,51,51,53,53,48,51,51,54,0,0,0,49,50,51,52,53,0,0,0,0,0,0,0,0,0,0,17,116,114,97,99,101,32,109,101,32,112,108,101,97,115,101,32,50,}); >>>> >> try { >>>> >> producer.send(message); >>>> >> } catch (ResourceAllocationException e) { >>>> >> e.printStackTrace(); >>>> >> } >>>> >> session.close(); >>>> >> } >>>> >> } catch (JMSException e) { >>>> >> throw new RuntimeException(e); >>>> >> } >>>> >> >>>> >> } >>>> >> } >>>> >> >>>> >> >>>> >> rajdavies wrote: >>>> >> > >>>> >> > Hi Scott, >>>> >> > >>>> >> > just change the below config to enable flow control - i.e: >>>> >> > >>>> >> > <policyEntry topic=">" producerFlowControl="true" >>>> memoryLimit="1mb"> >>>> >> > <policyEntry queue=">" producerFlowControl="true" >>>> memoryLimit="1mb"> >>>> >> > >>>> >> > in 5.3 - producerFlowControl is on by default - so just remove the >>>> >> > producerFlowControl entry from your configuration. >>>> >> > >>>> >> > If this all sounds double dutch - send in your config - and we'll >>>> help >>>> >> > with the correct settings :) >>>> >> > >>>> >> > >>>> >> > On 12 Feb 2010, at 20:49, scot.hale wrote: >>>> >> > >>>> >> >> >>>> >> >> Fred, >>>> >> >> >>>> >> >> Were you able to configure ActiveMQ to grow without surpassing >>>> the >>>> >> >> memory >>>> >> >> setting? I am trying to figure out how to do the same thing. >>>> >> >> >>>> >> >> -Scot >>>> >> >> >>>> >> >> >>>> >> >> Fred Moore-3 wrote: >>>> >> >>> >>>> >> >>> Hi, >>>> >> >>> >>>> >> >>> going back to Cursors and >>>> >> >>> >>>> >> >>>> http://activemq.apache.org/how-do-i-configure-activemq-to-hold-100s-of-millions-of-queue-messages-.html >>>> >> >>> ... >>>> >> >>> >>>> >> >>> ...can anyone shed some light on the actual role of memoryLimit >>>> in: >>>> >> >>> <policyEntry topic=">" producerFlowControl="false" >>>> >> >>> memoryLimit="1mb"> >>>> >> >>> <policyEntry queue=">" producerFlowControl="false" >>>> >> >>> memoryLimit="1mb"> >>>> >> >>> >>>> >> >>> ...moreover: *when* will producerFlowControl start slowing down >>>> >> >>> consumers? >>>> >> >>> >>>> >> >>> Cheers, >>>> >> >>> F. >>>> >> >>> >>>> >> >>> >>>> >> >> >>>> >> >> -- >>>> >> >> View this message in context: >>>> >> >> >>>> >> >>>> http://old.nabble.com/How-to-configure-5.3-broker-over-KahaDB-to-support-lots-of-unconsumed--persistent-msgs--tp27277849p27569119.html >>>> >> >> Sent from the ActiveMQ - User mailing list archive at Nabble.com. >>>> >> >> >>>> >> > >>>> >> > Rob Davies >>>> >> > http://twitter.com/rajdavies >>>> >> > I work here: http://fusesource.com >>>> >> > My Blog: http://rajdavies.blogspot.com/ >>>> >> > I'm writing this: http://www.manning.com/snyder/ >>>> >> > >>>> >> > >>>> >> > >>>> >> > >>>> >> > >>>> >> > >>>> >> > >>>> >> >>>> >> -- >>>> >> View this message in context: >>>> >> >>>> http://old.nabble.com/How-to-configure-5.3-broker-over-KahaDB-to-support-lots-of-unconsumed--persistent-msgs--tp27277849p27597050.html >>>> >> Sent from the ActiveMQ - User mailing list archive at Nabble.com. >>>> >> >>>> >> >>>> > >>>> > >>>> > -- >>>> > http://blog.garytully.com >>>> > >>>> > Open Source Integration >>>> > http://fusesource.com >>>> > >>>> > >>>> >>>> -- >>>> View this message in context: >>>> http://old.nabble.com/How-to-configure-5.3-broker-over-KahaDB-to-support-lots-of-unconsumed--persistent-msgs--tp27277849p27602503.html >>>> Sent from the ActiveMQ - User mailing list archive at Nabble.com. >>>> >>>> >>> >>> >>> -- >>> http://blog.garytully.com >>> >>> Open Source Integration >>> http://fusesource.com >>> >> >> >> >> -- >> http://blog.garytully.com >> >> Open Source Integration >> http://fusesource.com >> >> > > -- View this message in context: http://old.nabble.com/How-to-configure-5.3-broker-over-KahaDB-to-support-lots-of-unconsumed--persistent-msgs--tp27277849p27617794.html Sent from the ActiveMQ - User mailing list archive at Nabble.com.
