WSO2 CEP - Out of Memory

Question

We observe memory errors on the CEP. A dump of the stream indicates that there were about 32,000 threads on the monitor. In addition, although CEP JVM parameters are set to generate HeapDump on outofmemory, we do not see any heap dumps. Please advise. (CEP JVM -Xms256m -Xmx1536m)

1) Cassandra is disabled on this CEP 2) CEP version is 2.1.0
3) CEP is exposed on the WSO2 ESB (using a BAM intermediary).
4) In addition to sending actual payload data to the CEP, the ESB also sends a periodic beat to the CEP (every 15 eq.).

5) We also configured the JMX Agent on an ESB that monitors CEP every 15 minutes (cpu / memorythreads)
6) No heap dump found, although -XX: parameter HeapDumpPath =

• CEP continued uninterrupted 7 days before this OOM. After the restart, we observe that the number of threads is constantly increasing at a speed of about 4000-5000 threads per day.

CEP Logs ..

[2013-06-10 05:31:49,040] ERROR - Thread Thread[ActiveMQ InactivityMonitor WriteCheckTimer,5,main] died {org.apache.zookeeper.server.NIOServerCnxn} java.lang.OutOfMemoryError: unable to create new native thread at java.lang.Thread.start0(Native Method) at java.lang.Thread.start(Thread.java:640) at java.util.concurrent.ThreadPoolExecutor.addIfUnderMaximumPoolSize(ThreadPoolExecutor.java:727) at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:657) at org.apache.activemq.transport.AbstractInactivityMonitor.writeCheck(AbstractInactivityMonitor.java:153) at org.apache.activemq.transport.AbstractInactivityMonitor$2.run(AbstractInactivityMonitor.java:117) at org.apache.activemq.thread.SchedulerTimerTask.run(SchedulerTimerTask.java:33) at java.util.TimerThread.mainLoop(Timer.java:512) at java.util.TimerThread.run(Timer.java:462) [2013-06-10 05:31:49,040] ERROR - Thread Thread[ActiveMQ InactivityMonitor WriteCheckTimer,5,main] died {org.apache.zookeeper.server.NIOServerCnxn} java.lang.OutOfMemoryError: unable to create new native thread at java.lang.Thread.start0(Native Method) at java.lang.Thread.start(Thread.java:640) at java.util.concurrent.ThreadPoolExecutor.addIfUnderMaximumPoolSize(ThreadPoolExecutor.java:727) at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:657) at org.apache.activemq.transport.AbstractInactivityMonitor.writeCheck(AbstractInactivityMonitor.java:153) at org.apache.activemq.transport.AbstractInactivityMonitor$2.run(AbstractInactivityMonitor.java:117) at org.apache.activemq.thread.SchedulerTimerTask.run(SchedulerTimerTask.java:33) at java.util.TimerThread.mainLoop(Timer.java:512) at java.util.TimerThread.run(Timer.java:462) 

Some of the queries configured in CEP

 <cep:query name="xxxBuildUpQuery"> <cep:expression><![CDATA[from xxxCEPIntgDataStream[interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxx-xxx'or interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxxx-xxx' or interfaceInformationInterfaceName=='xxx-xxx' or interfaceInformationInterfaceName=='xxx-xxx' ]#window.time(60000) insert into buildUpStream interfaceInformationInterfaceName, count(interfaceInformationxxxId) as noOfInflowMsgs group by interfaceInformationInterfaceName]]></cep:expression> <cep:output brokerName="activemqJmsBroker" topic="xxxBuildUpInfoTopic"> <cep:xmlMapping> <xxxAnalytics> <buildUpInfo> <interfaceName>{interfaceInformationInterfaceName}</interfaceName> <buildUpPerMin>{noOfInflowMsgs}</buildUpPerMin> </buildUpInfo> </xxxAnalytics> </cep:xmlMapping> </cep:output> </cep:query> <cep:query name="xxxQueueDepthQuery"> <cep:expression><![CDATA[from xxxIntgrQueueDepthData_v1 insert into xxxIntgrQueueDepthStream flowName,appName, queueDepth]]> </cep:expression> <cep:output brokerName="activemqJmsBroker" topic="xxxIntgrQueueDepthTopic"> <cep:xmlMapping> <xxxAnalytics> <queueDepthInfo> <flowName>{flowName}</flowName> <appName>{appName}</appName> <depth>{queueDepth}</depth> </queueDepthInfo> </xxxAnalytics> </cep:xmlMapping> </cep:output> </cep:query> <cep:query name="xxxClockDataQuery"> <cep:expression><![CDATA[from testStream insert into testOutClockDataStream AEDateTime]]></cep:expression> <cep:output brokerName="activemqJmsBroker" topic="xxxClockDataTopic"> <cep:xmlMapping> <xxxClockFeed> <data> <XXDateTime>{XXDateTime}</XXDateTime> </data> </xxxClockFeed> </cep:xmlMapping> </cep:output> </cep:query> <cep:query name="xxxSimltrPaymntAvgQuery_1"> <cep:expression><![CDATA[from xxxCEPIntgDataStream#window.time(15000) insert into xxxSimltrPymntAvgData avg(amount) as avgAmount, currency group by currency]]></cep:expression> <cep:output brokerName="activemqJmsBroker" topic="xxxAvgPaymntDetails"> <cep:xmlMapping> <xxxAnalytics> <avgPaymentData> <avgAmount>{avgAmount}</avgAmount> <currency>{currency}</currency> </avgPaymentData> </xxxAnalytics> </cep:xmlMapping> </cep:output> 

Thanks Rajiv Patil

Answer 1

It looks like the timeBatch window is a problem, we had several requests using this ... after the stream was removed, it was arrested. The TimeBatch CEP window function probably needs additional testing.

Answer 2

I found that Siddhi Manager initiates a scheduled thread pool with Integer.MAX_VALUE as the primary pool. This means that each request will create a new thread, without a timeout policy. (ref: ThreadPoolExecutor )

Until WSO2 fixes this problem, you can resize the thread pool. Pe, in the class org.wso2.siddhi.core.SiddhiManager, change the line:

this.siddhiContext.setScheduledExecutorService (Executors.newScheduledThreadPool (Integer.MAX_VALUE));

(line 77 in SiddhiManager ver 1.1.0-wso2v1)

to that:

this.siddhiContext.setScheduledExecutorService (Executors.newScheduledThreadPool (100));

This change will create a kernel pool size of 100, a maximum pool size of Integer.MAX_VALUE, and idle threads (based on the size of the main pool) will be deleted as soon as they run out.