Output from RichAsyncFunction on failure

classic Classic list List threaded Threaded
3 messages Options
Reply | Threaded
Open this post in threaded view
|

Output from RichAsyncFunction on failure

Satish Saley
Hi, 
- I have a kafka consumer to read events.
- Then, I have RichAsyncFunction to call a remote service to get more information about that event.

If the remote call fails after X number of retries, I don't want flink to fail the job and start processing from the beginning. Instead I would like to push info about failed call to another Kafka topic. Is there a way to achieve this? 
Reply | Threaded
Open this post in threaded view
|

Re: Output from RichAsyncFunction on failure

Satish Saley
One way I thought to achieve this is - 
1. For failures, add a special record to collection in RichAsyncFunction
2. Filter out those special records from the DataStream and push to another Kafka
Let me know if it makes sense.


On Fri, Jun 11, 2021 at 10:40 AM Satish Saley <[hidden email]> wrote:
Hi, 
- I have a kafka consumer to read events.
- Then, I have RichAsyncFunction to call a remote service to get more information about that event.

If the remote call fails after X number of retries, I don't want flink to fail the job and start processing from the beginning. Instead I would like to push info about failed call to another Kafka topic. Is there a way to achieve this? 
Reply | Threaded
Open this post in threaded view
|

Re: Output from RichAsyncFunction on failure

Arvid Heise-4
Hi Satish,

usually you would side-outputs [1] for that but afaik asyncIO doesn't support that (yet).
So your option works well to use some union type. You can then chain a map function that uses side-outputs.


On Fri, Jun 11, 2021 at 7:49 PM Satish Saley <[hidden email]> wrote:
One way I thought to achieve this is - 
1. For failures, add a special record to collection in RichAsyncFunction
2. Filter out those special records from the DataStream and push to another Kafka
Let me know if it makes sense.


On Fri, Jun 11, 2021 at 10:40 AM Satish Saley <[hidden email]> wrote:
Hi, 
- I have a kafka consumer to read events.
- Then, I have RichAsyncFunction to call a remote service to get more information about that event.

If the remote call fails after X number of retries, I don't want flink to fail the job and start processing from the beginning. Instead I would like to push info about failed call to another Kafka topic. Is there a way to achieve this?