You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@nifi.apache.org by "McCullough, Alex" <Al...@capitalone.com> on 2015/11/05 15:49:57 UTC
Asynchronous JMS Consumer for IBM MQ
Hey There,
This is a bit of a generic question, and I am not sure if this is the right place to start but I am new to developing custom processors in Nifi but have been trying to write an asynchronous message consumer to consume from IBM MQ.
I have read all of the developer documentation and have been testing a processor I built, but I think where I am struggling is conceptualizing exactly how I should set this up within the framework and how the threading within Nifi works exactly.
Typically I am used to registering a listener and using the onMessage method to add messages that are received to local queue to then be processed. What would this look like within the Nifi framework? Are there any guiding principals you can point me towards or any source code that may be doing something somewhat similar?
Sorry this is a pretty vague question and I’m sure there are many ways to skin this particular cat, but hoping you can help point me generally down the right path.
Thanks a ton and please let me know if you want me to provide anymore detail!
Thanks,
Alex
________________________________________________________
The information contained in this e-mail is confidential and/or proprietary to Capital One and/or its affiliates and may only be used solely in performance of work or services for Capital One. The information transmitted herewith is intended only for use by the individual or entity to which it is addressed. If the reader of this message is not the intended recipient, you are hereby notified that any review, retransmission, dissemination, distribution, copying or other use of, or taking of any action in reliance upon this information is strictly prohibited. If you have received this communication in error, please contact the sender and delete the material from your computer.
Re: Asynchronous JMS Consumer for IBM MQ
Posted by Joe Witt <jo...@gmail.com>.
Ian,
With run duration the idea is that the processor will be allowed to
keep executing for that period of time and the framework will keep
giving it the same process session. For the developer this means they
get to keep their logic very simple and discrete to a single operation
but that the framework will take care of batching those operations
together as one for up to 'X secs' of run duration.
For event driven the idea is that rather than telling the framework
you want the processor to run every X units of time as is the case in
timer driven with event driven the framework will execute the
processor (give it thread/call ontrigger) whenever there is data being
placed into its queue. It can be more efficient in some cases.
Thanks
Joe
On Thu, Dec 10, 2015 at 8:59 PM, ianwork <ia...@gmail.com> wrote:
> Bryan/Aldrin, Adding yielding into my processor prevent the number of tasks
> was rapidly increasing. Thanks!
>
> Aldrin, I would like to dig a little more into the details. My application
> is basically set do process logs like logstash. The application is reading
> and parsing a high volume of logs. The listener is based up listenSyslog.
> The listener forwards the logs to various processors which run regex's on
> batches of logs. Increasing the number of regex processors reduces the
> performance so i'd like to determine how I can configure the system
> resources.
>
> I'm still struggling with run duration. Does setting run duration mean that
> when a thread is allocated to the ontrigger method of that processor it will
> run for a maximum of run duration? What if the method executes faster than
> the run duration, will ontrigger be called again if there is work to be
> done?
>
>
> Is the event driven mode something to consider in my type of processor?
> What use cases was that designed to satisfy and is there any documentation
> on that?
>
>
>
>
> --
> View this message in context: http://apache-nifi-developer-list.39713.n7.nabble.com/Asynchronous-JMS-Consumer-for-IBM-MQ-tp3919p5715.html
> Sent from the Apache NiFi Developer List mailing list archive at Nabble.com.
Re: Asynchronous JMS Consumer for IBM MQ
Posted by ianwork <ia...@gmail.com>.
Bryan/Aldrin, Adding yielding into my processor prevent the number of tasks
was rapidly increasing. Thanks!
Aldrin, I would like to dig a little more into the details. My application
is basically set do process logs like logstash. The application is reading
and parsing a high volume of logs. The listener is based up listenSyslog.
The listener forwards the logs to various processors which run regex's on
batches of logs. Increasing the number of regex processors reduces the
performance so i'd like to determine how I can configure the system
resources.
I'm still struggling with run duration. Does setting run duration mean that
when a thread is allocated to the ontrigger method of that processor it will
run for a maximum of run duration? What if the method executes faster than
the run duration, will ontrigger be called again if there is work to be
done?
Is the event driven mode something to consider in my type of processor?
What use cases was that designed to satisfy and is there any documentation
on that?
--
View this message in context: http://apache-nifi-developer-list.39713.n7.nabble.com/Asynchronous-JMS-Consumer-for-IBM-MQ-tp3919p5715.html
Sent from the Apache NiFi Developer List mailing list archive at Nabble.com.
Re: Asynchronous JMS Consumer for IBM MQ
Posted by Bryan Bende <bb...@gmail.com>.
Adding to what Aldrin already said...
If you are using the pattern where a background thread puts data on a
queue, and onTrigger polls the queue, we found that polling with a small
wait time and then yielding when no data is available works well.
For example:
MyObject obj = queue.poll(100, TimeUnit.MILLISECONDS);
if (obj == null) {
context.yield();
return;
}
The yield duration is configurable in the UI so this lets the user decide
how long to yield for. Also, the wait time on the poll is important because
without that it can sometimes unnecessarily yield... if you had several
concurrent tasks calling onTrigger and one of them happens to call it when
the queue was empty but 1ms later data is placed in the queue then it would
have yielded for no reason. Hope I'm not confusing things more, just
sharing my learning experience from working on ListenSyslog.
On Thu, Dec 3, 2015 at 12:15 AM, Aldrin Piri <al...@gmail.com> wrote:
> Ian,
>
> The onTrigger method is called by the processor when the Flow Controller
> allocates a thread to this processor to execute. This share of resources
> is driven by the number of concurrent tasks and overall processing of the
> system. In times of light load, perhaps polling for data, and not doing
> much else in the flow, the FlowController will continuously allocate an
> execution time to that processor instance, invoking its onTrigger. The way
> to reason about this, is that without yielding in the processor, it is
> continuously looking for actionable events from whatever system/source it
> is consuming.
>
> In terms of considerations and concerns, the key item is that when you
> spawn these items off, they are out of the purview of the framework itself
> and an appropriate amount of diligence is needed in maintaining these and
> handling them in conjunction with the lifecycle of the processor itself;
> the controller is not managing their execution.
>
> In terms of what affects run duration, this is a net view of the number of
> available timer threads in conjunction with the number of extensions and
> their relative number of concurrent tasks. The FlowController will
> effectively dole out these specified threads to processors and extensions
> and let them run with it until they have completed their trigger method.
> The effect that occurs depends on what your mechanism for the background
> threads and what they are doing. There are a few different approaches to
> how this is handled depending on specific application and context, but you
> can see various approaches to how the "source" processors are
> orchestrated. As mentioned previously by Mark, the GetTwitter processor
> and the assorted Listen processors handle this kind of background,
> continuously executing task in different ways. The key item is that what
> "consumes" threads by the FlowController are instances of processors and
> their onTrigger method.
>
> If you would like some additional guidance or insights any details or
> specifics on what you are trying to tackle could certainly aid us in
> getting into the nitty gritty a bit more.
>
> --aldrin
>
>
> On Wed, Dec 2, 2015 at 10:00 PM, ianwork <ia...@gmail.com> wrote:
>
> > I have some questions related to this type of processor...When is the
> > onTrigger method called? I'm noticing a high and constantly increasing
> > number of Tasks in tasks/time statistics for my processor.
> >
> > "For some source/sink type processors, which sounds like your processor,
> it
> > can be acceptable to create a controlled number of threads. " What are
> the
> > considerations/concerns with having multiple threads in this situation?
> > What affect would run duration have on reading incoming messages into
> queue
> > and in ontrigger?
> >
> >
> >
> > --
> > View this message in context:
> >
> http://apache-nifi-developer-list.39713.n7.nabble.com/Asynchronous-JMS-Consumer-for-IBM-MQ-tp3919p5574.html
> > Sent from the Apache NiFi Developer List mailing list archive at
> > Nabble.com.
> >
>
Re: Asynchronous JMS Consumer for IBM MQ
Posted by Aldrin Piri <al...@gmail.com>.
Ian,
The onTrigger method is called by the processor when the Flow Controller
allocates a thread to this processor to execute. This share of resources
is driven by the number of concurrent tasks and overall processing of the
system. In times of light load, perhaps polling for data, and not doing
much else in the flow, the FlowController will continuously allocate an
execution time to that processor instance, invoking its onTrigger. The way
to reason about this, is that without yielding in the processor, it is
continuously looking for actionable events from whatever system/source it
is consuming.
In terms of considerations and concerns, the key item is that when you
spawn these items off, they are out of the purview of the framework itself
and an appropriate amount of diligence is needed in maintaining these and
handling them in conjunction with the lifecycle of the processor itself;
the controller is not managing their execution.
In terms of what affects run duration, this is a net view of the number of
available timer threads in conjunction with the number of extensions and
their relative number of concurrent tasks. The FlowController will
effectively dole out these specified threads to processors and extensions
and let them run with it until they have completed their trigger method.
The effect that occurs depends on what your mechanism for the background
threads and what they are doing. There are a few different approaches to
how this is handled depending on specific application and context, but you
can see various approaches to how the "source" processors are
orchestrated. As mentioned previously by Mark, the GetTwitter processor
and the assorted Listen processors handle this kind of background,
continuously executing task in different ways. The key item is that what
"consumes" threads by the FlowController are instances of processors and
their onTrigger method.
If you would like some additional guidance or insights any details or
specifics on what you are trying to tackle could certainly aid us in
getting into the nitty gritty a bit more.
--aldrin
On Wed, Dec 2, 2015 at 10:00 PM, ianwork <ia...@gmail.com> wrote:
> I have some questions related to this type of processor...When is the
> onTrigger method called? I'm noticing a high and constantly increasing
> number of Tasks in tasks/time statistics for my processor.
>
> "For some source/sink type processors, which sounds like your processor, it
> can be acceptable to create a controlled number of threads. " What are the
> considerations/concerns with having multiple threads in this situation?
> What affect would run duration have on reading incoming messages into queue
> and in ontrigger?
>
>
>
> --
> View this message in context:
> http://apache-nifi-developer-list.39713.n7.nabble.com/Asynchronous-JMS-Consumer-for-IBM-MQ-tp3919p5574.html
> Sent from the Apache NiFi Developer List mailing list archive at
> Nabble.com.
>
Re: Asynchronous JMS Consumer for IBM MQ
Posted by ianwork <ia...@gmail.com>.
I have some questions related to this type of processor...When is the
onTrigger method called? I'm noticing a high and constantly increasing
number of Tasks in tasks/time statistics for my processor.
"For some source/sink type processors, which sounds like your processor, it
can be acceptable to create a controlled number of threads. " What are the
considerations/concerns with having multiple threads in this situation?
What affect would run duration have on reading incoming messages into queue
and in ontrigger?
--
View this message in context: http://apache-nifi-developer-list.39713.n7.nabble.com/Asynchronous-JMS-Consumer-for-IBM-MQ-tp3919p5574.html
Sent from the Apache NiFi Developer List mailing list archive at Nabble.com.
Re: Asynchronous JMS Consumer for IBM MQ
Posted by Mark Payne <ma...@hotmail.com>.
Alex,
You can certainly use a queue and queue up the messages in a callback. Typically, the way this works is:
You create a method with an @OnScheduled annotation. In this method, you create a client object and start
the client. Here you can provide a callback to the client, and have that callback add the incoming message
to the queue.
In the onTrigger method of your processor, you can call queue.poll(). If this returns null then you'd called context.yield()
and return. This allows you to avoid using a lot of CPU resources when there's nothing to do. If a message is returned,
then you would create a new FlowFile for it, write the contents of the message to the FlowFile, and then call session.commit().
After session.commit() has completed, you are now free to acknowledge the message (assuming that you are using Client Acknowledgment).
The GetTwitter processor actually behaves fairly similarly to this.
Does this help?
Thanks
-Mark
> On Nov 5, 2015, at 9:49 AM, McCullough, Alex <Al...@capitalone.com> wrote:
>
> Hey There,
>
> This is a bit of a generic question, and I am not sure if this is the right place to start but I am new to developing custom processors in Nifi but have been trying to write an asynchronous message consumer to consume from IBM MQ.
>
> I have read all of the developer documentation and have been testing a processor I built, but I think where I am struggling is conceptualizing exactly how I should set this up within the framework and how the threading within Nifi works exactly.
>
> Typically I am used to registering a listener and using the onMessage method to add messages that are received to local queue to then be processed. What would this look like within the Nifi framework? Are there any guiding principals you can point me towards or any source code that may be doing something somewhat similar?
>
> Sorry this is a pretty vague question and I’m sure there are many ways to skin this particular cat, but hoping you can help point me generally down the right path.
>
> Thanks a ton and please let me know if you want me to provide anymore detail!
>
> Thanks,
> Alex
>
> ________________________________________________________
>
> The information contained in this e-mail is confidential and/or proprietary to Capital One and/or its affiliates and may only be used solely in performance of work or services for Capital One. The information transmitted herewith is intended only for use by the individual or entity to which it is addressed. If the reader of this message is not the intended recipient, you are hereby notified that any review, retransmission, dissemination, distribution, copying or other use of, or taking of any action in reliance upon this information is strictly prohibited. If you have received this communication in error, please contact the sender and delete the material from your computer.
Re: Asynchronous JMS Consumer for IBM MQ
Posted by Bryan Bende <bb...@gmail.com>.
Alex,
Generally the threading in NiFi is handled by the concurrent tasks
configured for the processor, meaning more threads able to call the
onTrigger method. For some source/sink type processors, which sounds like
your processor, it can be acceptable to create a controlled number of
threads. There are a couple of processors that take the approach of
starting a background thread in an @OnScheduled method to put messages onto
a queue that are then pulled off for processing in the onTrigger method. A
good example to look at might be the GetTwitter processor which has a
BlockingQueue of messages. You will definitely want to set a capacity on
the queue to prevent out-of-memory scenarios if a high volume of messages
are coming through.
Let us know if this helps, or if there is any more information we can
provide.
-Bryan
On Thu, Nov 5, 2015 at 9:49 AM, McCullough, Alex <
Alex.McCullough@capitalone.com> wrote:
> Hey There,
>
> This is a bit of a generic question, and I am not sure if this is the
> right place to start but I am new to developing custom processors in Nifi
> but have been trying to write an asynchronous message consumer to consume
> from IBM MQ.
>
> I have read all of the developer documentation and have been testing a
> processor I built, but I think where I am struggling is conceptualizing
> exactly how I should set this up within the framework and how the threading
> within Nifi works exactly.
>
> Typically I am used to registering a listener and using the onMessage
> method to add messages that are received to local queue to then be
> processed. What would this look like within the Nifi framework? Are there
> any guiding principals you can point me towards or any source code that may
> be doing something somewhat similar?
>
> Sorry this is a pretty vague question and I’m sure there are many ways to
> skin this particular cat, but hoping you can help point me generally down
> the right path.
>
> Thanks a ton and please let me know if you want me to provide anymore
> detail!
>
> Thanks,
> Alex
>
> ________________________________________________________
>
> The information contained in this e-mail is confidential and/or
> proprietary to Capital One and/or its affiliates and may only be used
> solely in performance of work or services for Capital One. The information
> transmitted herewith is intended only for use by the individual or entity
> to which it is addressed. If the reader of this message is not the intended
> recipient, you are hereby notified that any review, retransmission,
> dissemination, distribution, copying or other use of, or taking of any
> action in reliance upon this information is strictly prohibited. If you
> have received this communication in error, please contact the sender and
> delete the material from your computer.
>