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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
