Re: Occasional Out of order tuples when emitting from a thread

[Ashwin Chandra Putta]

Sunil,

You can poll the queue in end window since process method in the input port
does not get called if there is no incoming tuple. However, end window is
called irrespective of there are incoming tuples or not.

Regards,
Ashwin.

On Tue, Feb 21, 2017 at 11:32 AM, Sunil Parmar <spar...@threatmetrix.com>
wrote:

> Ram,
> Thanks for the prompt response. If we use the approach you suggested we’re
> dependent on main thread’s process call I.e. Tuples in the thread safe
> queue gets only processed when main thread is processing incoming tuples.
> How can we explicitly call the process from polling of delay queue ?
>
> Just for reference here’s the sample code snippet for our operator.
>
> public class MyOperator extends BaseOperator implements
>
> Operator.ActivationListener {
> …..
>
> @InputPortFieldAnnotation
>
> public transient DefaultInputPort kafkaStreamInput =
>
> new DefaultInputPort() {
>
> List errors = new ArrayList();
>
> @Override
>
> public void process(String consumerRecord) {
>
> //Code for normal tuple process
>
> //Code to poll thread safe queue
>
> }
>
> ***—*
> *From: *Munagala Ramanath <r...@datatorrent.com>
> *To: *us...@apex.apache.org
> *CC: *"dev@apex.apache.org" <dev@apex.apache.org>, Allan De Leon <
> adel...@threatmetrix.com>, Tim Zhu <t...@threatmetrix.com>
> *Subject: *Re: Occasional Out of order tuples when emitting from a thread
> *Date: *2017-02-21 10:08 (-0800)
> *List: *us...@apex.apache.org
> <https://lists.apache.org/list.html?us...@apex.apache.org>
>
> Please note that tuples should not be emitted by any thread other than the
> main operator thread.
>
> A common pattern is to use a thread-safe queue and have worker threads
> enqueue
> tuples there; the main operator thread then pulls tuples from the queue and
> emits them.
>
> Ram
>
> ___
>
> Munagala V. Ramanath
>
> Software Engineer
>
> E: r...@datatorrent.com | M: (408) 331-5034 | Twitter: @UnknownRam
> www.datatorrent.com  |  apex.apache.org
>
>
> From: Sunil Parmar <spar...@threatmetrix.com>
> Date: Tuesday, February 21, 2017 at 10:05 AM
> To: "us...@apex.apache.org" <us...@apex.apache.org>, "dev@apex.apache.org"
> <dev@apex.apache.org>
> Cc: Allan De Leon <adel...@threatmetrix.com>, Tim Zhu <
> t...@threatmetrix.com>
> Subject: Occasional Out of order tuples when emitting from a thread
>
> Hi there,
> We have the following setup:
>
>- we have a generic operator that’s processing tuples in its input port
>- in the input port’s process method, we check for a condition, and:
>   - if the condition is met, the tuple is emitted to the next
>   operator right away (in the process method)
>   - Otherwise, if the condition is not met, we store the tuple  in
>   some cache and we use some threads that periodically check the 
> condition to
>   become true. Once the condition is true, the threads call the emit 
> method
>   on the stored tuples.
>
> With this setup, we occasionally encounter the following error:
> 2017-02-15 17:29:09,364 ERROR com.datatorrent.stram.engine.GenericNode:
> Catastrophic Error: Out of sequence BEGIN_WINDOW tuple 58a404613b7f on
> port transformedJSON while expecting 58a404613b7e
>
> Is there a way to make the above work correctly?
> If not, can you recommend a better way of doing this?
> How can we ensure window assignment is done synchronously before emitting
> tuples ?
>
> Thanks very much in advance…
> -allan
>



-- 

Regards,
Ashwin.

Re: Occasional Out of order tuples when emitting from a thread

[Sunil Parmar]

Ram,
Thanks for the prompt response. If we use the approach you suggested we're 
dependent on main thread's process call I.e. Tuples in the thread safe queue 
gets only processed when main thread is processing incoming tuples. How can we 
explicitly call the process from polling of delay queue ?

Just for reference here's the sample code snippet for our operator.


public class MyOperator extends BaseOperator implements

Operator.ActivationListener {

.


@InputPortFieldAnnotation

public transient DefaultInputPort kafkaStreamInput =

new DefaultInputPort() {

List errors = new ArrayList();

@Override

public void process(String consumerRecord) {

//Code for normal tuple process

//Code to poll thread safe queue

}

-
From: Munagala Ramanath <r...@datatorrent.com<mailto:r...@datatorrent.com>>
To: us...@apex.apache.org<mailto:us...@apex.apache.org>
CC: "dev@apex.apache.org<mailto:dev@apex.apache.org>" 
<dev@apex.apache.org<mailto:dev@apex.apache.org>>, Allan De Leon 
<adel...@threatmetrix.com<mailto:adel...@threatmetrix.com>>, Tim Zhu 
<t...@threatmetrix.com<mailto:t...@threatmetrix.com>>
Subject: Re: Occasional Out of order tuples when emitting from a thread
Date: 2017-02-21 10:08 (-0800)
List: 
us...@apex.apache.org<https://lists.apache.org/list.html?us...@apex.apache.org>

Please note that tuples should not be emitted by any thread other than the
main operator thread.

A common pattern is to use a thread-safe queue and have worker threads
enqueue
tuples there; the main operator thread then pulls tuples from the queue and
emits them.

Ram

___

Munagala V. Ramanath

Software Engineer

E: r...@datatorrent.com<mailto:r...@datatorrent.com> | M: (408) 331-5034 | 
Twitter: @UnknownRam

www.datatorrent.com  |  apex.apache.org

From: Sunil Parmar <spar...@threatmetrix.com<mailto:spar...@threatmetrix.com>>
Date: Tuesday, February 21, 2017 at 10:05 AM
To: "us...@apex.apache.org<mailto:us...@apex.apache.org>" 
<us...@apex.apache.org<mailto:us...@apex.apache.org>>, 
"dev@apex.apache.org<mailto:dev@apex.apache.org>" 
<dev@apex.apache.org<mailto:dev@apex.apache.org>>
Cc: Allan De Leon <adel...@threatmetrix.com<mailto:adel...@threatmetrix.com>>, 
Tim Zhu <t...@threatmetrix.com<mailto:t...@threatmetrix.com>>
Subject: Occasional Out of order tuples when emitting from a thread

Hi there,
We have the following setup:

  *   we have a generic operator that's processing tuples in its input port
  *   in the input port's process method, we check for a condition, and:
 *   if the condition is met, the tuple is emitted to the next operator 
right away (in the process method)
 *   Otherwise, if the condition is not met, we store the tuple  in some 
cache and we use some threads that periodically check the condition to become 
true. Once the condition is true, the threads call the emit method on the 
stored tuples.

With this setup, we occasionally encounter the following error:
2017-02-15 17:29:09,364 ERROR com.datatorrent.stram.engine.GenericNode: 
Catastrophic Error: Out of sequence BEGIN_WINDOW tuple 58a404613b7f on port 
transformedJSON while expecting 58a404613b7e

Is there a way to make the above work correctly?
If not, can you recommend a better way of doing this?
How can we ensure window assignment is done synchronously before emitting 
tuples ?

Thanks very much in advance...
-allan

Re: Occasional Out of order tuples when emitting from a thread

[Munagala Ramanath]

Please note that tuples should not be emitted by any thread other than the
main operator thread.

A common pattern is to use a thread-safe queue and have worker threads
enqueue
tuples there; the main operator thread then pulls tuples from the queue and
emits them.

Ram

On Tue, Feb 21, 2017 at 10:05 AM, Sunil Parmar 
wrote:

> Hi there,
> We have the following setup:
>
>- we have a generic operator that’s processing tuples in its input port
>- in the input port’s process method, we check for a condition, and:
>   - if the condition is met, the tuple is emitted to the next
>   operator right away (in the process method)
>   - Otherwise, if the condition is not met, we store the tuple  in
>   some cache and we use some threads that periodically check the 
> condition to
>   become true. Once the condition is true, the threads call the emit 
> method
>   on the stored tuples.
>
> With this setup, we occasionally encounter the following error:
> 2017-02-15 17:29:09,364 ERROR com.datatorrent.stram.engine.GenericNode:
> Catastrophic Error: Out of sequence BEGIN_WINDOW tuple 58a404613b7f on
> port transformedJSON while expecting 58a404613b7e
>
> Is there a way to make the above work correctly?
> If not, can you recommend a better way of doing this?
> How can we ensure window assignment is done synchronously before emitting
> tuples ?
>
> Thanks very much in advance…
> -allan
>



-- 

___

Munagala V. Ramanath

Software Engineer

E: r...@datatorrent.com | M: (408) 331-5034 | Twitter: @UnknownRam

www.datatorrent.com  |  apex.apache.org

Occasional Out of order tuples when emitting from a thread

[Sunil Parmar]

Hi there,
We have the following setup:

  *   we have a generic operator that's processing tuples in its input port
  *   in the input port's process method, we check for a condition, and:
 *   if the condition is met, the tuple is emitted to the next operator 
right away (in the process method)
 *   Otherwise, if the condition is not met, we store the tuple  in some 
cache and we use some threads that periodically check the condition to become 
true. Once the condition is true, the threads call the emit method on the 
stored tuples.

With this setup, we occasionally encounter the following error:
2017-02-15 17:29:09,364 ERROR com.datatorrent.stram.engine.GenericNode: 
Catastrophic Error: Out of sequence BEGIN_WINDOW tuple 58a404613b7f on port 
transformedJSON while expecting 58a404613b7e

Is there a way to make the above work correctly?
If not, can you recommend a better way of doing this?
How can we ensure window assignment is done synchronously before emitting 
tuples ?

Thanks very much in advance...
-allan