Re: [PR] Move DataprocSubmitJobOperator start-trigger kwargs out of __init__ [airflow]

["rjgoyln (via GitHub)" <gi...@apache.org>]

rjgoyln commented on code in PR #72083:
URL: https://github.com/apache/airflow/pull/72083#discussion_r3955844616


##########
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py:
##########
@@ -2005,23 +2005,26 @@ def __init__(
         self.openlineage_inject_transport_info = openlineage_inject_transport_info
 
         if self.deferrable and self.start_from_trigger:
-            self.start_trigger_args = StartTriggerArgs(
-                trigger_cls="airflow.providers.google.cloud.triggers.dataproc.DataprocSubmitJobDirectTrigger",
-                trigger_kwargs={
-                    "job": self.job,
-                    "project_id": self.project_id,
-                    "region": self.region,
-                    "gcp_conn_id": self.gcp_conn_id,
-                    "impersonation_chain": self.impersonation_chain,
-                    "polling_interval_seconds": self.polling_interval_seconds,
-                    "cancel_on_kill": self.cancel_on_kill,
-                    "request_id": self.request_id,
-                },
-                next_method="execute_complete",
-                next_kwargs=None,
-                timeout=None,
+            # Replaced rather than mutated: ``start_trigger_args`` is a class attribute, so
+            # assigning through it would overwrite the arguments of every other task built
+            # from this operator.
+            self.start_trigger_args = replace(
+                self.start_trigger_args,
+                trigger_kwargs=self.build_direct_submit_trigger_kwargs(),
             )
 
+    def build_direct_submit_trigger_kwargs(self) -> dict[str, Any]:

Review Comment:
   Could we make the helper private (`_build_direct_submit_trigger_kwargs`) since it’s an implementation detail?
   
   Also, a short comment explaining why it stays outside `__init__` would be helpful, since the reads still intentionally happen at construction time to satisfy `start_from_trigger`.
   



##########
providers/google/tests/unit/google/cloud/operators/test_dataproc.py:
##########
@@ -2521,6 +2521,21 @@ def test_start_from_trigger_without_deferrable_does_not_set_args(self):
         assert op.start_from_trigger is True
         assert op.start_trigger_args.trigger_kwargs == {}
 
+    def test_start_from_trigger_stores_templated_fields_verbatim(self):

Review Comment:
   This test also passes on `main`, so it cannot be failed by reverting the change. The existing test above already covers `trigger_kwargs`, including `job` and `project_id`.
   
   Since this is a pure refactor, could we either drop this test or repoint it to cover the `replace()` behaviour: preserving `next_method` / `next_kwargs` from the class attribute and leaving the class attribute unchanged?
   



##########
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py:
##########
@@ -2005,23 +2005,26 @@ def __init__(
         self.openlineage_inject_transport_info = openlineage_inject_transport_info
 
         if self.deferrable and self.start_from_trigger:
-            self.start_trigger_args = StartTriggerArgs(
-                trigger_cls="airflow.providers.google.cloud.triggers.dataproc.DataprocSubmitJobDirectTrigger",
-                trigger_kwargs={
-                    "job": self.job,
-                    "project_id": self.project_id,
-                    "region": self.region,
-                    "gcp_conn_id": self.gcp_conn_id,
-                    "impersonation_chain": self.impersonation_chain,
-                    "polling_interval_seconds": self.polling_interval_seconds,
-                    "cancel_on_kill": self.cancel_on_kill,
-                    "request_id": self.request_id,
-                },
-                next_method="execute_complete",
-                next_kwargs=None,
-                timeout=None,
+            # Replaced rather than mutated: ``start_trigger_args`` is a class attribute, so

Review Comment:
   Keeping the comment for consistency is fine, but could we reword the PR description? The previous code already created a fresh `StartTriggerArgs`, so there was no cross-task leakage. The current wording may imply this PR fixed a bug that did not exist here.
   



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org