Re: Best Strategy to process a large number of rows in File

[Michele]

Hi all,

first of all, thanks so much for your support.

I configured routing with SEDA Component configured like this:

I declared Bean with queueSize
<bean id="myseda" class="org.apache.camel.component.seda.SedaComponent">
     <property name="queueSize" value="150" /> 
</bean>

and then 

<route id="FileReader_Route">
    <from uri="file:incoming?....." />
     <split streaming="true" parallelProcessing="true">
           <tokenize token="\n" />
           <unmarshal ref="IncomingCSVFileDataFormat" />
           <process ref="DataProcessor" />
           <marshal ref="Gson" /> 
           <to uri="myseda:IF_INGESTATE_Inbound?blockWhenFull=true" />
     </split>
</route>

<route id="ProcessTicket_Route">
     <from uri="myseda:IF_INGESTATE_Inbound" />
     <throttle timePeriodMillis="10000" asyncDelayed="true">
     <constant>5</constant>           
           <to
uri="jetty:http://host/rs/v1.0/ticket?jettyHttpBindingRef=CustomJettyHttpBinding";
/>
     </throttle>      
</route>

and everything worked fine and very fast.

What is the meaning of "But now, I have an error java.lang.OutOfMemoryError:
Java heap space on 
ActiveMQ"?

Sorry, but I'm new in using Apache Camel and ActiveMQ... Is it normal that
Active MQ works OutOfMemory?

Here JMS queue configuration:

<bean id="jmsConnectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory">
                <property name="brokerURL" value="tcp://localhost:61616" />
                <property name="userName" value="admin" />
                <property name="password" value="admin" />
                <property name="producerWindowSize" value="1024000" />
        </bean>

        <bean id="pooledConnectionFactory"
class="org.apache.activemq.pool.PooledConnectionFactory"
                init-method="start" destroy-method="stop">
                <property name="maxConnections" value="5" />
                <property name="connectionFactory" ref="jmsConnectionFactory" />
        </bean>

        <bean id="jmsConfigNoTx"
class="org.apache.camel.component.jms.JmsConfiguration">
                <property name="connectionFactory" 
ref="pooledConnectionFactory" />
                <property name="concurrentConsumers" value="3" />
                <property name="transferExchange" value="true" />
        </bean>

        
        <bean id="activemq"
class="org.apache.activemq.camel.component.ActiveMQComponent">
                <property name="configuration" ref="jmsConfigNoTx" />
        </bean>

and then (following these topics
http://activemq.apache.org/javalangoutofmemory.html,
http://activemq.apache.org/how-do-i-configure-10s-of-1000s-of-queues-in-a-single-broker-.html)
I tried to change file activemq.xml in ${JBossFuseInstDir}/etc like this

<destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" producerFlowControl="true">
                  <pendingMessageLimitStrategy>
                    <constantPendingMessageLimitStrategy limit="1000"/>
                  </pendingMessageLimitStrategy>
                </policyEntry>
                <policyEntry queue=">" producerFlowControl="true"
memoryLimit="1024mb" optimizedDispatch="true">
                                        
                                        <pendingQueuePolicy>
                                                <storeCursor />
                                        </pendingQueuePolicy> 
                </policyEntry>
              </policyEntries>
            </policyMap>
        </destinationPolicy>

....
        <persistenceAdapter>

                         <levelDB directory="${data}/leveldb"  />
        </persistenceAdapter>

 <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="80"/>
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="100 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="50 gb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>

and I launch JB Fuse with VM Arguments
-server -Xms256M  -Xss512M  -Xmx512M -XX:+UnlockDiagnosticVMOptions
-XX:+UnsyncloadClass  -XX:PermSize=512M -XX:MaxPermSize=1024M 
-Dorg.apache.activemq.UseDedicatedTaskRunner=false

That's all...

I tried the same process done with Seda with Active MQ and soon I have seen
in the log file

09:23:58,948 | ERROR | Inbound_Worker-2 | DefaultErrorHandler              |
198 - org.apache.camel.camel-core - 2.15.1.redhat-620133 -
IF_INGESTATE-Inbound-Context - IMP-IF-Ingestate-20160330-090900 | Failed
delivery for (MessageId: ID-FGBAL201530-54640-1459321665233-0-2684 on
ExchangeId: ID-FGBAL201530-54640-1459321665233-0-2685). Exhausted after
delivery attempt: 1 caught: org.apache.camel.TypeConversionException: Error
during type conversion from type: java.lang.String to the required type:
byte[] with value [Body is instance of java.io.InputStream] due
java.lang.OutOfMemoryError: Java heap space

and then LevelDB Stop and the process ends...

I attach first OOM excpetion occurred during process  oof.txt
<http://camel.465427.n5.nabble.com/file/n5780003/oof.txt>  

Thanks for all

Best regards

Michele



--
View this message in context: 
http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5780003.html
Sent from the Camel - Users mailing list archive at Nabble.com.