Lee-W commented on code in PR #73144:
URL: https://github.com/apache/airflow/pull/73144#discussion_r4011220791
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -181,6 +181,18 @@ def on_task_instance_running(
if TYPE_CHECKING:
assert task
start_date = task_instance.start_date
+
+ # Only sensors in reschedule mode re-run without a new try_number,
so only they can
+ # repeat a START for a run id that already emitted one. Other
operators can also have
+ # TaskReschedule rows (a missing-DAG startup failure writes them
before any listener
Review Comment:
```suggestion
# TaskReschedule rows (a missing-Dag startup failure writes them
before any listener
```
##########
providers/openlineage/tests/unit/openlineage/plugins/test_listener.py:
##########
@@ -2309,6 +2314,64 @@ def
test_on_task_instance_running_correctly_calls_openlineage_adapter_run_id_met
map_index=-1,
)
+ @pytest.mark.parametrize(
+ ("sensor_mode", "task_reschedule_count", "should_emit_start"),
+ [
+ pytest.param("reschedule", 0, True,
id="sensor_first_poke_emits_start"),
+ pytest.param("reschedule", 3, False,
id="sensor_later_poke_skips_start"),
+ pytest.param("poke", 3, True,
id="poke_sensor_ignores_reschedule_rows"),
+ pytest.param(None, 3, True,
id="non_sensor_with_reschedule_rows_still_emits"),
+ ],
+ )
+ @mock.patch("airflow.providers.openlineage.conf.debug_mode",
return_value=True)
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_debug_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_task_parent_run_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_run_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_airflow_mapped_task_facet")
+
@mock.patch("airflow.providers.openlineage.plugins.listener.get_user_provided_run_facets")
+ @mock.patch(
+
"airflow.providers.openlineage.plugins.listener.OpenLineageListener._execute",
new=regular_call
+ )
+ def
test_on_task_instance_running_skips_start_event_only_for_rescheduled_sensors(
+ self,
+ mock_get_user_provided_run_facets,
+ mock_get_airflow_mapped_task_facet,
+ mock_get_airflow_run_facet,
+ mock_get_task_parent_run_facet,
+ mock_debug_facet,
+ mock_debug_mode,
+ sensor_mode,
+ task_reschedule_count,
+ should_emit_start,
+ ):
+ """Only a reschedule-mode sensor may have its repeat START suppressed.
+
+ Such a sensor re-runs to poke again without incrementing
``try_number``, so the run id is
+ unchanged and a second START would show consumers the same run
entering RUNNING twice. Any
+ operator can accumulate ``TaskReschedule`` rows though, because a
missing-DAG startup failure
Review Comment:
```suggestion
operator can accumulate ``TaskReschedule`` rows though, because a
missing-Dag startup failure
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
