Reading from mqueue in a controlled manner

[Anonymous]

Hi All

In a mqueue we are receiving continuous stream of data. I need to unload the data into a file. 

I have created a batch job to unload the data into a file and continuous job to call the batch job every 2 mins.

Now there could be a situation since the data is coming into the mqueue continuously the unload process might never end.

Is there a way to just unload messages in a controlled manner?

If not how can i handle this scenario more effectively?

For example: if there are 100 K messages

unload 20K messages ---> Sleep for 2 mins ---> unload 20K messages --> Sleep for 2 mins ........etc....

[vapukov]

Hi,

which MQ do you use?

limit number of received messages is possible for ActiveMQ

[Anonymous]

We are using WebShphere MQ.

Is there anyway we can design our process to read limited number of messages? 

I was thinking to create multiple files with limited messages in each file but not very lucky..

[Anonymous]

Hi,

    You need to use a Route for your use case as it is the best way to call a Talend job based on the input data.

    Please refer the details below to read about Routes.

https://help.talend.com/reader/TKUQ4WRBbYZRnl9OyAgr5w/sKkTT1kP7~AWAEv2EsZFhw

Warm Regards,
Nikhil Thampi

Please appreciate our Talend community members by giving Kudos for sharing their time for your query. If your query is answered, please mark the topic as resolved 🙂

[vapukov]

Hi Nikhil,

not sure - is it correct way in this case or not?

- first of all current task based not on input data, but on input data count

- second, author want write messages content into separate files up to 10k (an example) messages per file, but not wait till tomorrow if only 9k of messages ready right now

for ActiveMQ it is simple - component has both parameters - number of messages for consuming and timeout for wait before end 

with WebsphereMQ it is only 1 parameter - timeout

in theory, it is possible to count processed messages and finishing job (tDie) after receiving the expected number of messages but needs to test

but if there more efficient way - would be good to share it!

[Anonymous]

@vapukov 

I agree with your thoughts.

I was thinking to avoid the scenario where we are trying to make a batch job to do the work of a polling operation indirectly.

But the option outlined should work in theory. 

Warm Regards,
Nikhil Thampi

Please appreciate our Talend community members by giving Kudos for sharing their time for your query. If your query is answered, please mark the topic as resolved 🙂

[Anonymous]

Thanks for the suggestions.

Actually I have created a counter and for every 10000 messages I and increasing the counter by one and creating a global variable.

So my file name is like this and after reaching 10000 messages it's creating a new file until the complete mqueue is drained.

By another job will continuously looks for these files and will trigger the load job to DB tables.

Let me know if you see any issue with this design.

context.OUTPUT_DIR+"/"+"fal_mqueue_extract."+(Integer)globalMap.get("seq")+"."+((String)globalMap.get("curr_time"))

[Anonymous]

Hi,

    Could you please use a tDie command to come out of the flow once you get 10k messages?

    You can modify the schedule to start the job again.

     Could you please also try Memory run of the job (under Run tab in Studio) to analyse whether there is any memory related issues?

Warm Regards,
Nikhil Thampi

Please appreciate our Talend community members by giving Kudos for sharing their time for your query. If your query is answered, please mark the topic as resolved 🙂

[David_Beaty]

Hi,

I previously had this issue with Websphere MQ, when you look at the generated code within Talend, the looping mechanism checks  the queue depth every time around, so a constant stream of data in the queue never completes. In the end I copy/created a new custom component that checked the queue depth once at the start of the component and only looped that number of times before exiting.

It was at a previous company where I don't have access to the code base, but that was essentially how I resolved it.