Re: Using multithreaded library within ProcessFunction with callbacks relying on the out parameter

[Salva Alcántara <sa...@gmail.com>]

Hi again Piotr,

I have further considered the mailbox executor approach and I think it will
not be enough for my purposes. Here is why:

- My state consists of models created with a third party library
- These models have their own state, which means that when I forward events
in `ProcessElement1` to these models, the model's state will be updated
accordingly.

So, what would happen if:

- A new element E is processed in `ProcessElement1` and sent to the third
party library model
- A checkpoint is taken, in particular snapshotting all the library models
in use
- The element E that was sent to the library is expected to generate an
output O result when the callback is called, but a failure happens before
that
- Application recovers from the snapshot and continue processing elements,
but the callback generating the expected output O has been lost by now, so
that output will be lost

By considering the above case, I realize that the only option for me might
be to rely on AsyncIO. However, this is far from ideal because I am not
expecting an output result for each element I send to my models. I could use
a timeout but that may slow down processing as asyncIO has a limited queue
of "active" elements. Also, most of the times, I am not expecting a result
back at all from my models (callbacks will be invoked only a few times since
my modes are detecting anomalies).

In your opinion, what would be the best approach for handling this use case?



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Using multithreaded library within ProcessFunction with callbacks relying on the out parameter

[Salva Alcántara <sa...@gmail.com>]

Sounds like a plan Arvid! Taking note of it, this is gold!



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Using multithreaded library within ProcessFunction with callbacks relying on the out parameter

[Arvid Heise <ar...@ververica.com>]

I was wondering if you could actually really use AsyncWaitOperator in the
following way.

- Use a rather big timeout (so if callbacks usually take 1s, use 10).
- Use UNORDERED mode.
- Use a rather big queue size that would not cause any backpressure (you
could just experiment with different settings).

Then, you'd probably get to the operator that you would need to implement
manually anyways.
- Requests come in a specific order, that order is retained when calling
the external library.
- Results are immediately returned (depending on your watermark settings)
resulting in no additional latency (because of UNORDERED).
- The big timeouts guarantee that you will not dismiss a certain input too
quickly, if the callback takes longer than usual. It will clean up all
elements from state that have no callbacks after the given time though.
- The big queue size will avoid backpressure resulting from many pending
requests without response. Let's say you have 100 requests per second and a
timeout of 10s, that means a queue size of 1000 would allow all incoming
requests to be processed almost instantly (ignored the actual callbacks
that decrease the needed queue size as you said it to be a rather rare
event)

On Thu, Apr 9, 2020 at 11:09 AM Salva Alcántara <sa...@gmail.com>
wrote:

> Perfectly understood, thanks a lot for your reply/patience . I will take a
> look at AsyncWaitOperator and adapt from there if I really need that.
>
>
>
> --
> Sent from:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
>


-- 

Arvid Heise | Senior Java Developer

<https://www.ververica.com/>

Follow us @VervericaData

--

Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference

Stream Processing | Event Driven | Real Time

--

Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany

--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji
(Toni) Cheng

Re: Using multithreaded library within ProcessFunction with callbacks relying on the out parameter

[Salva Alcántara <sa...@gmail.com>]

Perfectly understood, thanks a lot for your reply/patience . I will take a
look at AsyncWaitOperator and adapt from there if I really need that.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Using multithreaded library within ProcessFunction with callbacks relying on the out parameter

[Piotr Nowojski <pi...@ververica.com>]

Hi,

With:

> Can not you take into account the pending element that’s stuck somewhere in the transit? Snapshot it as well and during recovery reprocess it? This is exactly that’s AsyncWaitOperator is doing.

I didn’t mean for you to use AsynWaitOperator, but what both me and Arvid suggested you previously:

> Also as Kostas pointed out, the easiest way would be to try use AsyncWaitOperator. If that’s not possible, you can implement your custom logic based on its code.

You can copy/duplicate & modify/adjust the AsyncWaitOperator logic inside your custom operator. You don’t have to use it if you have some special requirements, you can implement your own custom logic. Specifically I meant to mimic 

org.apache.flink.streaming.api.operators.async.AsyncWaitOperator#queue

Field and how is it being used during snapshotting state & recovery.

Piotrek

> On 9 Apr 2020, at 06:10, Salva Alcántara <sa...@gmail.com> wrote:
> 
> I agree with your point Piotrek, AsyncIO would handle all the pending data
> for me. However, the reason why I did not want to use it is because in my
> case, the callbacks are not always called in response of new data being sent
> to the third party lib. Indeed, the callback will be called rather
> uncommonly (since in my case it will mean that an anomaly has been
> detected). This means that If I go with AsyncIO I will need to setup a max
> timeout for every element, when only a few of them will actuallyinvoke the
> callback (i.e., produce any data in response). This seems rather drastic
> because it will probably add too much latency unnecessarily, but I agree on
> that maybe there is no other way if I need exactly once guarantees.
> 
> 
> 
> --
> Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Using multithreaded library within ProcessFunction with callbacks relying on the out parameter

[Salva Alcántara <sa...@gmail.com>]

I agree with your point Piotrek, AsyncIO would handle all the pending data
for me. However, the reason why I did not want to use it is because in my
case, the callbacks are not always called in response of new data being sent
to the third party lib. Indeed, the callback will be called rather
uncommonly (since in my case it will mean that an anomaly has been
detected). This means that If I go with AsyncIO I will need to setup a max
timeout for every element, when only a few of them will actuallyinvoke the
callback (i.e., produce any data in response). This seems rather drastic
because it will probably add too much latency unnecessarily, but I agree on
that maybe there is no other way if I need exactly once guarantees.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Using multithreaded library within ProcessFunction with callbacks relying on the out parameter

[Piotr Nowojski <pi...@ververica.com>]

Hi Salva,

Can not you take into account the pending element that’s stuck somewhere in the transit? Snapshot it as well and during recovery reprocess it? This is exactly that’s AsyncWaitOperator is doing.

Piotrek

> On 5 Apr 2020, at 15:00, Salva Alcántara <sa...@gmail.com> wrote:
> 
> Hi again Piotr,
> 
> I have further considered the mailbox executor approach and I think it will
> not be enough for my purposes. Here is why:
> 
> - My state consists of models created with a third party library
> - These models have their own state, which means that when I forward events
> in `ProcessElement1` to these models, the model's state will be updated
> accordingly.
> 
> So, what would happen if:
> 
> - A new element E is processed in `ProcessElement1` and sent to the third
> party library model
> - A checkpoint is taken, in particular snapshotting all the library models
> in use
> - The element E that was sent to the library is expected to generate an
> output O result when the callback is called, but a failure happens before
> that
> - Application recovers from the snapshot and continue processing elements,
> but the callback generating the expected output O has been lost by now, so
> that output will be lost
> 
> By considering the above case, I realize that the only option for me might
> be to rely on AsyncIO. However, this is far from ideal because I am not
> expecting an output result for each element I send to my models. I could use
> a timeout but that may slow down processing as asyncIO has a limited queue
> of "active" elements. Also, most of the times, I am not expecting a result
> back at all from my models (callbacks will be invoked only a few times since
> my modes are detecting anomalies).
> 
> In your opinion, what would be the best approach for handling this use case?
> 
> 
> 
> --
> Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/