Enable conditional Retrying based on task response

[awolf22-bv]

Description

I wanted a way to only retry an AF task when certain errors are triggered when executing a task.

It seems like Retries are triggered anytime a task doesn't return a 0 success code (at least for BashOperator), and there is no way in AF to specify which exit code we should retry on, and which should move the task to failed.

I see there is a skip_exit_codes param for bash operator, but we don't want to mark the task as skipped and move next ones, because downstream tasks rely o successful completetion of this one.

Use case/motivation

We have an airtflow job which has multiple tasks for ingesting data from a 3rd party, and doing things with that data after in other tasks, which all run Django manage commands, so using the BashOperator for tasks.

The problem is the data sent by the 3rd party could be delayed, and there is no guaranteed time it will be there, so we wanted to retry the task every 30 minutes until the file was sent, and if it wasn't after 7 hours raise an error.

The reason we want conditional retrying, is so we only retry when the file not sent yet exception is raised in Python for the file not there yet, which we can raise a specific exit code related to that error.

It's important we don't retry on other errors, because we want to be notified of other failures immediately instead of retrying any exception/error X amount of times, as we do not expect any other exceptions to be raised.

Related issues

Unsure

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[Taragolis]

That's looks like you need a Sensor and BashSensor