codewithsruthi commented on code in PR #72083:
URL: https://github.com/apache/airflow/pull/72083#discussion_r3962931657
##########
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:
Agreed — that test also passed on `main`, so it did not lock in this change.
I dropped it. The existing `start_from_trigger` cases still cover
`trigger_kwargs`, and the exemption removal is what the constructor-logic hook
actually checks.
##########
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:
Agreed. Renamed it to `_build_direct_submit_trigger_kwargs` and added a
short comment that it stays outside `__init__` so the constructor-logic hook
does not see the template-field reads, while `start_from_trigger` still needs
the copy at construct time.
##########
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:
Agreed — the previous code already allocated a fresh `StartTriggerArgs`, so
there was no cross-task leak to fix. I reworded the PR description. Left the
in-code comment so it stays consistent with the other `start_from_trigger`
operators.
--
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]
