This is an automated email from the ASF dual-hosted git repository.
amoghrajesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6ba8d5a85cb Replace `resume_glue_job_on_retry` with durable execution
for `GlueJobOperator` (#71211)
6ba8d5a85cb is described below
commit 6ba8d5a85cba9e32f07ca8b995d101c0ea860b04
Author: Amogh Desai <[email protected]>
AuthorDate: Thu Aug 13 13:57:59 2026 +0530
Replace `resume_glue_job_on_retry` with durable execution for
`GlueJobOperator` (#71211)
---
providers/amazon/docs/changelog.rst | 11 +
providers/amazon/docs/operators/glue.rst | 96 +++
.../src/airflow/providers/amazon/aws/exceptions.py | 4 +
.../airflow/providers/amazon/aws/operators/glue.py | 328 ++++++---
.../tests/unit/amazon/aws/operators/test_glue.py | 786 +++++++++++++++++----
5 files changed, 1005 insertions(+), 220 deletions(-)
diff --git a/providers/amazon/docs/changelog.rst
b/providers/amazon/docs/changelog.rst
index a63a7bdd078..7cae2b87a29 100644
--- a/providers/amazon/docs/changelog.rst
+++ b/providers/amazon/docs/changelog.rst
@@ -26,6 +26,17 @@
Changelog
---------
+.. warning::
+ On Airflow 3.3+, ``GlueJobOperator``'s ``durable`` parameter now defaults to
``True``: the Glue
+ job run id is persisted to task state store, and a worker crash on retry
reconnects to the
+ existing run instead of starting a duplicate. Pass ``durable=False`` to
restore the previous
+ behavior of always starting a fresh run on retry.
+
+ On Airflow versions below 3.3, ``durable`` still defaults to ``False`` --
upgrading the provider
+ alone does not change behavior there. Set ``durable=True`` explicitly (or
the now-deprecated
+ ``resume_glue_job_on_retry=True``) to opt in to the same reconnect behavior
via an older,
+ scan-based mechanism.
+
9.34.0
......
diff --git a/providers/amazon/docs/operators/glue.rst
b/providers/amazon/docs/operators/glue.rst
index ce97e693f3a..eac6e270a1d 100644
--- a/providers/amazon/docs/operators/glue.rst
+++ b/providers/amazon/docs/operators/glue.rst
@@ -139,6 +139,102 @@ To submit a new AWS Glue job you can use
:class:`~airflow.providers.amazon.aws.o
The same AWS IAM role used for the crawler can be used here as well, but it
will need
policies to provide access to the output location for result data.
+A Glue job run that ends in ``STOPPED`` is treated as a failure, not a success
-- on every
+attempt, first or retry, regardless of ``durable``. Glue's API has no way to
tell a run cancelled
+manually (for example, in the AWS console) apart from one this operator's own
+:meth:`~airflow.providers.amazon.aws.operators.glue.GlueJobOperator.on_kill`
stopped, which happens
+whenever ``stop_job_run_on_kill=True`` and the task is killed -- on SIGTERM, on
+``execution_timeout``, or when the task is cleared while running. The task
fails and a normal
+retry resubmits, rather than a self-inflicted stop being silently reported as
a false success.
+Setting ``durable=False`` does not change this.
+
+Durable execution
+==================
+
+``GlueJobOperator`` submits a job run and then polls it to completion on the
worker. By default
+the operator runs in a *durable* mode that makes this crash-safe: the Glue job
run id is
+persisted to :doc:`task state store
<apache-airflow:core-concepts/task-state-store>` before
+polling begins, so if the worker crashes or is preempted and the task is
retried, the operator
+reconnects to the run that is already executing in Glue instead of starting a
new one.
+
+This matters more for Glue because a Glue job's ``concurrent_run_limit``
defaults to ``1``, so
+submitting a second run while the first is still active does not create a
harmless duplicate, it
+fails outright with ``ConcurrentRunsExceededException`` and the task keeps
retrying against a run
+it can never see. Durable execution turns that retry into a normal reconnect.
+
+On retry the operator checks the prior run's state:
+
+* if it is still starting, running, waiting for capacity, or being stopped,
the operator
+ reconnects and continues polling
+* if it already succeeded, the operator returns immediately without
resubmitting
+* if it failed terminally, or its id has expired and is no longer found, the
operator submits the
+ job fresh
+* if it already stopped, the operator submits the job fresh
+
+If the stored state is already ``STOPPED``, the operator submits fresh rather
than reconnecting to
+it. If a reconnect finds the run still stopping and it settles into
``STOPPED`` while polling, the
+operator raises instead of returning a result -- see above, this applies
regardless of ``durable``.
+
+This protection also applies when ``wait_for_completion=False`` -- even though
that task attempt
+never polls at all, a retry after a successful submission still reconnects
rather than
+resubmitting, since the run id is persisted immediately after submission
regardless of whether the
+task waits for it to finish.
+
+Durable execution requires Airflow 3.3 or newer for the task state store
lookup above. Below
+3.3, ``durable`` has no effect: setting it explicitly only emits a warning,
and its value is
+ignored either way. The deprecated ``resume_glue_job_on_retry`` parameter is
the only way to opt
+into crash recovery there, and it still works via an older mechanism: the
operator checks XCom
+for a cached run id first, then falls back to scanning the job's run history
for a run tagged
+with this task instance's identity, and reconnects if it finds one that is
still active. The
+XCom check only ever succeeds on Airflow 2.x -- every Airflow 3 release clears
task XComs before
+each non-deferral attempt, so on Airflow 3.0-3.2 every retry pays the full
scan.
+
+That older mechanism only activates below 3.3 when
``resume_glue_job_on_retry=True`` is set
+explicitly -- ``durable=True`` does not turn it on there. Upgrading the
provider alone, with no
+DAG change, does not turn it on either: below 3.3, behavior is unchanged from
before this feature
+existed unless ``resume_glue_job_on_retry`` is set. On Airflow 3.3+,
``durable`` defaults to
+``True`` as described above, since the task state store makes it cheap.
+
+Like the persisted state itself, the stored run id isn't deleted
automatically, that only happens
+when someone runs ``airflow state-store clean`` or ``airflow db clean`` (which
also targets the
+``task_state_store`` table). If a task's ``retry_delay`` is longer than
+``[state_store] default_retention_days`` (30 days by default) and cleanup runs
in between, the run
+id won't be there for the next retry, and the operator falls back to the
XCom/scan mechanism
+above rather than reconnecting via task state store. Avoid running cleanup on
a schedule shorter
+than your longest ``retry_delay``.
+
+Clearing a task is treated the same as a retry, which matters specifically for
a task whose job
+already succeeded: clearing does not delete the stored run id, so the next
attempt reads it back
+and returns immediately without submitting anything to Glue. See
+:doc:`apache-airflow:core-concepts/resumable-tasks` for why, and for the
+``[state_store] clear_on_success`` setting that restores "clearing always
resubmits."
+
+To opt out and always start a fresh run on retry, set ``durable=False``:
+
+.. code-block:: python
+
+ glue_job = GlueJobOperator(
+ task_id="glue_job",
+ job_name="my_glue_job",
+ script_location="s3://glue-examples/glue-scripts/sample_aws_glue_job.py",
+ durable=False,
+ )
+
+The task state store lookup above is only used on the synchronous path -- when
``deferrable=True``
+is set, the Triggerer already tracks the run across the wait, so a run id is
never persisted there.
+``durable`` still has an effect on retry, though: a retry of a deferrable task
would otherwise
+resubmit, and with ``concurrent_run_limit=1`` that fails with
``ConcurrentRunsExceededException``
+against the run it can't see. To avoid that, ``durable=True`` (or, below
Airflow 3.3,
+``resume_glue_job_on_retry=True``) tags the job's arguments with this task
instance's identity on
+every attempt and, on retry, scans the job's run history for that tag before
submitting -- the
+same mechanism used as a fallback on the synchronous path.
+
+``durable`` supersedes the deprecated ``resume_glue_job_on_retry`` parameter
on Airflow 3.3+,
+where passing ``resume_glue_job_on_retry`` still works and maps its value onto
``durable``. Below
+3.3, ``resume_glue_job_on_retry`` remains the only working option, since
``durable`` is a no-op
+there. Either way, passing it emits an ``AirflowProviderDeprecationWarning``,
since the parameter
+will be removed once this provider's minimum supported Airflow version reaches
3.3.
+
.. _howto/operator:GlueDataQualityOperator:
Create an AWS Glue Data Quality
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/exceptions.py
b/providers/amazon/src/airflow/providers/amazon/aws/exceptions.py
index 2591d33bfdf..fc6d668ae74 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/exceptions.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/exceptions.py
@@ -78,3 +78,7 @@ class
NeptuneImportTaskCancellationFailedError(AirflowException):
class NeptuneImportTaskFailedError(AirflowException):
"""Raised when a Neptune Analytics import task fails to complete
successfully."""
+
+
+class GlueJobRunStoppedError(AirflowException):
+ """Raised when a Glue job run finishes in a state that is not a real
success."""
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py
index 30077380eb6..c488965ddde 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py
@@ -19,11 +19,14 @@ from __future__ import annotations
import os
import urllib.parse
+import warnings
from collections.abc import Sequence
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
from botocore.exceptions import ClientError
+from airflow.exceptions import AirflowProviderDeprecationWarning
+from airflow.providers.amazon.aws.exceptions import GlueJobRunStoppedError
from airflow.providers.amazon.aws.hooks.glue import GlueDataQualityHook,
GlueJobHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.amazon.aws.links.glue import GlueJobRunDetailsLink
@@ -35,17 +38,65 @@ from airflow.providers.amazon.aws.triggers.glue import (
)
from airflow.providers.amazon.aws.utils import validate_execute_complete_event
from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
+from airflow.providers.amazon.version_compat import AIRFLOW_V_3_3_PLUS
from airflow.providers.common.compat.openlineage.utils.spark import (
inject_parent_job_information_into_glue_arguments,
inject_transport_information_into_glue_arguments,
)
from airflow.providers.common.compat.sdk import AirflowException, conf
+_DURABLE_UNSET = object()
+
+
+def _warn_and_disable_durable_pre_3_3(durable: Any) -> bool:
+ """Shared by the <3.3 compat stub: durable has no effect below 3.3, warn
if it was set."""
+ if durable is not _DURABLE_UNSET:
+ warnings.warn(
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
+ stacklevel=3,
+ )
+ return False
+
+
+# ResumableJobMixin only exists on Airflow 3.3+; this provider still targets
>=2.11. Drop this
+# fallback once the provider's minimum Airflow version is >=3.3.
+try:
+ from airflow.sdk import ResumableJobMixin
+except ImportError:
+
+ class ResumableJobMixin: # type: ignore[no-redef]
+ """Airflow <3.3 stub, task_state_store unavailable, always submits
fresh."""
+
+ external_id_key: str = "glue_job_run_id"
+
+ def __init__(self, *, durable: Any = _DURABLE_UNSET, **kwargs: Any) ->
None:
+ super().__init__(**kwargs)
+ self.durable = _warn_and_disable_durable_pre_3_3(durable)
+
+ def execute_resumable(self, context):
+ external_id = self.submit_job(context)
+ self.poll_until_complete(external_id, context)
+ return self.get_job_result(external_id, context)
+
+
if TYPE_CHECKING:
+ from pydantic import JsonValue
+
from airflow.sdk import Context
+# Glue job run states, see
+#
https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-jobs-runs.html#aws-glue-api-jobs-runs-JobRun
+# STOPPED is deliberately NOT a success state here, unlike
GlueJobHook.job_completion: Glue can't
+# tell a console cancellation apart from on_kill stopping its own run, so
treating STOPPED as
+# failure avoids silently reporting a self-inflicted stop as success.
+JOB_RUN_SUCCESS_STATES = ("SUCCEEDED",)
+JOB_RUN_TERMINAL_STATES = (*JOB_RUN_SUCCESS_STATES, "STOPPED", "FAILED",
"TIMEOUT", "ERROR", "EXPIRED")
+# Synthetic state for a run id Glue no longer knows about, so a retry submits
fresh rather than failing.
+NOT_FOUND_STATE = "NOT_FOUND"
+
-class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
+class GlueJobOperator(ResumableJobMixin, AwsBaseOperator[GlueJobHook]):
"""
Create an AWS Glue Job.
@@ -90,6 +141,12 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
Defaults to the ``openlineage.spark_inject_transport_info`` config
value.
:param waiter_delay: Time in seconds to wait between status checks.
(default: 60)
:param waiter_max_attempts: Maximum number of attempts to check for job
completion. (default: 20)
+ :param resume_glue_job_on_retry: deprecated, use ``durable`` instead.
+ :param durable: When ``True``, the Glue job run id is persisted to task
state before polling
+ begins. A worker crash on retry reconnects to the existing run instead
of starting a
+ duplicate. Defaults to ``True`` on Airflow 3.3+, which uses task state
store for the
+ persisted lookup; on earlier versions it defaults to ``False`` and, if
set explicitly,
+ recovers the run by searching job runs for the task's
``--airflow_task_uuid`` argument.
:param aws_conn_id: The Airflow connection used for AWS credentials.
If this is ``None`` or empty then the default boto3 behaviour is used.
If
running Airflow in a distributed manner and aws_conn_id is None or
@@ -123,6 +180,7 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
operator_extra_links = (GlueJobRunDetailsLink(),)
TASK_UUID_ARG = "--airflow_task_uuid"
+ external_id_key = "glue_job_run_id"
def __init__(
self,
@@ -149,7 +207,8 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
job_poll_interval: int | float = 6,
waiter_delay: int = 60,
waiter_max_attempts: int = 75,
- resume_glue_job_on_retry: bool = False,
+ resume_glue_job_on_retry: bool | None = None,
+ durable: bool | None = None,
openlineage_inject_parent_job_info: bool = conf.getboolean(
"openlineage", "spark_inject_parent_job_info", fallback=False
),
@@ -158,7 +217,28 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
),
**kwargs,
):
+ if resume_glue_job_on_retry is not None:
+ warnings.warn(
+ "`resume_glue_job_on_retry` is deprecated and will be removed
once this provider's "
+ "minimum supported Airflow version reaches 3.3. "
+ + (
+ "Use `durable` instead."
+ if AIRFLOW_V_3_3_PLUS
+ else "On Airflow 3.3+, use `durable` instead."
+ ),
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
+ if AIRFLOW_V_3_3_PLUS:
+ kwargs.setdefault("durable", resume_glue_job_on_retry)
+ # durable is also named parameter here (not left to **kwargs) so
default_args={"durable": ...} reaches
+ # it on every supported Airflow version.
+ if durable is not None:
+ kwargs["durable"] = durable
super().__init__(**kwargs)
+ if not AIRFLOW_V_3_3_PLUS and resume_glue_job_on_retry is not None:
+ # durable itself has no effect below 3.3, so we take value of
resume_glue_job_on_retry instead.
+ self.durable = resume_glue_job_on_retry
self.job_name = job_name
self.job_desc = job_desc
self.script_location = script_location
@@ -185,7 +265,6 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
self.s3_script_location: str | None = None
self.waiter_delay = waiter_delay
self.waiter_max_attempts = waiter_max_attempts
- self.resume_glue_job_on_retry = resume_glue_job_on_retry
self.openlineage_inject_parent_job_info =
openlineage_inject_parent_job_info
self.openlineage_inject_transport_info =
openlineage_inject_transport_info
@@ -250,6 +329,8 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
return script_args, task_uuid
def _find_job_run_id_by_task_uuid(self, task_uuid: str) -> tuple[str, str]
| None:
+ # Unbounded walk with no page cap; a no-match run is the common retry
shape and pays the
+ # full scan. Tracked at https://github.com/apache/airflow/issues/71489.
next_token: str | None = None
while True:
request = {"JobName": self.job_name, "MaxResults": 50}
@@ -267,83 +348,20 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
if not next_token:
return None
- def execute(self, context: Context):
+ def execute(self, context: Context) -> str | None:
"""
Execute AWS Glue Job from Airflow.
:return: the current Glue job ID.
"""
- previous_job_run_id = None
- script_args = dict(self.script_args)
- task_uuid = None
-
- if self.openlineage_inject_parent_job_info:
- self.log.info("Injecting OpenLineage parent job information into
Glue job arguments.")
- script_args =
inject_parent_job_information_into_glue_arguments(script_args, context)
- if self.openlineage_inject_transport_info:
- self.log.info("Injecting OpenLineage transport information into
Glue job arguments.")
- script_args =
inject_transport_information_into_glue_arguments(script_args, context)
-
- if self.resume_glue_job_on_retry:
- ti = context["ti"]
- script_args, task_uuid =
self._prepare_script_args_with_task_uuid(context, base_args=script_args)
- previous_job_run_id = ti.xcom_pull(key="glue_job_run_id",
task_ids=ti.task_id)
- if previous_job_run_id:
- try:
- job_run =
self.hook.conn.get_job_run(JobName=self.job_name, RunId=previous_job_run_id)
- state = job_run.get("JobRun", {}).get("JobRunState")
- self.log.info("Previous Glue job_run_id: %s, state: %s",
previous_job_run_id, state)
- if state in ("RUNNING", "STARTING"):
- self._job_run_id = previous_job_run_id
- except Exception:
- self.log.warning("Failed to get previous Glue job run
state", exc_info=True)
- elif task_uuid:
- try:
- existing = self._find_job_run_id_by_task_uuid(task_uuid)
- if existing:
- existing_job_run_id, existing_job_run_state = existing
- self.log.info(
- "Found Glue job_run_id by task UUID: %s, state:
%s",
- existing_job_run_id,
- existing_job_run_state,
- )
- if existing_job_run_state in ("RUNNING", "STARTING"):
- self._job_run_id = existing_job_run_id
- ti.xcom_push(key="glue_job_run_id",
value=self._job_run_id)
- except Exception:
- self.log.warning("Failed to find previous Glue job run by
task UUID", exc_info=True)
-
- if not self._job_run_id:
- self.log.info(
- "Initializing AWS Glue Job: %s. Wait for completion: %s",
- self.job_name,
- self.wait_for_completion,
- )
- glue_job_run = self.hook.initialize_job(script_args,
self.run_job_kwargs)
- self._job_run_id = glue_job_run["JobRunId"]
- context["ti"].xcom_push(key="glue_job_run_id",
value=self._job_run_id)
-
- glue_job_run_url = GlueJobRunDetailsLink.format_str.format(
-
aws_domain=GlueJobRunDetailsLink.get_aws_domain(self.hook.conn_partition),
- region_name=self.hook.conn_region_name,
- job_name=urllib.parse.quote(self.job_name, safe=""),
- job_run_id=self._job_run_id,
- )
- GlueJobRunDetailsLink.persist(
- context=context,
- operator=self,
- region_name=self.hook.conn_region_name,
- aws_partition=self.hook.conn_partition,
- job_name=urllib.parse.quote(self.job_name, safe=""),
- job_run_id=self._job_run_id,
- )
- self.log.info("You can monitor this Glue Job run at: %s",
glue_job_run_url)
-
if self.deferrable:
+ # The Triggerer tracks the run, so no id is persisted to
task_state_store here -- but
+ # durable still reattaches a retry via the task-UUID scan in
submit_job.
+ job_run_id = self.submit_job(context)
self.defer(
trigger=GlueJobCompleteTrigger(
job_name=self.job_name,
- run_id=self._job_run_id,
+ run_id=job_run_id,
verbose=self.verbose,
aws_conn_id=self.aws_conn_id,
waiter_delay=self.waiter_delay,
@@ -352,18 +370,7 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
),
method_name="execute_complete",
)
- elif self.wait_for_completion:
- glue_job_run = self.hook.job_completion(
- self.job_name, self._job_run_id, self.verbose,
self.sleep_before_return
- )
- self.log.info(
- "AWS Glue Job: %s status: %s. Run Id: %s",
- self.job_name,
- glue_job_run["JobRunState"],
- self._job_run_id,
- )
- else:
- self.log.info("AWS Glue Job: %s. Run Id: %s", self.job_name,
self._job_run_id)
+ self.execute_resumable(context)
return self._job_run_id
def execute_complete(self, context: Context, event: dict[str, Any] | None
= None) -> str:
@@ -375,7 +382,7 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
def on_kill(self):
"""Cancel the running AWS Glue Job."""
- if self.stop_job_run_on_kill:
+ if self.stop_job_run_on_kill and self._job_run_id:
self.log.info("Stopping AWS Glue Job: %s. Run Id: %s",
self.job_name, self._job_run_id)
response = self.hook.conn.batch_stop_job_run(
JobName=self.job_name,
@@ -384,6 +391,157 @@ class GlueJobOperator(AwsBaseOperator[GlueJobHook]):
if not response["SuccessfulSubmissions"]:
self.log.error("Failed to stop AWS Glue Job: %s. Run Id: %s",
self.job_name, self._job_run_id)
+ def _set_job_run_id(self, context: Context, job_run_id: str) -> None:
+ """Record the run id and surface its console link; guarded so one
attempt logs it once."""
+ if self._job_run_id == job_run_id:
+ return
+ self._job_run_id = job_run_id
+ context["ti"].xcom_push(key="glue_job_run_id", value=job_run_id)
+ GlueJobRunDetailsLink.persist(
+ context=context,
+ operator=self,
+ region_name=self.hook.conn_region_name,
+ aws_partition=self.hook.conn_partition,
+ job_name=urllib.parse.quote(self.job_name, safe=""),
+ job_run_id=job_run_id,
+ )
+ self.log.info(
+ "You can monitor this Glue Job run at: %s",
+ GlueJobRunDetailsLink.format_str.format(
+
aws_domain=GlueJobRunDetailsLink.get_aws_domain(self.hook.conn_partition),
+ region_name=self.hook.conn_region_name,
+ job_name=urllib.parse.quote(self.job_name, safe=""),
+ job_run_id=job_run_id,
+ ),
+ )
+
+ def _build_script_args(self, context: Context) -> dict:
+ script_args = dict(self.script_args)
+ if self.openlineage_inject_parent_job_info:
+ self.log.info("Injecting OpenLineage parent job information into
Glue job arguments.")
+ script_args =
inject_parent_job_information_into_glue_arguments(script_args, context)
+ if self.openlineage_inject_transport_info:
+ self.log.info("Injecting OpenLineage transport information into
Glue job arguments.")
+ script_args =
inject_transport_information_into_glue_arguments(script_args, context)
+ # Skipped for a synchronous run on 3.3+: task_state_store is the
primary reconnect
+ # mechanism there, so tagging every script's args is not worth it just
for the narrow
+ # crash-before-persist window this tag would otherwise cover.
+ if self.durable and (self.deferrable or not AIRFLOW_V_3_3_PLUS):
+ script_args, _ = self._prepare_script_args_with_task_uuid(context,
base_args=script_args)
+ return script_args
+
+ def _find_previous_job_run(self, context: Context, task_uuid: str) -> str
| None:
+ """
+ Look for a Glue job run this task instance already started.
+
+ Checks XCom for a cached run id first, then falls back to a task-UUID
scan. The XCom tier
+ only works on Airflow 2.x; every Airflow 3 release clears task XComs
before each
+ non-deferral attempt, so it always misses there and the scan runs
every time.
+ """
+ ti = context["ti"]
+ previous_job_run_id = ti.xcom_pull(key="glue_job_run_id",
task_ids=ti.task_id)
+ if previous_job_run_id:
+ try:
+ job_run = self.hook.conn.get_job_run(JobName=self.job_name,
RunId=previous_job_run_id)
+ state = job_run.get("JobRun", {}).get("JobRunState")
+ self.log.info("Previous Glue job_run_id: %s, state: %s",
previous_job_run_id, state)
+ if self.is_job_active(state):
+ return previous_job_run_id
+ except Exception:
+ self.log.warning("Failed to get previous Glue job run state",
exc_info=True)
+ else:
+ try:
+ existing = self._find_job_run_id_by_task_uuid(task_uuid)
+ if existing:
+ existing_job_run_id, existing_job_run_state = existing
+ self.log.info(
+ "Found Glue job_run_id by task UUID: %s, state: %s",
+ existing_job_run_id,
+ existing_job_run_state,
+ )
+ if self.is_job_active(existing_job_run_state):
+ return existing_job_run_id
+ except Exception:
+ self.log.warning("Failed to find previous Glue job run by task
UUID", exc_info=True)
+ return None
+
+ def _has_stored_external_id(self, context: Context) -> bool:
+ task_state_store = context.get("task_state_store")
+ return task_state_store is not None and
task_state_store.get(self.external_id_key) is not None
+
+ def submit_job(self, context: Context) -> str:
+ """Start a Glue job run and return its run id, or reconnect to one
this task already started."""
+ script_args = self._build_script_args(context)
+ # Scan only when there's nothing else to go on: first attempt, no
store, or a store that
+ # never recorded this key. The store check itself is skipped on the
deferrable path, where
+ # nothing is ever written to it -- a store error there would otherwise
block the scan.
+ # Deferrable retries could reconnect via task_state_store too,
avoiding this scan entirely,
+ # if ResumableJobMixin exposed its reconnect decision apart from its
polling loop;
+ # tracked at https://github.com/apache/airflow/issues/71485.
+ if (
+ self.durable
+ and (self.deferrable or not AIRFLOW_V_3_3_PLUS)
+ and context["ti"].try_number > 1
+ and (self.deferrable or not self._has_stored_external_id(context))
+ ):
+ existing_job_run_id = self._find_previous_job_run(context,
script_args[self.TASK_UUID_ARG])
+ if existing_job_run_id:
+ self._set_job_run_id(context, existing_job_run_id)
+ return existing_job_run_id
+ self.log.info(
+ "Initializing AWS Glue Job: %s. Wait for completion: %s",
+ self.job_name,
+ self.wait_for_completion,
+ )
+ # A prior get_job_status call may have set this to a stale, terminal
run id while checking
+ # whether to reconnect. Clear it so on_kill has nothing to act on if
initialize_job raises.
+ self._job_run_id = None
+ glue_job_run = self.hook.initialize_job(script_args,
self.run_job_kwargs)
+ # Set before polling so on_kill can stop the run even if the worker
dies immediately after.
+ self._set_job_run_id(context, glue_job_run["JobRunId"])
+ return glue_job_run["JobRunId"]
+
+ def get_job_status(self, external_id: JsonValue, context: Context) -> str:
+ """Query the raw job run state; a run id Glue no longer knows about
degrades to NOT_FOUND."""
+ job_run_id = cast("str", external_id)
+ # This is the first place a reconnecting attempt learns the run id.
+ self._set_job_run_id(context, job_run_id)
+ try:
+ return self.hook.get_job_state(self.job_name, job_run_id)
+ except ClientError as e:
+ if e.response["Error"]["Code"] == "EntityNotFoundException":
+ return NOT_FOUND_STATE
+ raise
+
+ def is_job_active(self, status: str) -> bool:
+ return status not in (*JOB_RUN_TERMINAL_STATES, NOT_FOUND_STATE)
+
+ def is_job_succeeded(self, status: str) -> bool:
+ return status in JOB_RUN_SUCCESS_STATES
+
+ def poll_until_complete(self, external_id: JsonValue, context: Context) ->
None:
+ job_run_id = cast("str", external_id)
+ self._set_job_run_id(context, job_run_id)
+ if not self.wait_for_completion:
+ self.log.info("AWS Glue Job: %s. Run Id: %s", self.job_name,
job_run_id)
+ return
+ glue_job_run = self.hook.job_completion(
+ self.job_name, job_run_id, self.verbose, self.sleep_before_return
+ )
+ state = glue_job_run["JobRunState"]
+ self.log.info("AWS Glue Job: %s status: %s. Run Id: %s",
self.job_name, state, job_run_id)
+ if state not in JOB_RUN_SUCCESS_STATES:
+ # job_completion's own finished_states also accepts STOPPED; this
is narrower on purpose.
+ raise GlueJobRunStoppedError(
+ f"Glue job run {job_run_id} for job {self.job_name} ended in
state {state!r} "
+ "instead of succeeding."
+ )
+
+ def get_job_result(self, external_id: JsonValue, context: Context) -> str:
+ job_run_id = cast("str", external_id)
+ self._set_job_run_id(context, job_run_id)
+ return job_run_id
+
class GlueDataQualityOperator(AwsBaseOperator[GlueDataQualityHook]):
"""
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_glue.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_glue.py
index 6c2c3c316a9..6ea43e60064 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_glue.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_glue.py
@@ -16,6 +16,8 @@
# under the License.
from __future__ import annotations
+import re
+import warnings
from collections.abc import Generator
from datetime import datetime
from typing import TYPE_CHECKING
@@ -24,20 +26,26 @@ from unittest import mock
import boto3
import pytest
from boto3 import client
+from botocore.exceptions import ClientError
from moto import mock_aws
+from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.models.dag import DAG
+from airflow.providers.amazon.aws.exceptions import GlueJobRunStoppedError
from airflow.providers.amazon.aws.hooks.glue import GlueDataQualityHook,
GlueJobHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.amazon.aws.links.glue import GlueJobRunDetailsLink
from airflow.providers.amazon.aws.operators.glue import (
+ _DURABLE_UNSET,
GlueDataQualityOperator,
GlueDataQualityRuleRecommendationRunOperator,
GlueDataQualityRuleSetEvaluationRunOperator,
GlueJobOperator,
+ _warn_and_disable_durable_pre_3_3,
)
from airflow.providers.common.compat.sdk import AirflowException, TaskDeferred
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS
from unit.amazon.aws.utils.test_template_fields import validate_template_fields
if TYPE_CHECKING:
@@ -49,6 +57,16 @@ DAG_ID = "test_dag_id"
JOB_NAME = "test_job_name/with_slash"
JOB_RUN_ID = "11111"
+_DEPRECATION_MESSAGE_PREFIX = (
+ "`resume_glue_job_on_retry` is deprecated and will be removed once this
provider's "
+ "minimum supported Airflow version reaches 3.3. "
+)
+DEPRECATION_MESSAGE_PRE_3_3 = _DEPRECATION_MESSAGE_PREFIX + "On Airflow 3.3+,
use `durable` instead."
+DEPRECATION_MESSAGE_3_3_PLUS = _DEPRECATION_MESSAGE_PREFIX + "Use `durable`
instead."
+EXPECTED_DEPRECATION_MESSAGE = (
+ DEPRECATION_MESSAGE_3_3_PLUS if AIRFLOW_V_3_3_PLUS else
DEPRECATION_MESSAGE_PRE_3_3
+)
+
class TestGlueJobOperator:
@pytest.mark.db_test
@@ -99,6 +117,7 @@ class TestGlueJobOperator:
script_location,
):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location=script_location,
@@ -120,6 +139,7 @@ class TestGlueJobOperator:
@mock.patch.object(GlueJobHook, "get_conn")
def test_role_arn_execute_deferrable(self, _, mock_initialize_job):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -141,6 +161,7 @@ class TestGlueJobOperator:
@mock.patch.object(GlueJobHook, "get_conn")
def test_execute_deferrable(self, _, mock_initialize_job):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -163,6 +184,116 @@ class TestGlueJobOperator:
assert defer.value.trigger.attempts == 75
assert defer.value.trigger.aws_conn_id == "aws_default"
+ @mock.patch.object(GlueJobHook, "conn", new_callable=mock.PropertyMock)
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_deferrable_first_attempt_injects_task_uuid_but_skips_scan(
+ self, mock_get_conn, mock_initialize_job, mock_conn
+ ):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(EXPECTED_DEPRECATION_MESSAGE)}$"
+ ):
+ glue = GlueJobOperator(
+ task_id=TASK_ID,
+ job_name=JOB_NAME,
+ script_location="s3://folder/file",
+ deferrable=True,
+ resume_glue_job_on_retry=True,
+ )
+ mock_initialize_job.return_value = {"JobRunState": "RUNNING",
"JobRunId": JOB_RUN_ID}
+ mock_ti = mock.MagicMock()
+ mock_ti.try_number = 1
+
+ with pytest.raises(TaskDeferred):
+ glue.execute({"ti": mock_ti})
+
+ call_args = mock_initialize_job.call_args[0][0]
+ assert GlueJobOperator.TASK_UUID_ARG in call_args
+ mock_conn.return_value.get_job_runs.assert_not_called()
+
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_deferrable_retry_reattaches_via_task_uuid_scan(self,
mock_get_conn, mock_initialize_job):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(EXPECTED_DEPRECATION_MESSAGE)}$"
+ ):
+ glue = GlueJobOperator(
+ task_id=TASK_ID,
+ job_name=JOB_NAME,
+ script_location="s3://folder/file",
+ deferrable=True,
+ resume_glue_job_on_retry=True,
+ )
+ mock_ti = mock.MagicMock()
+ mock_ti.dag_id = "test_dag_id"
+ mock_ti.task_id = TASK_ID
+ mock_ti.run_id = "manual__2024-01-01T00:00:00+00:00"
+ mock_ti.map_index = -1
+ mock_ti.try_number = 2
+ mock_ti.xcom_pull.return_value = None
+ task_uuid =
f"{mock_ti.dag_id}:{mock_ti.task_id}:{mock_ti.run_id}:{mock_ti.map_index}"
+
+ glue.hook.conn = mock.MagicMock()
+ glue.hook.conn.get_job_runs.return_value = {
+ "JobRuns": [
+ {
+ "Id": JOB_RUN_ID,
+ "Arguments": {GlueJobOperator.TASK_UUID_ARG: task_uuid},
+ "JobRunState": "RUNNING",
+ }
+ ]
+ }
+
+ with pytest.raises(TaskDeferred) as defer:
+ glue.execute({"ti": mock_ti})
+
+ assert defer.value.trigger.run_id == JOB_RUN_ID
+
+ @pytest.mark.skipif(
+ not AIRFLOW_V_3_3_PLUS,
+ reason="task_state_store only exists as an execute() context key on
Airflow 3.3+",
+ )
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_deferrable_retry_reattaches_via_fallback_when_task_store_errors(
+ self, mock_get_conn, mock_initialize_job
+ ):
+ glue = GlueJobOperator(
+ durable=True,
+ task_id=TASK_ID,
+ job_name=JOB_NAME,
+ script_location="s3://folder/file",
+ deferrable=True,
+ )
+ mock_ti = mock.MagicMock()
+ mock_ti.dag_id = "test_dag_id"
+ mock_ti.task_id = TASK_ID
+ mock_ti.run_id = "manual__2024-01-01T00:00:00+00:00"
+ mock_ti.map_index = -1
+ mock_ti.try_number = 2
+ mock_ti.xcom_pull.return_value = None
+ task_uuid =
f"{mock_ti.dag_id}:{mock_ti.task_id}:{mock_ti.run_id}:{mock_ti.map_index}"
+
+ glue.hook.conn = mock.MagicMock()
+ glue.hook.conn.get_job_runs.return_value = {
+ "JobRuns": [
+ {
+ "Id": JOB_RUN_ID,
+ "Arguments": {GlueJobOperator.TASK_UUID_ARG: task_uuid},
+ "JobRunState": "RUNNING",
+ }
+ ]
+ }
+ erroring_store = mock.MagicMock()
+ erroring_store.get.side_effect = RuntimeError("store unavailable")
+
+ with pytest.raises(TaskDeferred) as defer:
+ glue.execute({"ti": mock_ti, "task_state_store": erroring_store})
+
+ assert defer.value.trigger.run_id == JOB_RUN_ID
+ mock_initialize_job.assert_not_called()
+ mock_initialize_job.assert_not_called()
+
@mock.patch.object(GlueJobHook, "print_job_logs")
@mock.patch.object(GlueJobHook, "get_job_state")
@mock.patch.object(GlueJobHook, "initialize_job")
@@ -172,6 +303,7 @@ class TestGlueJobOperator:
self, mock_load_file, mock_get_conn, mock_initialize_job,
mock_get_job_state, mock_print_job_logs
):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3_uri",
@@ -199,6 +331,7 @@ class TestGlueJobOperator:
self, mock_load_file, mock_get_conn, mock_initialize_job,
mock_get_job_state, mock_print_job_logs
):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3_uri",
@@ -224,6 +357,7 @@ class TestGlueJobOperator:
self, mock_load_file, mock_get_conn, mock_initialize_job,
mock_job_completion, mock_print_job_logs
):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://glue-examples/glue-scripts/sample_aws_glue_job.py",
@@ -253,6 +387,7 @@ class TestGlueJobOperator:
):
region = "us-west-2"
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://glue-examples/glue-scripts/sample_aws_glue_job.py",
@@ -328,6 +463,7 @@ class TestGlueJobOperator:
self, mock_load_file, mock_conn, mock_get_connection,
mock_initialize_job, mock_get_job_state
):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="folder/file",
@@ -360,6 +496,7 @@ class TestGlueJobOperator:
mock_get_job_state,
):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://my_bucket/folder/file",
@@ -391,6 +528,7 @@ class TestGlueJobOperator:
mock_get_job_state,
):
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
iam_role_name="role_arn",
@@ -436,56 +574,49 @@ class TestGlueJobOperator:
@mock.patch.object(GlueJobHook, "get_conn")
@mock.patch.object(GlueJobHook, "initialize_job")
- def test_check_previous_job_id_run_reuse_in_progress(self,
mock_initialize_job, mock_get_conn):
- """Test that when resume_glue_job_on_retry=True and previous job is in
progress, it is reused."""
- glue = GlueJobOperator(
- task_id=TASK_ID,
- job_name=JOB_NAME,
- script_location="s3://folder/file",
- aws_conn_id="aws_default",
- region_name="us-west-2",
- s3_bucket="some_bucket",
- iam_role_name="my_test_role",
- resume_glue_job_on_retry=True,
- wait_for_completion=False,
- )
-
- # Mock the context and task instance
- mock_ti = mock.MagicMock()
- mock_context = {"ti": mock_ti}
-
- # Simulate previous job_run_id in XCom
- previous_job_run_id = "previous_run_12345"
- mock_ti.xcom_pull.return_value = previous_job_run_id
-
- # Mock the Glue client to return RUNNING state for the previous job
- mock_glue_client = mock.MagicMock()
- glue.hook.conn = mock_glue_client
- mock_glue_client.get_job_run.return_value = {
- "JobRun": {
- "JobRunState": "RUNNING",
- }
- }
-
- # Execute the operator
- glue.execute(mock_context)
-
- # Verify that the previous job_run_id was reused
+ @pytest.mark.parametrize("state", ["RUNNING", "STARTING", "WAITING",
"STOPPING"])
+ def test_find_previous_job_run_reuses_from_xcom(self, mock_initialize_job,
mock_get_conn, state):
+ with
mock.patch("airflow.providers.amazon.aws.operators.glue.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(DEPRECATION_MESSAGE_PRE_3_3)}$"
+ ):
+ glue = GlueJobOperator(
+ task_id=TASK_ID,
+ job_name=JOB_NAME,
+ script_location="s3://folder/file",
+ aws_conn_id="aws_default",
+ region_name="us-west-2",
+ s3_bucket="some_bucket",
+ iam_role_name="my_test_role",
+ wait_for_completion=False,
+ resume_glue_job_on_retry=True,
+ )
+
+ mock_ti = mock.MagicMock()
+ mock_ti.try_number = 2 # the lookup only runs on a retry
+ previous_job_run_id = "previous_run_12345"
+ mock_ti.xcom_pull.return_value = previous_job_run_id
+ mock_context = {"ti": mock_ti}
+
+ mock_glue_client = mock.MagicMock()
+ glue.hook.conn = mock_glue_client
+ mock_glue_client.get_job_run.return_value = {"JobRun":
{"JobRunState": state}}
+
+ job_run_id = glue.execute(mock_context)
+
+ assert job_run_id == previous_job_run_id
assert glue._job_run_id == previous_job_run_id
- # Verify that initialize_job was NOT called
mock_initialize_job.assert_not_called()
- # Verify that XCom push was not called for glue_job_run_id (since we
reused the previous one)
- # Note: xcom_push may be called for other purposes like
glue_job_run_details
- xcom_calls = [
- call for call in mock_ti.xcom_push.call_args_list if
call[1].get("key") == "glue_job_run_id"
- ]
- assert len(xcom_calls) == 0, "Should not push new glue_job_run_id when
reusing previous one"
+ mock_glue_client.get_job_runs.assert_not_called()
@mock.patch.object(GlueJobHook, "get_conn")
@mock.patch.object(GlueJobHook, "initialize_job")
- def test_check_previous_job_id_run_new_on_finished(self,
mock_initialize_job, mock_get_conn):
- """Test that when previous job is finished, a new job is started and
pushed to XCom."""
+ def
test_find_previous_job_run_does_not_fall_back_to_scan_on_xcom_state_mismatch(
+ self, mock_initialize_job, mock_get_conn
+ ):
+ """A stale XCom state doesn't fall back to the task-UUID scan -- it's
elif, not a chain."""
glue = GlueJobOperator(
+ durable=True,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -493,100 +624,77 @@ class TestGlueJobOperator:
region_name="us-west-2",
s3_bucket="some_bucket",
iam_role_name="my_test_role",
- resume_glue_job_on_retry=True,
wait_for_completion=False,
)
- # Mock the context and task instance
mock_ti = mock.MagicMock()
+ mock_ti.try_number = 2
+ mock_ti.xcom_pull.return_value = "previous_run_12345"
mock_context = {"ti": mock_ti}
- # Simulate previous job_run_id in XCom
- previous_job_run_id = "previous_run_12345"
- mock_ti.xcom_pull.return_value = previous_job_run_id
-
- # Mock the Glue client to return SUCCEEDED state for the previous job
mock_glue_client = mock.MagicMock()
glue.hook.conn = mock_glue_client
- mock_glue_client.get_job_run.return_value = {
- "JobRun": {
- "JobRunState": "SUCCEEDED",
- }
- }
+ mock_glue_client.get_job_run.return_value = {"JobRun": {"JobRunState":
"SUCCEEDED"}}
- # Mock initialize_job to return a new job run ID
new_job_run_id = "new_run_67890"
- mock_initialize_job.return_value = {
- "JobRunState": "RUNNING",
- "JobRunId": new_job_run_id,
- }
+ mock_initialize_job.return_value = {"JobRunId": new_job_run_id}
- # Execute the operator
- glue.execute(mock_context)
+ job_run_id = glue.execute(mock_context)
- # Verify that a new job_run_id was created
- assert glue._job_run_id == new_job_run_id
- # Verify that initialize_job was called
+ assert job_run_id == new_job_run_id
mock_initialize_job.assert_called_once()
- # Verify that the new job_run_id was pushed to XCom
- xcom_calls = [
- call for call in mock_ti.xcom_push.call_args_list if
call[1].get("key") == "glue_job_run_id"
- ]
- assert len(xcom_calls) == 1, "Should push new glue_job_run_id"
- assert xcom_calls[0][1]["value"] == new_job_run_id
+ mock_glue_client.get_job_runs.assert_not_called()
+ mock_ti.xcom_push.assert_any_call(key="glue_job_run_id",
value=new_job_run_id)
@mock.patch.object(GlueJobHook, "get_conn")
@mock.patch.object(GlueJobHook, "initialize_job")
- def test_resume_glue_job_on_retry_find_job_run_by_task_uuid(self,
mock_initialize_job, mock_get_conn):
- """Test that when XCom is missing, job run is found by task UUID."""
- glue = GlueJobOperator(
- task_id=TASK_ID,
- job_name=JOB_NAME,
- script_location="s3://folder/file",
- aws_conn_id="aws_default",
- region_name="us-west-2",
- s3_bucket="some_bucket",
- iam_role_name="my_test_role",
- resume_glue_job_on_retry=True,
- wait_for_completion=False,
- )
-
- mock_ti = mock.MagicMock()
- mock_ti.dag_id = "test_dag_id"
- mock_ti.task_id = TASK_ID
- mock_ti.run_id = "manual__2024-01-01T00:00:00+00:00"
- mock_ti.map_index = -1
- mock_ti.xcom_pull.return_value = None
- mock_context = {"ti": mock_ti}
-
- task_uuid =
f"{mock_ti.dag_id}:{mock_ti.task_id}:{mock_ti.run_id}:{mock_ti.map_index}"
-
- mock_glue_client = mock.MagicMock()
- glue.hook.conn = mock_glue_client
- mock_glue_client.get_job_runs.return_value = {
- "JobRuns": [
- {
- "Id": "existing_run_123",
- "Arguments": {GlueJobOperator.TASK_UUID_ARG: task_uuid},
- "JobRunState": "STARTING",
- }
- ]
- }
- mock_glue_client.get_job_run.return_value = {
- "JobRun": {
- "JobRunState": "RUNNING",
+ @pytest.mark.parametrize("state", ["RUNNING", "STARTING", "WAITING",
"STOPPING"])
+ def test_find_job_run_by_task_uuid_reconnects(self, mock_initialize_job,
mock_get_conn, state):
+ with
mock.patch("airflow.providers.amazon.aws.operators.glue.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(DEPRECATION_MESSAGE_PRE_3_3)}$"
+ ):
+ glue = GlueJobOperator(
+ task_id=TASK_ID,
+ job_name=JOB_NAME,
+ script_location="s3://folder/file",
+ aws_conn_id="aws_default",
+ region_name="us-west-2",
+ s3_bucket="some_bucket",
+ iam_role_name="my_test_role",
+ wait_for_completion=False,
+ resume_glue_job_on_retry=True,
+ )
+
+ mock_ti = mock.MagicMock()
+ mock_ti.dag_id = "test_dag_id"
+ mock_ti.task_id = TASK_ID
+ mock_ti.run_id = "manual__2024-01-01T00:00:00+00:00"
+ mock_ti.map_index = -1
+ mock_ti.try_number = 2
+ mock_ti.xcom_pull.return_value = None
+ mock_context = {"ti": mock_ti}
+
+ task_uuid =
f"{mock_ti.dag_id}:{mock_ti.task_id}:{mock_ti.run_id}:{mock_ti.map_index}"
+
+ mock_glue_client = mock.MagicMock()
+ glue.hook.conn = mock_glue_client
+ mock_glue_client.get_job_runs.return_value = {
+ "JobRuns": [
+ {
+ "Id": "existing_run_123",
+ "Arguments": {GlueJobOperator.TASK_UUID_ARG:
task_uuid},
+ "JobRunState": state,
+ }
+ ]
}
- }
- glue.execute(mock_context)
+ job_run_id = glue.execute(mock_context)
+ assert job_run_id == "existing_run_123"
assert glue._job_run_id == "existing_run_123"
mock_initialize_job.assert_not_called()
- xcom_calls = [
- call for call in mock_ti.xcom_push.call_args_list if
call[1].get("key") == "glue_job_run_id"
- ]
- assert len(xcom_calls) == 1, "Should push existing glue_job_run_id
when found by task UUID"
- assert xcom_calls[0][1]["value"] == "existing_run_123"
+ mock_ti.xcom_push.assert_any_call(key="glue_job_run_id",
value="existing_run_123")
class TestGlueJobOperatorOpenLineageInjection:
@@ -607,6 +715,7 @@ class TestGlueJobOperatorOpenLineageInjection:
mock_initialize_job.return_value = {"JobRunState": "RUNNING",
"JobRunId": JOB_RUN_ID}
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -633,6 +742,7 @@ class TestGlueJobOperatorOpenLineageInjection:
mock_initialize_job.return_value = {"JobRunState": "RUNNING",
"JobRunId": JOB_RUN_ID}
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -659,6 +769,7 @@ class TestGlueJobOperatorOpenLineageInjection:
mock_initialize_job.return_value = {"JobRunState": "RUNNING",
"JobRunId": JOB_RUN_ID}
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -696,6 +807,7 @@ class TestGlueJobOperatorOpenLineageInjection:
mock_initialize_job.return_value = {"JobRunState": "RUNNING",
"JobRunId": JOB_RUN_ID}
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -727,6 +839,7 @@ class TestGlueJobOperatorOpenLineageInjection:
mock_initialize_job.return_value = {"JobRunState": "RUNNING",
"JobRunId": JOB_RUN_ID}
glue = GlueJobOperator(
+ durable=False,
task_id=TASK_ID,
job_name=JOB_NAME,
script_location="s3://folder/file",
@@ -746,7 +859,7 @@ class TestGlueJobOperatorOpenLineageInjection:
@mock.patch(
"airflow.providers.amazon.aws.operators.glue.inject_parent_job_information_into_glue_arguments"
)
- def test_inject_parent_job_info_with_resume_on_retry(
+ def test_inject_parent_job_info_with_durable_scan(
self, mock_inject_parent, mock_initialize_job, mock_get_conn
):
"""OL injection is applied before task UUID is added; both end up in
the args passed to initialize_job."""
@@ -756,23 +869,28 @@ class TestGlueJobOperatorOpenLineageInjection:
}
mock_initialize_job.return_value = {"JobRunState": "RUNNING",
"JobRunId": JOB_RUN_ID}
- glue = GlueJobOperator(
- task_id=TASK_ID,
- job_name=JOB_NAME,
- script_location="s3://folder/file",
- iam_role_name="my_test_role",
- wait_for_completion=False,
- openlineage_inject_parent_job_info=True,
- resume_glue_job_on_retry=True,
- )
-
- mock_ti = mock.MagicMock()
- mock_ti.xcom_pull.return_value = None # no previous run
- context = {"ti": mock_ti}
- mock_glue_client = mock.MagicMock()
- glue.hook.conn = mock_glue_client
- mock_glue_client.get_job_runs.return_value = {"JobRuns": []}
- glue.execute(context)
+ with
mock.patch("airflow.providers.amazon.aws.operators.glue.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(DEPRECATION_MESSAGE_PRE_3_3)}$"
+ ):
+ glue = GlueJobOperator(
+ task_id=TASK_ID,
+ job_name=JOB_NAME,
+ script_location="s3://folder/file",
+ iam_role_name="my_test_role",
+ wait_for_completion=False,
+ openlineage_inject_parent_job_info=True,
+ resume_glue_job_on_retry=True,
+ )
+
+ mock_ti = mock.MagicMock()
+ mock_ti.try_number = 2
+ mock_ti.xcom_pull.return_value = None
+ context = {"ti": mock_ti}
+ mock_glue_client = mock.MagicMock()
+ glue.hook.conn = mock_glue_client
+ mock_glue_client.get_job_runs.return_value = {"JobRuns": []}
+ glue.execute(context)
mock_inject_parent.assert_called_once()
# The injected OL arg and the task UUID arg should both be present
@@ -781,6 +899,404 @@ class TestGlueJobOperatorOpenLineageInjection:
assert GlueJobOperator.TASK_UUID_ARG in call_args
+class TestWarnAndDisableDurableAirflowPre3_3:
+ def test_no_warning_when_unset(self):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ result = _warn_and_disable_durable_pre_3_3(_DURABLE_UNSET)
+ assert result is False
+ assert caught == []
+
+ @pytest.mark.parametrize("value", [True, False])
+ def test_warns_and_disables_when_explicitly_set(self, value):
+ with pytest.warns(UserWarning, match="durable.*no effect"):
+ result = _warn_and_disable_durable_pre_3_3(value)
+ assert result is False
+
+
+class TestGlueJobOperatorDeprecation:
+ @pytest.mark.parametrize("resume_value", [True, False])
+ def test_warns_and_maps_to_durable_old_flag(self, resume_value):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(EXPECTED_DEPRECATION_MESSAGE)}$"
+ ):
+ glue = GlueJobOperator(task_id=TASK_ID, job_name=JOB_NAME,
resume_glue_job_on_retry=resume_value)
+ assert glue.durable is resume_value
+
+ def test_warns_on_every_supported_airflow_version(self):
+ with
mock.patch("airflow.providers.amazon.aws.operators.glue.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(DEPRECATION_MESSAGE_PRE_3_3)}$"
+ ):
+ GlueJobOperator(task_id=TASK_ID, job_name=JOB_NAME,
resume_glue_job_on_retry=True)
+
+ def test_default_args_durable_reaches_operator(self):
+ with DAG(
+ dag_id="test_glue_durable_default_args",
+ schedule=None,
+ start_date=datetime(2024, 1, 1),
+ default_args={"durable": False},
+ ):
+ glue = GlueJobOperator(task_id=TASK_ID, job_name=JOB_NAME)
+ assert glue.durable is False
+
+ def test_legacy_flag_wins_over_conflicting_durable_below_3_3(self):
+ with
mock.patch("airflow.providers.amazon.aws.operators.glue.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(DEPRECATION_MESSAGE_PRE_3_3)}$"
+ ):
+ glue = GlueJobOperator(
+ task_id=TASK_ID,
+ job_name=JOB_NAME,
+ durable=False,
+ resume_glue_job_on_retry=True,
+ )
+ # assert that glube.durable is True even though durable is set to
False, because resume_glue_job_on_retry takes precedence in this case.
+ assert glue.durable is True
+
+
+class FakeTaskStateStore:
+ """In-memory task state store for tests."""
+
+ def __init__(self, stored: dict[str, str] | None = None):
+ self._store: dict[str, str] = dict(stored or {})
+
+ def get(self, key: str) -> str | None:
+ return self._store.get(key)
+
+ def set(self, key: str, value: str) -> None:
+ self._store[key] = value
+
+
[email protected](
+ not AIRFLOW_V_3_3_PLUS,
+ reason="ResumableJobMixin reconnect requires task_state_store, available
in Airflow 3.3+",
+)
+class TestGlueJobOperatorDurableExecution:
+ def _build(self, **kwargs):
+ return GlueJobOperator(task_id=TASK_ID, job_name=JOB_NAME, **kwargs)
+
+ def _stub_empty_scan(self, glue):
+ # submit_job scans for a task UUID tagged run whenever durable is set,
regardless of why it
+ # was called. Stub it to return no matches so tests don't depend on
that fallback mechanism.
+ glue.hook.conn = mock.MagicMock()
+ glue.hook.conn.get_job_runs.return_value = {"JobRuns": []}
+
+ def _context(self, store=None, try_number=2):
+ ti = mock.MagicMock()
+ ti.try_number = try_number
+ ti.xcom_pull.return_value = None
+ ctx = {"ti": ti}
+ if store is not None:
+ ctx["task_state_store"] = store
+ return ctx
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_fresh_submit_persists_before_polling(
+ self, mock_get_conn, mock_initialize_job, mock_job_completion
+ ):
+ glue = self._build(durable=True)
+ self._stub_empty_scan(glue)
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ store = FakeTaskStateStore()
+ persisted_before_poll = []
+ mock_job_completion.side_effect = lambda *a, **k: (
+ persisted_before_poll.append(store.get("glue_job_run_id")) or
{"JobRunState": "SUCCEEDED"}
+ )
+
+ job_run_id = glue.execute(self._context(store))
+
+ assert job_run_id == "jr_new"
+ assert store.get("glue_job_run_id") == "jr_new"
+ assert persisted_before_poll == ["jr_new"]
+ mock_initialize_job.assert_called_once()
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_reconnect_when_stored_run_is_running(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion
+ ):
+ glue = self._build(durable=True)
+ mock_get_job_state.return_value = "RUNNING"
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+ context = self._context(store)
+
+ job_run_id = glue.execute(context)
+
+ assert job_run_id == "jr_old"
+ mock_initialize_job.assert_not_called()
+ mock_job_completion.assert_called_once_with(JOB_NAME, "jr_old", False,
0)
+ context["ti"].xcom_push.assert_any_call(key="glue_job_run_id",
value="jr_old")
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_legacy_flag_still_warns_and_reconnects_like_durable(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion
+ ):
+ """resume_glue_job_on_retry warns on 3.3+, but durable execution
behaves identically to durable=True."""
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(DEPRECATION_MESSAGE_3_3_PLUS)}$"
+ ):
+ glue = self._build(resume_glue_job_on_retry=True)
+ mock_get_job_state.return_value = "RUNNING"
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ job_run_id = glue.execute(self._context(store))
+
+ assert job_run_id == "jr_old"
+ mock_initialize_job.assert_not_called()
+
+ @pytest.mark.parametrize("status", ["STARTING", "RUNNING", "WAITING",
"STOPPING"])
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_reconnect_from_every_active_state(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion, status
+ ):
+ glue = self._build(durable=True)
+ mock_get_job_state.return_value = status
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ glue.execute(self._context(store))
+
+ mock_initialize_job.assert_not_called()
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_reconnect_to_stopping_run_that_settles_stopped_raises(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion
+ ):
+ glue = self._build(durable=True)
+ mock_get_job_state.return_value = "STOPPING"
+ mock_job_completion.return_value = {"JobRunState": "STOPPED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ with pytest.raises(GlueJobRunStoppedError, match="jr_old"):
+ glue.execute(self._context(store))
+
+ mock_initialize_job.assert_not_called()
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_already_succeeded_returns_without_resubmit(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion
+ ):
+ glue = self._build(durable=True)
+ mock_get_job_state.return_value = "SUCCEEDED"
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+ context = self._context(store)
+
+ job_run_id = glue.execute(context)
+
+ assert job_run_id == "jr_old"
+ mock_initialize_job.assert_not_called()
+ mock_job_completion.assert_not_called()
+ context["ti"].xcom_push.assert_any_call(key="glue_job_run_id",
value="jr_old")
+
+ @pytest.mark.parametrize("status", ["FAILED", "TIMEOUT", "STOPPED"])
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_terminal_failure_resubmits_fresh(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion, status
+ ):
+ glue = self._build(durable=True)
+ self._stub_empty_scan(glue)
+ mock_get_job_state.return_value = status
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ job_run_id = glue.execute(self._context(store))
+
+ assert job_run_id == "jr_new"
+ assert store.get("glue_job_run_id") == "jr_new"
+ mock_initialize_job.assert_called_once()
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_terminal_resubmit_skips_the_scan_when_store_already_had_an_id(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion
+ ):
+ glue = self._build(durable=True)
+ glue.hook.conn = mock.MagicMock()
+ mock_get_job_state.return_value = "FAILED"
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ job_run_id = glue.execute(self._context(store))
+
+ assert job_run_id == "jr_new"
+ glue.hook.conn.get_job_runs.assert_not_called()
+ mock_initialize_job.assert_called_once()
+
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_terminal_resubmit_clears_stale_id_if_initialize_job_fails(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state
+ ):
+ glue = self._build(durable=True, stop_job_run_on_kill=True)
+ glue.hook.conn = mock.MagicMock()
+ mock_get_job_state.return_value = "FAILED"
+ mock_initialize_job.side_effect = ClientError(
+ {"Error": {"Code": "Throttling", "Message": "slow down"}},
"StartJobRun"
+ )
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ with pytest.raises(ClientError):
+ glue.execute(self._context(store))
+
+ assert glue._job_run_id is None
+ glue.on_kill()
+ glue.hook.conn.batch_stop_job_run.assert_not_called()
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "get_job_state")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_not_found_resubmits_fresh(
+ self, mock_get_conn, mock_initialize_job, mock_get_job_state,
mock_job_completion
+ ):
+ glue = self._build(durable=True)
+ self._stub_empty_scan(glue)
+ mock_get_job_state.side_effect = ClientError(
+ {"Error": {"Code": "EntityNotFoundException", "Message": "gone"}},
"GetJobRun"
+ )
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ job_run_id = glue.execute(self._context(store))
+
+ assert job_run_id == "jr_new"
+ mock_initialize_job.assert_called_once()
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_durable_false_never_touches_store(self, mock_get_conn,
mock_initialize_job, mock_job_completion):
+ glue = self._build(durable=False)
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore({"glue_job_run_id": "jr_old"})
+
+ job_run_id = glue.execute(self._context(store))
+
+ assert job_run_id == "jr_new"
+ assert store.get("glue_job_run_id") == "jr_old", "store must be left
untouched"
+ mock_initialize_job.assert_called_once()
+
+ @mock.patch.object(GlueJobHook, "conn", new_callable=mock.PropertyMock)
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_first_attempt_skips_the_retry_lookup_entirely(
+ self, mock_get_conn, mock_initialize_job, mock_job_completion,
mock_conn
+ ):
+ glue = self._build(durable=True)
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore()
+
+ job_run_id = glue.execute(self._context(store, try_number=1))
+
+ assert job_run_id == "jr_new"
+ mock_initialize_job.assert_called_once()
+ # No tag needed on 3.3+ synchronous runs: task_state_store is the sole
reconnect mechanism.
+ assert GlueJobOperator.TASK_UUID_ARG not in
mock_initialize_job.call_args[0][0]
+ mock_conn.return_value.get_job_run.assert_not_called()
+ mock_conn.return_value.get_job_runs.assert_not_called()
+
+ @mock.patch.object(GlueJobHook, "conn", new_callable=mock.PropertyMock)
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_retry_on_3_3_plus_sync_never_scans_or_tags(
+ self, mock_get_conn, mock_initialize_job, mock_job_completion,
mock_conn
+ ):
+ glue = self._build(durable=True)
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ mock_job_completion.return_value = {"JobRunState": "SUCCEEDED"}
+ store = FakeTaskStateStore()
+
+ job_run_id = glue.execute(self._context(store, try_number=3))
+
+ assert job_run_id == "jr_new"
+ assert GlueJobOperator.TASK_UUID_ARG not in
mock_initialize_job.call_args[0][0]
+ mock_conn.return_value.get_job_run.assert_not_called()
+ mock_conn.return_value.get_job_runs.assert_not_called()
+
+ @mock.patch.object(GlueJobHook, "job_completion")
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_wait_for_completion_false_still_persists_immediately(
+ self, mock_get_conn, mock_initialize_job, mock_job_completion
+ ):
+ glue = self._build(durable=True, wait_for_completion=False)
+ self._stub_empty_scan(glue)
+ mock_initialize_job.return_value = {"JobRunId": "jr_new"}
+ store = FakeTaskStateStore()
+
+ job_run_id = glue.execute(self._context(store))
+
+ assert job_run_id == "jr_new"
+ assert store.get("glue_job_run_id") == "jr_new"
+ mock_job_completion.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("status", "expected_active"),
+ [
+ ("STARTING", True),
+ ("RUNNING", True),
+ ("WAITING", True),
+ ("STOPPING", True),
+ ("SUCCEEDED", False),
+ ("STOPPED", False),
+ ("FAILED", False),
+ ("TIMEOUT", False),
+ ("ERROR", False),
+ ("EXPIRED", False),
+ ("NOT_FOUND", False),
+ ("SOME_FUTURE_STATE", True),
+ ],
+ )
+ def test_is_job_active(self, status, expected_active):
+ glue = self._build()
+ assert glue.is_job_active(status) is expected_active
+
+ @pytest.mark.parametrize(
+ ("status", "expected_succeeded"),
+ [
+ ("SUCCEEDED", True),
+ ("STOPPED", False),
+ ("RUNNING", False),
+ ("FAILED", False),
+ ],
+ )
+ def test_is_job_succeeded(self, status, expected_succeeded):
+ glue = self._build()
+ assert glue.is_job_succeeded(status) is expected_succeeded
+
+
class TestGlueDataQualityOperator:
RULE_SET_NAME = "TestRuleSet"
RULE_SET = 'Rules=[ColumnLength "review_id" = 15]'
