Respect "soft_fail" for core async sensors

[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 the core async sensors potentially raising non-AirflowSkipException even when soft_fail is set to True. It achieves this by modifying the DateTimeTrigger to include the soft_fail argument.

---

Not related to this PR: I noticed that we are currently using separate sensors for synchronous and asynchronous operations in our core sensors. Do you think it would be beneficial to merge them into a single sensor, like what the providers do?

---

^ 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.

[hussein-awala]

Let's wait #33424 before deciding if we need to merge this one.

[Lee-W]

Let's wait #33424 before deciding if we need to merge this one.

Hi, I just checked #33424. We might still need this one, as the exceptions are raised in the execute method before we enter deferring. (but the solution in #33424 can still save us time handling these kinds of stuff in execute_complete which is still great!)

[hussein-awala]

Hi, I just checked #33424. We might still need this one, as the exceptions are raised in the execute method before we enter deferring. (but the solution in #33424 can still save us time handling these kinds of stuff in execute_complete which is still great!)

I agree that #33424 is not completed yet, but I don't think that this PR could fix the issue. The concept of soft fail is if the sensor fails, we raise the specific skip exception which is processed by the job runner to change the state to skipped instead of failed.

Could you add a test for the operator? (the added test is for the trigger and not the task)

[hussein-awala]

How could raising an AirflowSkipException in the trigger pass the task state to skipped? The trigger exception is not propagated to task resuming if I am not mistaken. WDYT?

[Lee-W]

Take DateTimeSensorAsync as an example. The second line will raise the AirflowSkipException before we execute self.defer

        self.defer(
            trigger=DateTimeTrigger(moment=target_dttm, soft_fail=self.soft_fail),
            method_name="execute_complete",
        )

[hussein-awala]

What about:

         try:
           self.defer(
              trigger=DateTimeTrigger(moment=target_dttm, soft_fail=self.soft_fail),
              method_name="execute_complete",
          )
        except Exception as e:
          raise AirflowException(str(e))

or just changing raise TypeError(message) and raise ValueError to AirflowException.

In both cases the exception will be cached by BaseSensor, and a AirflowSkipException will be raised when soft_fail is True. I'm trying to avoid complicating the trigger code, and duplicating the soft fail logic.

cc: @uranusjr @potiuk WDYT?

[Lee-W]

Or what do you think about my first commit? b2f2662 I was trying not to duplicate the check logic and thus moved it to the trigger

[hussein-awala]

for me it's much better

[hussein-awala]

Since it's a problem in the initialization of the trigger, and it is not a runtime issue, what about raising an exception in parsing time, to give the user a fast feedback about the wrong parameters, for example:

diff --git a/airflow/sensors/date_time.py b/airflow/sensors/date_time.py
index 1425028870..a036d09e01 100644
--- a/airflow/sensors/date_time.py
+++ b/airflow/sensors/date_time.py
@@ -69,6 +69,7 @@ class DateTimeSensor(BaseSensorOperator):
             raise TypeError(
                 f"Expected str or datetime.datetime type for target_time. Got {type(target_time)}"
             )
+        self.trigger = DateTimeTrigger(moment=timezone.parse(self.target_time))
 
     def poke(self, context: Context) -> bool:
         self.log.info("Checking if the time (%s) has come", self.target_time)
@@ -87,7 +88,7 @@ class DateTimeSensorAsync(DateTimeSensor):
 
     def execute(self, context: Context):
         self.defer(
-            trigger=DateTimeTrigger(moment=timezone.parse(self.target_time)),
+            trigger=self.trigger,
             method_name="execute_complete",
         )

This way the dag parsing will fail, and the user will fix the issue.

WDYT?

[Lee-W]

Since it's a problem in the initialization of the trigger, and it is not a runtime issue, what about raising an exception in parsing time, to give the user a fast feedback about the wrong parameters, for example:

diff --git a/airflow/sensors/date_time.py b/airflow/sensors/date_time.py
index 1425028870..a036d09e01 100644
--- a/airflow/sensors/date_time.py
+++ b/airflow/sensors/date_time.py
@@ -69,6 +69,7 @@ class DateTimeSensor(BaseSensorOperator):
             raise TypeError(
                 f"Expected str or datetime.datetime type for target_time. Got {type(target_time)}"
             )
+        self.trigger = DateTimeTrigger(moment=timezone.parse(self.target_time))
 
     def poke(self, context: Context) -> bool:
         self.log.info("Checking if the time (%s) has come", self.target_time)
@@ -87,7 +88,7 @@ class DateTimeSensorAsync(DateTimeSensor):
 
     def execute(self, context: Context):
         self.defer(
-            trigger=DateTimeTrigger(moment=timezone.parse(self.target_time)),
+            trigger=self.trigger,
             method_name="execute_complete",
         )

This way the dag parsing will fail, and the user will fix the issue.

WDYT?

This might not be doable for https://github.com/apache/airflow/blob/a6d803303bf71a84e9e59e94d9c088e3120bedb5/airflow/sensors/time_delta.py#L65-L66

[Lee-W]

hi @hussein-awala, could you please check whether it works for you when you have time? Thanks!