VladaZakharova commented on code in PR #52005:
URL: https://github.com/apache/airflow/pull/52005#discussion_r2161138353
##########
providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py:
##########
@@ -214,6 +216,123 @@ async def run(self):
raise e
+class DataprocSubmitJobTrigger(DataprocBaseTrigger):
+ """DataprocSubmitJobTrigger runs on the trigger worker to perform Build
operation."""
+
+ def __init__(
+ self,
+ job: dict,
+ request_id: str | None = None,
+ retry: Retry | _MethodDefault = DEFAULT,
+ timeout: float | None = None,
+ metadata: Sequence[tuple[str, str]] = (),
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.job = job
+ self.request_id = request_id
+ self.retry = retry
+ self.timeout = timeout
+ self.metadata = metadata
+
+ def serialize(self):
+ return (
+
"airflow.providers.google.cloud.triggers.dataproc.DataprocSubmitJobTrigger",
+ {
+ "project_id": self.project_id,
+ "region": self.region,
+ "job": self.job,
+ "request_id": self.request_id,
+ "retry": self.retry,
+ "timeout": self.timeout,
+ "metadata": self.metadata,
+ "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,
+ },
+ )
+
+ @provide_session
+ def get_task_instance(self, session: Session) -> TaskInstance:
+ """
+ Get the task instance for the current task.
+
+ :param session: Sqlalchemy session
+ """
+ query = session.query(TaskInstance).filter(
+ TaskInstance.dag_id == self.task_instance.dag_id,
+ TaskInstance.task_id == self.task_instance.task_id,
+ TaskInstance.run_id == self.task_instance.run_id,
+ TaskInstance.map_index == self.task_instance.map_index,
+ )
+ task_instance = query.one_or_none()
Review Comment:
This code will not be compatible with AF3. If you were looking for solutions
for this functionality, I think it is better to use already existing trigger.
If I understand correctly. Can you please show also your system tests results
for this code, both for AF2 and AF3 as screenshots? thanks
--
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]
