kgw7401 commented on code in PR #52005:
URL: https://github.com/apache/airflow/pull/52005#discussion_r2162743354
##########
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:
Thank you for reviewing.
Actually when i ran system test, it failed so that i should fix the code.
And i can't find ref about why the code that you mentioned couldn't be
compatible with AF3. Could you give me ref for that?
--
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]
