SameerMesiah97 commented on code in PR #70103: URL: https://github.com/apache/airflow/pull/70103#discussion_r3744929476
########## providers/snowflake/src/airflow/providers/snowflake/triggers/snowpark_containers.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import asyncio +import time +from collections.abc import AsyncIterator +from enum import Enum +from typing import Any + +from airflow.providers.common.sql.hooks.handlers import fetch_one_handler +from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook +from airflow.triggers.base import BaseTrigger, TriggerEvent + + +class SnowparkContainerJobStatus(str, Enum): + """Statuses of a Snowpark Container Services job service.""" + + PENDING = "PENDING" + RUNNING = "RUNNING" + CANCELLING = "CANCELLING" + SUSPENDING = "SUSPENDING" + DELETING = "DELETING" + DONE = "DONE" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + INTERNAL_ERROR = "INTERNAL_ERROR" + + +TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.DONE, + SnowparkContainerJobStatus.FAILED, + SnowparkContainerJobStatus.CANCELLED, + SnowparkContainerJobStatus.INTERNAL_ERROR, + } +) +NON_TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.PENDING, + SnowparkContainerJobStatus.RUNNING, + SnowparkContainerJobStatus.CANCELLING, + SnowparkContainerJobStatus.SUSPENDING, + SnowparkContainerJobStatus.DELETING, + } +) Review Comment: These statuses describe the Snowpark job/service rather than the trigger itself, and they're also consumed by the operator. Would it make more sense to keep the enum/status sets with the hook (similar to dbt Cloud), so the operator module does not need to import shared abstractions from the trigger module? ########## providers/snowflake/src/airflow/providers/snowflake/triggers/snowpark_containers.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import asyncio +import time +from collections.abc import AsyncIterator +from enum import Enum +from typing import Any + +from airflow.providers.common.sql.hooks.handlers import fetch_one_handler +from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook +from airflow.triggers.base import BaseTrigger, TriggerEvent + + +class SnowparkContainerJobStatus(str, Enum): + """Statuses of a Snowpark Container Services job service.""" + + PENDING = "PENDING" + RUNNING = "RUNNING" + CANCELLING = "CANCELLING" + SUSPENDING = "SUSPENDING" + DELETING = "DELETING" + DONE = "DONE" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + INTERNAL_ERROR = "INTERNAL_ERROR" + + +TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.DONE, + SnowparkContainerJobStatus.FAILED, + SnowparkContainerJobStatus.CANCELLED, + SnowparkContainerJobStatus.INTERNAL_ERROR, + } +) +NON_TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.PENDING, + SnowparkContainerJobStatus.RUNNING, + SnowparkContainerJobStatus.CANCELLING, + SnowparkContainerJobStatus.SUSPENDING, + SnowparkContainerJobStatus.DELETING, + } +) + + +class SnowparkContainerJobTrigger(BaseTrigger): + """ + Poll a Snowpark Container Services job until it reaches a terminal status. + + :param job_name: name of the submitted job service to poll. + :param snowflake_conn_id: reference to the Snowflake connection id. + :param poll_interval: seconds to sleep between ``DESCRIBE SERVICE`` polls. + :param end_time: epoch deadline (``time.time()`` seconds) after which a ``timeout`` + event is emitted. + :param database: (Optional) name of database. + :param schema: (Optional) name of schema. + :param role: (Optional) name of role. + :param warehouse: (Optional) name of warehouse. Review Comment: I think you should add the default value for each parameter to the docstring entries. ########## providers/snowflake/tests/unit/snowflake/operators/test_snowpark_containers.py: ########## @@ -280,3 +312,58 @@ def test_execute_skips_drop_when_disabled(self, mock_hook_cls, mock_submit, mock op = _make_operator(drop_on_completion=False) op.execute(context=None) mock_hook.run.assert_not_called() + + @mock.patch.object(SnowparkContainerJobOperator, "_submit_job", return_value=JOB_NAME) + def test_execute_defers_when_deferrable(self, mock_submit): + op = _make_operator(deferrable=True) + with pytest.raises(TaskDeferred) as exc: + op.execute(context=None) + assert isinstance(exc.value.trigger, SnowparkContainerJobTrigger) + assert exc.value.trigger.job_name == JOB_NAME + assert exc.value.method_name == "execute_complete" Review Comment: If you implement the deadline handling suggested above, could we add coverage for it here? In particular, an explicit `execution_timeout` should determine the trigger's execution deadline, while the timeout passed to `defer()` should include the additional polling buffer. ########## providers/snowflake/tests/unit/snowflake/triggers/test_snowpark_containers.py: ########## @@ -0,0 +1,143 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import time +from unittest import mock + +import pytest + +from airflow.providers.snowflake.triggers.snowpark_containers import SnowparkContainerJobTrigger +from airflow.triggers.base import TriggerEvent + +TRIGGER_PATH = "airflow.providers.snowflake.triggers.snowpark_containers" +CLASSPATH = f"{TRIGGER_PATH}.SnowparkContainerJobTrigger" +HOOK = f"{TRIGGER_PATH}.SnowflakeHook" + +JOB_NAME = "TEST_JOB" +CONN_ID = "snowflake_default" +POLL_INTERVAL = 1.0 + + +class TestSnowparkContainerJobTrigger: Review Comment: If you implement the suggested deadline handling in the trigger above, these tests will need updating as well to verify that the timeout takes precedence once the deadline has been reached, including over a terminal status. ########## providers/snowflake/src/airflow/providers/snowflake/operators/snowpark_containers.py: ########## @@ -242,10 +247,35 @@ def execute(self, context: Context) -> str: raise RuntimeError("Job name was not returned") if not self.wait_for_completion: return self.job_name + if self.deferrable: + self.defer( + trigger=SnowparkContainerJobTrigger( + job_name=self.job_name, + snowflake_conn_id=self.snowflake_conn_id, + poll_interval=self.poll_interval, + end_time=time.time() + self.timeout, + database=self.database, + schema=self.schema, + role=self.role, + warehouse=self.warehouse, + ), + # Pad past the trigger's end_time so its timeout event, which drops the service, + # fires before this hard backstop. A user-set execution_timeout takes precedence. + timeout=self.execution_timeout or timedelta(seconds=self.timeout + self.poll_interval + 60), Review Comment: I don't think the timeout passed to `defer() `is necessarily consistent with the deadline used by the trigger. The trigger's `end_time` is always derived from self.timeout, whereas `defer(timeout=...) `uses self.execution_timeout when one is provided. For example, `execution_timeout=60s` with the `default timeout=86400s` gives the framework a 60s timeout while the trigger still has a 24h deadline. This may result in an orphaned job as the trigger will timeout before the timeout event can be emitted (which will be caught by `execute_complete` to trigger service cancellation). I know you added an on_kill method to potentially handle this situation but looking at the BaseTrigger class, it appears that this specific method is inert in timeout scenarios. We could pass the execution deadline to the trigger as well and let it determine the applicable deadline, (please see what has been done for the Airbyte provider). The timeout passed to `defer()` can then be padded slightly so the trigger has an opportunity to emit the corresponding timeout event first. ########## providers/snowflake/src/airflow/providers/snowflake/triggers/snowpark_containers.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import asyncio +import time +from collections.abc import AsyncIterator +from enum import Enum +from typing import Any + +from airflow.providers.common.sql.hooks.handlers import fetch_one_handler +from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook +from airflow.triggers.base import BaseTrigger, TriggerEvent + + +class SnowparkContainerJobStatus(str, Enum): + """Statuses of a Snowpark Container Services job service.""" + + PENDING = "PENDING" + RUNNING = "RUNNING" + CANCELLING = "CANCELLING" + SUSPENDING = "SUSPENDING" + DELETING = "DELETING" + DONE = "DONE" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + INTERNAL_ERROR = "INTERNAL_ERROR" + + +TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.DONE, + SnowparkContainerJobStatus.FAILED, + SnowparkContainerJobStatus.CANCELLED, + SnowparkContainerJobStatus.INTERNAL_ERROR, + } +) +NON_TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.PENDING, + SnowparkContainerJobStatus.RUNNING, + SnowparkContainerJobStatus.CANCELLING, + SnowparkContainerJobStatus.SUSPENDING, + SnowparkContainerJobStatus.DELETING, + } +) + + +class SnowparkContainerJobTrigger(BaseTrigger): + """ + Poll a Snowpark Container Services job until it reaches a terminal status. + + :param job_name: name of the submitted job service to poll. + :param snowflake_conn_id: reference to the Snowflake connection id. + :param poll_interval: seconds to sleep between ``DESCRIBE SERVICE`` polls. + :param end_time: epoch deadline (``time.time()`` seconds) after which a ``timeout`` + event is emitted. + :param database: (Optional) name of database. + :param schema: (Optional) name of schema. + :param role: (Optional) name of role. + :param warehouse: (Optional) name of warehouse. + """ + + def __init__( + self, + job_name: str, + snowflake_conn_id: str, + poll_interval: float, + end_time: float, + database: str | None = None, + schema: str | None = None, + role: str | None = None, + warehouse: str | None = None, + ) -> None: + super().__init__() + self.job_name = job_name + self.snowflake_conn_id = snowflake_conn_id + self.poll_interval = poll_interval + self.end_time = end_time + self.database = database + self.schema = schema + self.role = role + self.warehouse = warehouse + + def serialize(self) -> tuple[str, dict[str, Any]]: + """Serialize SnowparkContainerJobTrigger arguments and class path.""" + return ( + "airflow.providers.snowflake.triggers.snowpark_containers.SnowparkContainerJobTrigger", + { + "job_name": self.job_name, + "snowflake_conn_id": self.snowflake_conn_id, + "poll_interval": self.poll_interval, + "end_time": self.end_time, + "database": self.database, + "schema": self.schema, + "role": self.role, + "warehouse": self.warehouse, + }, + ) + + def _get_hook(self) -> SnowflakeHook: + """Build a ``SnowflakeHook`` from the trigger's connection settings.""" + return SnowflakeHook( + snowflake_conn_id=self.snowflake_conn_id, + warehouse=self.warehouse, + database=self.database, + schema=self.schema, + role=self.role, + ) + + async def _describe_status(self, hook: SnowflakeHook) -> str | None: + """Return the job's current status via ``DESCRIBE SERVICE``, or ``None`` if absent.""" + # SnowflakeHook is synchronous. Run the blocking poll off the event loop so a + # single query does not stall every other trigger on this triggerer. + response: Any = await asyncio.to_thread( + hook.run, + f"DESCRIBE SERVICE {self.job_name}", + handler=fetch_one_handler, + return_dictionaries=True, + ) + return response.get("status") if response else None Review Comment: Just checking if 'status' is guaranteed to be upper case in the response object. ########## providers/snowflake/src/airflow/providers/snowflake/triggers/snowpark_containers.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import asyncio +import time +from collections.abc import AsyncIterator +from enum import Enum +from typing import Any + +from airflow.providers.common.sql.hooks.handlers import fetch_one_handler +from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook +from airflow.triggers.base import BaseTrigger, TriggerEvent + + +class SnowparkContainerJobStatus(str, Enum): + """Statuses of a Snowpark Container Services job service.""" + + PENDING = "PENDING" + RUNNING = "RUNNING" + CANCELLING = "CANCELLING" + SUSPENDING = "SUSPENDING" + DELETING = "DELETING" + DONE = "DONE" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + INTERNAL_ERROR = "INTERNAL_ERROR" + + +TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.DONE, + SnowparkContainerJobStatus.FAILED, + SnowparkContainerJobStatus.CANCELLED, + SnowparkContainerJobStatus.INTERNAL_ERROR, + } +) +NON_TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.PENDING, + SnowparkContainerJobStatus.RUNNING, + SnowparkContainerJobStatus.CANCELLING, + SnowparkContainerJobStatus.SUSPENDING, + SnowparkContainerJobStatus.DELETING, + } +) + + +class SnowparkContainerJobTrigger(BaseTrigger): + """ + Poll a Snowpark Container Services job until it reaches a terminal status. + + :param job_name: name of the submitted job service to poll. + :param snowflake_conn_id: reference to the Snowflake connection id. + :param poll_interval: seconds to sleep between ``DESCRIBE SERVICE`` polls. + :param end_time: epoch deadline (``time.time()`` seconds) after which a ``timeout`` + event is emitted. + :param database: (Optional) name of database. + :param schema: (Optional) name of schema. + :param role: (Optional) name of role. + :param warehouse: (Optional) name of warehouse. + """ + + def __init__( + self, + job_name: str, + snowflake_conn_id: str, + poll_interval: float, + end_time: float, + database: str | None = None, + schema: str | None = None, + role: str | None = None, + warehouse: str | None = None, + ) -> None: + super().__init__() + self.job_name = job_name + self.snowflake_conn_id = snowflake_conn_id + self.poll_interval = poll_interval + self.end_time = end_time + self.database = database + self.schema = schema + self.role = role + self.warehouse = warehouse + + def serialize(self) -> tuple[str, dict[str, Any]]: + """Serialize SnowparkContainerJobTrigger arguments and class path.""" + return ( + "airflow.providers.snowflake.triggers.snowpark_containers.SnowparkContainerJobTrigger", + { + "job_name": self.job_name, + "snowflake_conn_id": self.snowflake_conn_id, + "poll_interval": self.poll_interval, + "end_time": self.end_time, + "database": self.database, + "schema": self.schema, + "role": self.role, + "warehouse": self.warehouse, + }, + ) + + def _get_hook(self) -> SnowflakeHook: + """Build a ``SnowflakeHook`` from the trigger's connection settings.""" + return SnowflakeHook( + snowflake_conn_id=self.snowflake_conn_id, + warehouse=self.warehouse, + database=self.database, + schema=self.schema, + role=self.role, + ) + + async def _describe_status(self, hook: SnowflakeHook) -> str | None: + """Return the job's current status via ``DESCRIBE SERVICE``, or ``None`` if absent.""" + # SnowflakeHook is synchronous. Run the blocking poll off the event loop so a + # single query does not stall every other trigger on this triggerer. + response: Any = await asyncio.to_thread( + hook.run, + f"DESCRIBE SERVICE {self.job_name}", + handler=fetch_one_handler, + return_dictionaries=True, + ) + return response.get("status") if response else None + + async def run(self) -> AsyncIterator[TriggerEvent]: + """Poll the job status and yield exactly one terminal event.""" + hook = self._get_hook() + while True: + try: + status = await self._describe_status(hook=hook) + except Exception as e: Review Comment: Do we want a single transient `DESCRIBE SERVICE `failure to terminate the trigger, given these jobs may run for hours, or should polling failures be retried up to the deadline? ########## providers/snowflake/src/airflow/providers/snowflake/triggers/snowpark_containers.py: ########## @@ -0,0 +1,179 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import asyncio +import time +from collections.abc import AsyncIterator +from enum import Enum +from typing import Any + +from airflow.providers.common.sql.hooks.handlers import fetch_one_handler +from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook +from airflow.triggers.base import BaseTrigger, TriggerEvent + + +class SnowparkContainerJobStatus(str, Enum): + """Statuses of a Snowpark Container Services job service.""" + + PENDING = "PENDING" + RUNNING = "RUNNING" + CANCELLING = "CANCELLING" + SUSPENDING = "SUSPENDING" + DELETING = "DELETING" + DONE = "DONE" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + INTERNAL_ERROR = "INTERNAL_ERROR" + + +TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.DONE, + SnowparkContainerJobStatus.FAILED, + SnowparkContainerJobStatus.CANCELLED, + SnowparkContainerJobStatus.INTERNAL_ERROR, + } +) +NON_TERMINAL_STATUSES: frozenset[SnowparkContainerJobStatus] = frozenset( + { + SnowparkContainerJobStatus.PENDING, + SnowparkContainerJobStatus.RUNNING, + SnowparkContainerJobStatus.CANCELLING, + SnowparkContainerJobStatus.SUSPENDING, + SnowparkContainerJobStatus.DELETING, + } +) + + +class SnowparkContainerJobTrigger(BaseTrigger): + """ + Poll a Snowpark Container Services job until it reaches a terminal status. + + :param job_name: name of the submitted job service to poll. + :param snowflake_conn_id: reference to the Snowflake connection id. + :param poll_interval: seconds to sleep between ``DESCRIBE SERVICE`` polls. + :param end_time: epoch deadline (``time.time()`` seconds) after which a ``timeout`` + event is emitted. + :param database: (Optional) name of database. + :param schema: (Optional) name of schema. + :param role: (Optional) name of role. + :param warehouse: (Optional) name of warehouse. + """ + + def __init__( + self, + job_name: str, + snowflake_conn_id: str, + poll_interval: float, + end_time: float, + database: str | None = None, + schema: str | None = None, + role: str | None = None, + warehouse: str | None = None, + ) -> None: + super().__init__() + self.job_name = job_name + self.snowflake_conn_id = snowflake_conn_id + self.poll_interval = poll_interval + self.end_time = end_time + self.database = database + self.schema = schema + self.role = role + self.warehouse = warehouse + + def serialize(self) -> tuple[str, dict[str, Any]]: + """Serialize SnowparkContainerJobTrigger arguments and class path.""" + return ( + "airflow.providers.snowflake.triggers.snowpark_containers.SnowparkContainerJobTrigger", + { + "job_name": self.job_name, + "snowflake_conn_id": self.snowflake_conn_id, + "poll_interval": self.poll_interval, + "end_time": self.end_time, + "database": self.database, + "schema": self.schema, + "role": self.role, + "warehouse": self.warehouse, + }, + ) + + def _get_hook(self) -> SnowflakeHook: + """Build a ``SnowflakeHook`` from the trigger's connection settings.""" + return SnowflakeHook( + snowflake_conn_id=self.snowflake_conn_id, + warehouse=self.warehouse, + database=self.database, + schema=self.schema, + role=self.role, + ) + + async def _describe_status(self, hook: SnowflakeHook) -> str | None: + """Return the job's current status via ``DESCRIBE SERVICE``, or ``None`` if absent.""" + # SnowflakeHook is synchronous. Run the blocking poll off the event loop so a + # single query does not stall every other trigger on this triggerer. + response: Any = await asyncio.to_thread( + hook.run, + f"DESCRIBE SERVICE {self.job_name}", + handler=fetch_one_handler, + return_dictionaries=True, + ) + return response.get("status") if response else None + + async def run(self) -> AsyncIterator[TriggerEvent]: + """Poll the job status and yield exactly one terminal event.""" + hook = self._get_hook() + while True: + try: + status = await self._describe_status(hook=hook) + except Exception as e: + yield TriggerEvent({"status": "error", "job_name": self.job_name, "message": str(e)}) + return + + if status in TERMINAL_STATUSES: + yield TriggerEvent({"status": status, "job_name": self.job_name}) + return + + if status not in NON_TERMINAL_STATUSES: + yield TriggerEvent( + { + "status": "error", + "job_name": self.job_name, + "message": f"Job {self.job_name} returned unexpected status: {status}", + } + ) + return + + if time.time() > self.end_time: Review Comment: 1) Change this to `>=` to handle timeouts on the boundary. 2) Should the timeout check happen before polling/evaluating the job status? Currently, if the trigger wakes after `end_time` (or the `DESCRIBE SERVICE` call itself takes us past it) and the service returns a terminal status, we'll emit that terminal event even though the trigger deadline has already expired. If `end_time` is intended to be a hard deadline, I think it should be checked before starting another poll. -- 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]
