respect "soft_fail" argument when running BatchSensor in deferrable mode

[Lee-W]

This is the following PR to #33196 and intends to solve the issue that the soft_fail argument might not work on all sensors.

This pull request addresses the issue of running BatchSensor in deferrable mode with soft_fail=True still raises an AirflowException, which should be AirflowSkipException. As raise_failed_or_skiping_exception won't exist in BaseSensorOperator before airflow (probably 2.7.1?), I add it to airflow.providers.amazon.aws.sensors. Not sure whether it's the correct way to do so. If so, I'll continue work on fixing other providers' sensors as well

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[eladkal]

so basically what you are after is to backport a fix we have in 2.7.1
The proper way to do it is:

1. Try to import the function from base_sensor if failed on import error then use the local version.
2. The local version should be private _raise_failed_or_skiping_exception so we can remove it without deprecation cycle.
3. There should be a note on the function to remove it when min airflow version for provider is >=2.7.1

[hussein-awala]

Try to import the function from base_sensor if failed on import error then use the local version.

This method was removed from base sensor after merging #33424.

The local version should be private _raise_failed_or_skiping_exception so we can remove it without deprecation cycle.
There should be a note on the function to remove it when min airflow version for provider is >=2.7.

a better solution could be removing this method and duplicating this small block in all the operators, it's the most safe solution to avoid breaking changes. WDYT?

[eladkal]

I didn't track the flow of PR around the soft_fail but that is exactly what troubles me.
I can be the one that pick up the todo in future releases when we want to do the cleanup. I do not understand what happens here. What is the old and what is the new.

I trust you guys have clarity on this thus i'm removing my review and you can proceed as you see fit (as I'm not in the details)

[hussein-awala]

Basically we fixed the issue in a more proper way in base sensor, and all the providers will work fine with Airflow +2.7.1, but to fix this issue with older versions of Airflow, we need to implement a temporary solution in each one until bumping the min Airflow version for these providers to 2.7.1

[eladkal]

what if-else block are you referring to?

Lets others decide when it's ready

[hussein-awala]

This solution will work if there is no errors in the trigger (and it is not canceled after a timeout).
Let me check if we can implement something which always work or we should accept a partial solution as it's temporary.

[Lee-W]

This solution will work if there is no errors in the trigger (and it is not canceled after a timeout). Let me check if we can implement something which always work or we should accept a partial solution as it's temporary.

or should we always send a failure event to execute_complete?

[hussein-awala]

or should we always send a failure event to execute_complete?

The problem is that in the older versions of Airflow, the failure is handled without executing the task methods.

I checked and I didn't find a simple solution. W can accept a partial fix, and if the user want a full fix, he can upgrade to 2.7.1

[hussein-awala]

Could we remove this method and add this to each operator:

    if self.soft_fail:
        raise AirflowSkipException(...)
    raise AirflowException(...)

In this way we will be able to remove it later without worrying about breaking changes

[Lee-W]

Sure! I just updated it as suggested