Fork join pattern for parallel processing mule esb throwing NullPointer exception

Question

I am using fork-join pattern to achieve parallel processing in Batch.

I have take reference of the following question : Mule File Inbound Flow : Control Number of threads

Since I do have too many files in my input folder, but i need to achieve parallel processing. Hence, thought of using this pattern. Here is my config flow how it looks.

            <quartz:connector name="Quartz1" validateConnections="true" doc:name="Quartz">
                    <receiver-threading-profile maxThreadsActive="1"/>
           </quartz:connector>
            <flow name="Mainflow" processingStrategy="synchronous">
          <quartz:inbound-endpoint jobName="EventGeneration" repeatInterval="1000" connector-ref="Quartz1" responseTimeout="10000" doc:name="Quartz">
                    <quartz:event-generator-job/>
                </quartz:inbound-endpoint>

                <mulerequester:request-collection config-ref="Mule_Requester" resource="file:///FileLocation?connector=FileMRTransformer" count="3" doc:name="Mule Requester"/>
                <expression-filter expression="#[payload.size() != 0]" doc:name="Expression"/>
        <request-reply doc:name="Request-Reply" timeout="300000">
                    <processor-chain doc:name="Processor Chain">
                        <collection-splitter doc:name="Collection Splitter"/>
                        <vm:outbound-endpoint exchange-pattern="one-way"  doc:name="VM" connector-ref="VM" path="Batchinput" />
                    </processor-chain>
                    <vm:inbound-endpoint exchange-pattern="one-way"  doc:name="VM" connector-ref="VM" path="Batchoutput">
                        <message-properties-transformer>
                            <add-message-property key="MULE_CORRELATION_GROUP_SIZE" value="3" />
                        </message-properties-transformer>
                        <collection-aggregator />
                    </vm:inbound-endpoint>
    </request-reply>
</flow>

<batch:job name="BatchDemo" max-failed-records="-1">
        <batch:input>
            <vm:inbound-endpoint exchange-pattern="one-way" path="Batchinput" connector-ref="VM" doc:name="VM"/>
....
required processing.....
.
.
<batch:on-complete>
 <vm:outbound-endpoint exchange-pattern="one-way"  doc:name="VM" connector-ref="VM" path="Batchoutput"/>
</batch:on-complete>

As soon as the control enter the request-reply scope the following exception is thrown:

ERROR 2016-06-23 10:32:56,190 [scheduler-multithreadint019.1.2_productindexing_hybris_fh_Worker-1] org.mule.exception.CatchMessagingExceptionStrategy: 
********************************************************************************
Message               : null (java.lang.NullPointerException). Message payload is of type: CopyOnWriteArrayList
Type                  : org.mule.api.MessagingException
Code                  : MULE_ERROR--2
Payload               : [[B@26589e4d, [B@400e4e6, [B@56b3ba17]
JavaDoc               : http://www.mulesoft.org/docs/site/current3/apidocs/org/mule/api/MessagingException.html
********************************************************************************
Exception stack is:
1. null (java.lang.NullPointerException)
  java.util.concurrent.ConcurrentHashMap:-1 (null)
2. null (java.lang.NullPointerException). Message payload is of type: CopyOnWriteArrayList (org.mule.api.MessagingException)
  org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor:32 (http://www.mulesoft.org/docs/site/current3/apidocs/org/mule/api/MessagingException.html)
********************************************************************************
Root Exception stack trace:
java.lang.NullPointerException
    at java.util.concurrent.ConcurrentHashMap.hash(Unknown Source)
    at java.util.concurrent.ConcurrentHashMap.put(Unknown Source)
    at org.mule.routing.requestreply.AbstractAsyncRequestReplyRequester.process(AbstractAsyncRequestReplyRequester.java:85)
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:24)
    at org.mule.execution.MessageProcessorNotificationExecutionInterceptor.execute(MessageProcessorNotificationExecutionInterceptor.java:107)

Alternate config tried for request part in Request-Reply processor:

 <vm:outbound-endpoint exchange-pattern="one-way"  doc:name="VM" connector-ref="VM" path="Batchinput">
                <collection-splitter />
            </vm:outbound-endpoint>

But it resulted the same exception.

Only when using MuleRequester i am getting this exception. If i use some java snippet to return a collection of File objects i am not getting this exception, and the control is entering as expected into the batch flow. But, i do have a transformer(DataWeave) in my input phase of batch, and my transformer is not able to parse this file object (say java.io.file or java.io.FileInputStream). Hence used MuleRequester so that i can have streaming enabled.

I am not sure what went wrong when using this Mulerequester??

masetti mani kanta masetti mani kanta · Accepted Answer · 2016-07-04T10:45:32

Found an alternative for it, if anyone is interested.

Used a poll component, with Java to return fileNames collection (rather than file Objects). Used the same requestreply pattern. However, this time i am retrieving payload from file in the input phase of batch, using MuleRequester. giving the filename to this MuleRequester as input path.

This way it is running fine.

Fork join pattern for parallel processing mule esb throwing NullPointer exception

1 Answers