You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@storm.apache.org by Srividhya Shanmugam <Sr...@fico.com> on 2015/05/21 05:40:52 UTC
RotatingMap and Spout
Hi,
I have been reading through code and online to find how and where exactly a tuple gets timed out and therefore the spout's fail method is invoked. As I look through the executor.clj class, a RotatingMap is always associated with a Spout. This is where the emitted tuples pending ack are stored.
Every time the RotatingMap's rotate() method is called, it invokes the passed in ExpiryCallback implementation to notify the expired tuple.
What I am trying to understand is who calls the rotate() method?
I also see a put method in the RotatingMap - I am guessing this is invoked when the Spout emits a new tuple.
Is my understanding correct?
Thank you very much,
Srividhya
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
RE: RotatingMap and Spout
Posted by Srividhya Shanmugam <Sr...@fico.com>.
Thanks for all the help explaining how message time out works.
From: 임정택 [mailto:kabhwan@gmail.com]
Sent: Monday, May 25, 2015 10:53 PM
To: user@storm.apache.org
Subject: Re: RotatingMap and Spout
Yes, Acker just evaluates that tuple is completed or failed when other bolts notifies ack / fail about tuple and notify to Spout. Spout handles all remain works.
In order to clean invalid tuples (timed-out), Acker uses rotating map to discard tuples which don't respond (ack / fail) to Acker in time.
2015-05-26 11:36 GMT+09:00 Srividhya Shanmugam <Sr...@fico.com>>:
Thanks again for the explanation. I spent some more time looking at the source code for executor and acker. The rotating Map instance in acker just holds the tuples received by the acker. This instance does not have an ExpiredCallBack implementation. So the message time out delta is not evaluated when rotate method is invoked here. All those logic actually happens in the executor’s mk-thread method for spout.
Let me know if my understanding is different.
From: 임정택 [mailto:kabhwan@gmail.com<ma...@gmail.com>]
Sent: Thursday, May 21, 2015 6:40 PM
To: user@storm.apache.org<ma...@storm.apache.org>
Subject: Re: RotatingMap and Spout
2015년 5월 22일 금요일, Srividhya Shanmugam<Sr...@fico.com>>님이 작성한 메시지:
Thank you. This was helpful. Just to confirm, the RotatingMap is associated with the Executor thread of the Spout task. The messages are verified for their expiration when rotate() is invoked on the RotatingMap.
And this gets invoked when for every tick tuple received by the Spout. Is this correct?
Yes, right.
I also see the acker holding on to a rotating Map…so every tuple emitted by the spout will be in the rotating Map of both Executor thread for spout task and the executor thread of the acker bolt. And in both cases, only for tick tuples the rotate method is invoked. Why there are rotating map instances in two different threads?
I don't know about design concept about these.
But logically two things are different executors so it can be run from different workers, and though it runs with same worker, synchronization for the map could introduce worse performance.
Thanks again,
Srividhya
Thanks!
Jungtaek Lim (HeartSaVioR)
From: 임정택 [mailto:kabhwan@gmail.com]
Sent: Thursday, May 21, 2015 12:36 AM
To: user@storm.apache.org<ma...@storm.apache.org>
Subject: Re: RotatingMap and Spout
Hi.
AFAIK, Spout sets up timer to emit tick tuples each TOPOLOGY-TICK-TUPLE-FREQ-SECS.
(Please refer setup-ticks!)
Spout changes TOPOLOGY-TICK-TUPLE-FREQ-SECS to TOPOLOGY-MESSAGEE-TIMEOUT-SECS for the first time, so timer is scheduled to each TOPOLOGY-MESSAGE-TIMEOUT-SECS.
(Please refer add-acker!)
When Spout receives tick tuple, it calls pending.rotate(). Its size is 2, so actual tuple expiration time is between TOPOLOGY-MESSAGE-TIMEOUT-SECS and (2 * TOPOLOGY-MESSAGE-TIMEOUT-SECS).
(Please refer mk-threads :spout)
Hope this helps.
Thanks!
Jungtaek Lim (HeartSaVioR)
2015-05-21 12:40 GMT+09:00 Srividhya Shanmugam <Sr...@fico.com>>:
Hi,
I have been reading through code and online to find how and where exactly a tuple gets timed out and therefore the spout’s fail method is invoked. As I look through the executor.clj class, a RotatingMap is always associated with a Spout. This is where the emitted tuples pending ack are stored.
Every time the RotatingMap’s rotate() method is called, it invokes the passed in ExpiryCallback implementation to notify the expired tuple.
What I am trying to understand is who calls the rotate() method?
I also see a put method in the RotatingMap – I am guessing this is invoked when the Spout emits a new tuple.
Is my understanding correct?
Thank you very much,
Srividhya
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
Re: RotatingMap and Spout
Posted by 임정택 <ka...@gmail.com>.
Yes, Acker just evaluates that tuple is completed or failed when other
bolts notifies ack / fail about tuple and notify to Spout. Spout handles
all remain works.
In order to clean invalid tuples (timed-out), Acker uses rotating map to
discard tuples which don't respond (ack / fail) to Acker in time.
2015-05-26 11:36 GMT+09:00 Srividhya Shanmugam <Sr...@fico.com>
:
> Thanks again for the explanation. I spent some more time looking at the
> source code for executor and acker. The rotating Map instance in acker just
> holds the tuples received by the acker. This instance does not have an
> ExpiredCallBack implementation. So the message time out delta is not
> evaluated when rotate method is invoked here. All those logic actually
> happens in the executor’s mk-thread method for spout.
>
> Let me know if my understanding is different.
>
>
>
> *From:* 임정택 [mailto:kabhwan@gmail.com]
> *Sent:* Thursday, May 21, 2015 6:40 PM
>
> *To:* user@storm.apache.org
> *Subject:* Re: RotatingMap and Spout
>
>
>
>
>
> 2015년 5월 22일 금요일, Srividhya Shanmugam<Sr...@fico.com>님이 작성한
> 메시지:
>
> Thank you. This was helpful. Just to confirm, the RotatingMap is
> associated with the Executor thread of the Spout task. The messages are
> verified for their expiration when rotate() is invoked on the RotatingMap.
>
>
>
> And this gets invoked when for every tick tuple received by the Spout. Is
> this correct?
>
>
>
> Yes, right.
>
>
>
> I also see the acker holding on to a rotating Map…so every tuple emitted
> by the spout will be in the rotating Map of both Executor thread for spout
> task and the executor thread of the acker bolt. And in both cases, only for
> tick tuples the rotate method is invoked. Why there are rotating map
> instances in two different threads?
>
> I don't know about design concept about these.
>
> But logically two things are different executors so it can be run
> from different workers, and though it runs with same
> worker, synchronization for the map could introduce worse performance.
>
>
>
> Thanks again,
>
> Srividhya
>
>
>
> Thanks!
>
> Jungtaek Lim (HeartSaVioR)
>
>
>
>
>
> *From:* 임정택 [mailto:kabhwan@gmail.com]
> *Sent:* Thursday, May 21, 2015 12:36 AM
> *To:* user@storm.apache.org
> *Subject:* Re: RotatingMap and Spout
>
>
>
> Hi.
>
>
>
> AFAIK, Spout sets up timer to emit tick tuples each
> TOPOLOGY-TICK-TUPLE-FREQ-SECS.
>
> (Please refer setup-ticks!)
>
>
>
> Spout changes TOPOLOGY-TICK-TUPLE-FREQ-SECS to
> TOPOLOGY-MESSAGEE-TIMEOUT-SECS for the first time, so timer is scheduled to
> each TOPOLOGY-MESSAGE-TIMEOUT-SECS.
>
> (Please refer add-acker!)
>
>
>
> When Spout receives tick tuple, it calls pending.rotate(). Its size is 2,
> so actual tuple expiration time is between TOPOLOGY-MESSAGE-TIMEOUT-SECS
> and (2 * TOPOLOGY-MESSAGE-TIMEOUT-SECS).
>
> (Please refer mk-threads :spout)
>
>
>
> Hope this helps.
>
>
>
> Thanks!
>
> Jungtaek Lim (HeartSaVioR)
>
>
>
>
>
> 2015-05-21 12:40 GMT+09:00 Srividhya Shanmugam <
> SrividhyaShanmugam@fico.com>:
>
> Hi,
>
>
>
> I have been reading through code and online to find how and where exactly
> a tuple gets timed out and therefore the spout’s fail method is invoked. As
> I look through the executor.clj class, a RotatingMap is always associated
> with a Spout. This is where the emitted tuples pending ack are stored.
>
>
>
> Every time the RotatingMap’s rotate() method is called, it invokes the
> passed in ExpiryCallback implementation to notify the expired tuple.
>
>
>
> What I am trying to understand is who calls the rotate() method?
>
> I also see a put method in the RotatingMap – I am guessing this is invoked
> when the Spout emits a new tuple.
>
>
>
> Is my understanding correct?
>
>
>
> Thank you very much,
>
> Srividhya
>
>
> This email and any files transmitted with it are confidential, proprietary
> and intended solely for the individual or entity to whom they are
> addressed. If you have received this email in error please delete it
> immediately.
>
>
>
>
>
> --
>
> Name : 임 정택
> Blog : http://www.heartsavior.net / http://dev.heartsavior.net
> Twitter : http://twitter.com/heartsavior
>
> LinkedIn : http://www.linkedin.com/in/heartsavior
>
>
> This email and any files transmitted with it are confidential, proprietary
> and intended solely for the individual or entity to whom they are
> addressed. If you have received this email in error please delete it
> immediately.
>
>
>
> --
>
> Name : 임 정택
> Blog : http://www.heartsavior.net / http://dev.heartsavior.net
> Twitter : http://twitter.com/heartsavior
>
> LinkedIn : http://www.linkedin.com/in/heartsavior
>
>
>
> This email and any files transmitted with it are confidential, proprietary
> and intended solely for the individual or entity to whom they are
> addressed. If you have received this email in error please delete it
> immediately.
>
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
RE: RotatingMap and Spout
Posted by Srividhya Shanmugam <Sr...@fico.com>.
Thanks again for the explanation. I spent some more time looking at the source code for executor and acker. The rotating Map instance in acker just holds the tuples received by the acker. This instance does not have an ExpiredCallBack implementation. So the message time out delta is not evaluated when rotate method is invoked here. All those logic actually happens in the executor’s mk-thread method for spout.
Let me know if my understanding is different.
From: 임정택 [mailto:kabhwan@gmail.com]
Sent: Thursday, May 21, 2015 6:40 PM
To: user@storm.apache.org
Subject: Re: RotatingMap and Spout
2015년 5월 22일 금요일, Srividhya Shanmugam<Sr...@fico.com>>님이 작성한 메시지:
Thank you. This was helpful. Just to confirm, the RotatingMap is associated with the Executor thread of the Spout task. The messages are verified for their expiration when rotate() is invoked on the RotatingMap.
And this gets invoked when for every tick tuple received by the Spout. Is this correct?
Yes, right.
I also see the acker holding on to a rotating Map…so every tuple emitted by the spout will be in the rotating Map of both Executor thread for spout task and the executor thread of the acker bolt. And in both cases, only for tick tuples the rotate method is invoked. Why there are rotating map instances in two different threads?
I don't know about design concept about these.
But logically two things are different executors so it can be run from different workers, and though it runs with same worker, synchronization for the map could introduce worse performance.
Thanks again,
Srividhya
Thanks!
Jungtaek Lim (HeartSaVioR)
From: 임정택 [mailto:kabhwan@gmail.com<javascript:_e(%7B%7D,'cvml','kabhwan@gmail.com');>]
Sent: Thursday, May 21, 2015 12:36 AM
To: user@storm.apache.org<javascript:_e(%7B%7D,'cvml','user@storm.apache.org');>
Subject: Re: RotatingMap and Spout
Hi.
AFAIK, Spout sets up timer to emit tick tuples each TOPOLOGY-TICK-TUPLE-FREQ-SECS.
(Please refer setup-ticks!)
Spout changes TOPOLOGY-TICK-TUPLE-FREQ-SECS to TOPOLOGY-MESSAGEE-TIMEOUT-SECS for the first time, so timer is scheduled to each TOPOLOGY-MESSAGE-TIMEOUT-SECS.
(Please refer add-acker!)
When Spout receives tick tuple, it calls pending.rotate(). Its size is 2, so actual tuple expiration time is between TOPOLOGY-MESSAGE-TIMEOUT-SECS and (2 * TOPOLOGY-MESSAGE-TIMEOUT-SECS).
(Please refer mk-threads :spout)
Hope this helps.
Thanks!
Jungtaek Lim (HeartSaVioR)
2015-05-21 12:40 GMT+09:00 Srividhya Shanmugam <SrividhyaShanmugam@fico.com<javascript:_e(%7B%7D,'cvml','SrividhyaShanmugam@fico.com');>>:
Hi,
I have been reading through code and online to find how and where exactly a tuple gets timed out and therefore the spout’s fail method is invoked. As I look through the executor.clj class, a RotatingMap is always associated with a Spout. This is where the emitted tuples pending ack are stored.
Every time the RotatingMap’s rotate() method is called, it invokes the passed in ExpiryCallback implementation to notify the expired tuple.
What I am trying to understand is who calls the rotate() method?
I also see a put method in the RotatingMap – I am guessing this is invoked when the Spout emits a new tuple.
Is my understanding correct?
Thank you very much,
Srividhya
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
Re: RotatingMap and Spout
Posted by 임정택 <ka...@gmail.com>.
2015년 5월 22일 금요일, Srividhya Shanmugam<Sr...@fico.com>님이 작성한
메시지:
> Thank you. This was helpful. Just to confirm, the RotatingMap is
> associated with the Executor thread of the Spout task. The messages are
> verified for their expiration when rotate() is invoked on the RotatingMap.
>
>
>
> And this gets invoked when for every tick tuple received by the Spout. Is
> this correct?
>
>
> Yes, right.
>
>
> I also see the acker holding on to a rotating Map…so every tuple emitted
> by the spout will be in the rotating Map of both Executor thread for spout
> task and the executor thread of the acker bolt. And in both cases, only for
> tick tuples the rotate method is invoked. Why there are rotating map
> instances in two different threads?
>
> I don't know about design concept about these.
But logically two things are different executors so it can be run
from different workers, and though it runs with same
worker, synchronization for the map could introduce worse performance.
>
>
Thanks again,
>
> Srividhya
>
>
>
> Thanks!
Jungtaek Lim (HeartSaVioR)
>
>
>
>
> *From:* 임정택 [mailto:kabhwan@gmail.com
> <javascript:_e(%7B%7D,'cvml','kabhwan@gmail.com');>]
> *Sent:* Thursday, May 21, 2015 12:36 AM
> *To:* user@storm.apache.org
> <javascript:_e(%7B%7D,'cvml','user@storm.apache.org');>
> *Subject:* Re: RotatingMap and Spout
>
>
>
> Hi.
>
>
>
> AFAIK, Spout sets up timer to emit tick tuples each
> TOPOLOGY-TICK-TUPLE-FREQ-SECS.
>
> (Please refer setup-ticks!)
>
>
>
> Spout changes TOPOLOGY-TICK-TUPLE-FREQ-SECS to
> TOPOLOGY-MESSAGEE-TIMEOUT-SECS for the first time, so timer is scheduled to
> each TOPOLOGY-MESSAGE-TIMEOUT-SECS.
>
> (Please refer add-acker!)
>
>
>
> When Spout receives tick tuple, it calls pending.rotate(). Its size is 2,
> so actual tuple expiration time is between TOPOLOGY-MESSAGE-TIMEOUT-SECS
> and (2 * TOPOLOGY-MESSAGE-TIMEOUT-SECS).
>
> (Please refer mk-threads :spout)
>
>
>
> Hope this helps.
>
>
>
> Thanks!
>
> Jungtaek Lim (HeartSaVioR)
>
>
>
>
>
> 2015-05-21 12:40 GMT+09:00 Srividhya Shanmugam <
> SrividhyaShanmugam@fico.com
> <javascript:_e(%7B%7D,'cvml','SrividhyaShanmugam@fico.com');>>:
>
> Hi,
>
>
>
> I have been reading through code and online to find how and where exactly
> a tuple gets timed out and therefore the spout’s fail method is invoked. As
> I look through the executor.clj class, a RotatingMap is always associated
> with a Spout. This is where the emitted tuples pending ack are stored.
>
>
>
> Every time the RotatingMap’s rotate() method is called, it invokes the
> passed in ExpiryCallback implementation to notify the expired tuple.
>
>
>
> What I am trying to understand is who calls the rotate() method?
>
> I also see a put method in the RotatingMap – I am guessing this is invoked
> when the Spout emits a new tuple.
>
>
>
> Is my understanding correct?
>
>
>
> Thank you very much,
>
> Srividhya
>
>
> This email and any files transmitted with it are confidential, proprietary
> and intended solely for the individual or entity to whom they are
> addressed. If you have received this email in error please delete it
> immediately.
>
>
>
>
>
> --
>
> Name : 임 정택
> Blog : http://www.heartsavior.net / http://dev.heartsavior.net
> Twitter : http://twitter.com/heartsavior
>
> LinkedIn : http://www.linkedin.com/in/heartsavior
>
> This email and any files transmitted with it are confidential, proprietary
> and intended solely for the individual or entity to whom they are
> addressed. If you have received this email in error please delete it
> immediately.
>
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
RE: RotatingMap and Spout
Posted by Srividhya Shanmugam <Sr...@fico.com>.
Thank you. This was helpful. Just to confirm, the RotatingMap is associated with the Executor thread of the Spout task. The messages are verified for their expiration when rotate() is invoked on the RotatingMap.
And this gets invoked when for every tick tuple received by the Spout. Is this correct?
I also see the acker holding on to a rotating Map…so every tuple emitted by the spout will be in the rotating Map of both Executor thread for spout task and the executor thread of the acker bolt. And in both cases, only for tick tuples the rotate method is invoked. Why there are rotating map instances in two different threads?
Thanks again,
Srividhya
From: 임정택 [mailto:kabhwan@gmail.com]
Sent: Thursday, May 21, 2015 12:36 AM
To: user@storm.apache.org
Subject: Re: RotatingMap and Spout
Hi.
AFAIK, Spout sets up timer to emit tick tuples each TOPOLOGY-TICK-TUPLE-FREQ-SECS.
(Please refer setup-ticks!)
Spout changes TOPOLOGY-TICK-TUPLE-FREQ-SECS to TOPOLOGY-MESSAGEE-TIMEOUT-SECS for the first time, so timer is scheduled to each TOPOLOGY-MESSAGE-TIMEOUT-SECS.
(Please refer add-acker!)
When Spout receives tick tuple, it calls pending.rotate(). Its size is 2, so actual tuple expiration time is between TOPOLOGY-MESSAGE-TIMEOUT-SECS and (2 * TOPOLOGY-MESSAGE-TIMEOUT-SECS).
(Please refer mk-threads :spout)
Hope this helps.
Thanks!
Jungtaek Lim (HeartSaVioR)
2015-05-21 12:40 GMT+09:00 Srividhya Shanmugam <Sr...@fico.com>>:
Hi,
I have been reading through code and online to find how and where exactly a tuple gets timed out and therefore the spout’s fail method is invoked. As I look through the executor.clj class, a RotatingMap is always associated with a Spout. This is where the emitted tuples pending ack are stored.
Every time the RotatingMap’s rotate() method is called, it invokes the passed in ExpiryCallback implementation to notify the expired tuple.
What I am trying to understand is who calls the rotate() method?
I also see a put method in the RotatingMap – I am guessing this is invoked when the Spout emits a new tuple.
Is my understanding correct?
Thank you very much,
Srividhya
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior
This email and any files transmitted with it are confidential, proprietary and intended solely for the individual or entity to whom they are addressed. If you have received this email in error please delete it immediately.
Re: RotatingMap and Spout
Posted by 임정택 <ka...@gmail.com>.
Hi.
AFAIK, Spout sets up timer to emit tick tuples each
TOPOLOGY-TICK-TUPLE-FREQ-SECS.
(Please refer setup-ticks!)
Spout changes TOPOLOGY-TICK-TUPLE-FREQ-SECS to
TOPOLOGY-MESSAGEE-TIMEOUT-SECS for the first time, so timer is scheduled to
each TOPOLOGY-MESSAGE-TIMEOUT-SECS.
(Please refer add-acker!)
When Spout receives tick tuple, it calls pending.rotate(). Its size is 2,
so actual tuple expiration time is between TOPOLOGY-MESSAGE-TIMEOUT-SECS
and (2 * TOPOLOGY-MESSAGE-TIMEOUT-SECS).
(Please refer mk-threads :spout)
Hope this helps.
Thanks!
Jungtaek Lim (HeartSaVioR)
2015-05-21 12:40 GMT+09:00 Srividhya Shanmugam <Sr...@fico.com>
:
> Hi,
>
>
>
> I have been reading through code and online to find how and where exactly
> a tuple gets timed out and therefore the spout’s fail method is invoked. As
> I look through the executor.clj class, a RotatingMap is always associated
> with a Spout. This is where the emitted tuples pending ack are stored.
>
>
>
> Every time the RotatingMap’s rotate() method is called, it invokes the
> passed in ExpiryCallback implementation to notify the expired tuple.
>
>
>
> What I am trying to understand is who calls the rotate() method?
>
> I also see a put method in the RotatingMap – I am guessing this is invoked
> when the Spout emits a new tuple.
>
>
>
> Is my understanding correct?
>
>
>
> Thank you very much,
>
> Srividhya
>
> This email and any files transmitted with it are confidential, proprietary
> and intended solely for the individual or entity to whom they are
> addressed. If you have received this email in error please delete it
> immediately.
>
--
Name : 임 정택
Blog : http://www.heartsavior.net / http://dev.heartsavior.net
Twitter : http://twitter.com/heartsavior
LinkedIn : http://www.linkedin.com/in/heartsavior