kaxil commented on code in PR #71211:
URL: https://github.com/apache/airflow/pull/71211#discussion_r3764425760
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -384,6 +364,147 @@ def on_kill(self):
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
+ 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)
+ if self.durable:
+ 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.11-3.2; Airflow 3.3+ clears task XComs before
every 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 state in ("RUNNING", "STARTING"):
+ 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 existing_job_run_state in ("RUNNING", "STARTING"):
+ ti.xcom_push(key="glue_job_run_id",
value=existing_job_run_id)
+ 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. A stored id (even terminal) means the
caller already decided.
+ if self.durable and context["ti"].try_number > 1 and 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"])
+ # Downstream tasks read this key directly; it's also what feeds the
XCom tier above.
+ context["ti"].xcom_push(key="glue_job_run_id",
value=glue_job_run["JobRunId"])
Review Comment:
Follow-up on my round-1 comment. The push is back, but only on the branch
that submits. `execute_resumable` returns from the reconnect branch and from
the already-succeeded branch
(`task-sdk/src/airflow/sdk/bases/resumablejobmixin.py:196-199`) without ever
reaching `submit_job`, and every non-deferral attempt clears the task
instance's XComs first. So after a retry that reconnects, or a clear that
short-circuits on an already-succeeded run, `xcom_pull(task_ids="glue_job",
key="glue_job_run_id")` returns None.
The asymmetry is what makes it read as an oversight: `_set_job_run_id`
already carries the other per-attempt side effect
(`GlueJobRunDetailsLink.persist`) on all four paths, so the link survives and
only the run-id key was left behind.
Scope, stated honestly: `execute` still returns the run id, so
`glue_job.output` and keyless `xcom_pull` are unaffected, and there is no
in-repo consumer of the key. It is a de-facto contract rather than a documented
one, but it is the one the comment on the line above asserts ("Downstream tasks
read this key directly").
Ask: move the push into `_set_job_run_id` beside the `persist` call and drop
it here. It is already called on all four paths, already de-duplicated by the
`if self._job_run_id == job_run_id: return` guard, and same-key `xcom_push` is
delete-then-insert, so a stale-then-fresh sequence still ends on the correct
id, exactly as the link already does.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -384,6 +364,147 @@ def on_kill(self):
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
+ 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)
+ if self.durable:
+ 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.11-3.2; Airflow 3.3+ clears task XComs before
every 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 state in ("RUNNING", "STARTING"):
+ 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 existing_job_run_state in ("RUNNING", "STARTING"):
+ ti.xcom_push(key="glue_job_run_id",
value=existing_job_run_id)
+ 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. A stored id (even terminal) means the
caller already decided.
+ if self.durable and context["ti"].try_number > 1 and not
self._has_stored_external_id(context):
Review Comment:
On the deferrable path this read cannot help and can hurt. Nothing persists
an id there, so `_has_stored_external_id` can only return False and never
changes the decision. But `TaskStateStoreAccessor.get` raises
`AirflowRuntimeError` for every error other than `TASK_STORE_NOT_FOUND`
(`task-sdk/src/airflow/sdk/execution_time/context.py:585-587`), and
`task_state_store` is unconditionally present in the runtime context on 3.3+
(`task-sdk/src/airflow/sdk/execution_time/task_runner.py:334`).
So a store that errors makes a deferrable retry fail from inside
`submit_job` before the task-UUID scan can run, which removes the recovery
glue.rst:213-216 promises for that exact path. This arrived with the guard
added for my round-2 comment on line 454, so it is worth closing while that
change is fresh.
Ask: `and not self.deferrable` on the condition, or hoist the lookup out of
`submit_job` and have the deferrable branch pass the decision in.
Related design question, if you would rather fix the cause than the symptom:
should the deferrable branch persist the run id before `defer()`? The
deferrable path would then reconnect through the cheap store lookup instead of
a full `get_job_runs` walk, and this plus the `WAITING`/`STOPPING` gap above
would both stop existing there.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -384,6 +364,147 @@ def on_kill(self):
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
+ 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)
+ if self.durable:
+ 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.11-3.2; Airflow 3.3+ clears task XComs before
every 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 state in ("RUNNING", "STARTING"):
+ 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 existing_job_run_state in ("RUNNING", "STARTING"):
+ ti.xcom_push(key="glue_job_run_id",
value=existing_job_run_id)
+ 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. A stored id (even terminal) means the
caller already decided.
+ if self.durable and context["ti"].try_number > 1 and not
self._has_stored_external_id(context):
+ existing_job_run_id = self._find_previous_job_run(context,
script_args[self.TASK_UUID_ARG])
Review Comment:
Worth a second look now that this runs by default rather than under an
opt-in flag. `_find_job_run_id_by_task_uuid` is a `while True` over
`get_job_runs(MaxResults=50)` with no page cap and no age window, and the
no-match case (the prior attempt died before submitting anything, which is the
common retry shape) is the one that walks the job's entire run history.
Two consequences that are new because the flag flipped, not because the
function changed:
The scan needs `glue:GetJobRuns`. A task policy granting only `StartJobRun`
and `GetJobRun`, which was sufficient before, now hits AccessDenied on every
retry.
The failure is swallowed. Both `except Exception` blocks (lines 419 and 434)
log at warning and return None, after which the operator submits fresh,
producing `ConcurrentRunsExceededException` against the live run, which is the
failure this feature exists to prevent. There is no error-level signal to tell
an operator why.
Not asking for fail-closed: those blocks are verbatim from the merge base,
and turning a transient throttle into a task failure would be worse.
Ask: bound the walk (a page cap, or stop once `StartedOn` predates the DAG
run), narrow the `except Exception` to `ClientError`, log the fall-through at
error level with the reason, and add `glue:GetJobRuns` to the IAM note in
glue.rst.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -158,6 +203,16 @@ def __init__(
),
**kwargs,
):
+ if resume_glue_job_on_retry is not None:
+ # Kept as a real named parameter (not **kwargs) so `default_args`
still applies correctly.
+ if AIRFLOW_V_3_3_PLUS:
Review Comment:
Provider deprecations are scoped by provider version, not Airflow version.
The same amazon wheel serves 2.11 through 3.4, and the removal will land in a
future amazon release for all of them, so gating the warning on
`AIRFLOW_V_3_3_PLUS` hides it from part of the install base for the whole
deprecation window.
The usual justification for such a gate does not apply here: below 3.3 the
compat stub still sets `self.durable`, and the operator honours it at lines 398
and 447, so `durable` is a functional drop-in for what
`resume_glue_job_on_retry` did. The shipped precedent warns unconditionally
(`providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py:405-411`,
no version gate).
A team on Airflow 3.1 passing `resume_glue_job_on_retry=True` therefore sees
nothing until the parameter is removed, and then gets a `TypeError`.
Ask: drop the gate, invert `test_silent_below_3_3` to assert the warning
fires on every supported version, and remove the "on Airflow 3.3 and newer"
qualifier at glue.rst:220.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -384,6 +364,147 @@ def on_kill(self):
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
+ 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)
+ if self.durable:
+ 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.11-3.2; Airflow 3.3+ clears task XComs before
every 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 state in ("RUNNING", "STARTING"):
+ 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 existing_job_run_state in ("RUNNING", "STARTING"):
+ ti.xcom_push(key="glue_job_run_id",
value=existing_job_run_id)
+ 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. A stored id (even terminal) means the
caller already decided.
+ if self.durable and context["ti"].try_number > 1 and 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"])
+ # Downstream tasks read this key directly; it's also what feeds the
XCom tier above.
+ context["ti"].xcom_push(key="glue_job_run_id",
value=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:
Review Comment:
Completing the thread from my round-2 comment on glue.rst:158 rather than
reversing it. This raise is the fix I asked for, and it is the right end state:
the deferrable non-verbose path's `job_complete` waiter already treats
`STOPPED` as a failure, so this makes the synchronous path agree with it.
The issue is scoping. The raise lives in `poll_until_complete`, which
`execute_resumable` also calls on its non-durable branch, and it fires on first
attempts too, so it is not gated on `durable` in any way. Concretely: someone
cancels a Glue run in the console, and on main `_handle_state` buckets
`STOPPED` in `finished_states` so the task logs `status: STOPPED` and goes
green. At this HEAD it goes red and burns a retry that resubmits the job they
just cancelled.
The docs do not carry that. The stopped-run paragraph at glue.rst:165-173
sits entirely under "Durable execution", and glue.rst:199 offers
`durable=False` as the opt-out, so a reader reasonably concludes
`durable=False` restores the old behaviour. It does not.
Ask: move that paragraph out of the durable section and reword it to cover
first attempts and `durable=False`, add a sentence saying explicitly that
`durable=False` does not restore `STOPPED`-as-success, and drop the "durable"
framing from the code comment on the line above.
##########
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 reconnected Glue job run finishes in a state that is not
a real success."""
Review Comment:
"reconnected" is narrower than the behaviour. `poll_until_complete` raises
this on the fresh-submit path as well, and on first attempts, so the docstring
understates the blast radius for `durable=False` users.
Ask: drop the word, for example "Raised when a Glue job run finishes in a
state that is not a real success."
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -384,6 +364,147 @@ def on_kill(self):
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
+ 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)
+ if self.durable:
+ 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.11-3.2; Airflow 3.3+ clears task XComs before
every 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 state in ("RUNNING", "STARTING"):
Review Comment:
This keeps the merge-base literal `if state in ("RUNNING", "STARTING")` here
and again at line 431, but the new `is_job_active` at line 479 is a deny-list
that counts `WAITING` and `STOPPING` as active, and the docs added at
glue.rst:158 promise the operator reconnects when a run is "waiting for
capacity, or being stopped".
This tier matters more than it looks: it is the only recovery mechanism on
the deferrable path (`execute` calls `submit_job` directly and nothing persists
an id there, so `_has_stored_external_id` is always False), the only one below
Airflow 3.3, and the one used on 3.3+ when the worker died before the store
write. `WAITING` is exactly "queued behind the concurrent-run limit", which is
the situation glue.rst:151-154 builds the whole feature around.
To be fair about severity, this is not a regression: on main the default was
`resume_glue_job_on_retry=False`, so attempt 2 resubmitted anyway. The narrower
claim is that the implementation does not do what `is_job_active` and its own
new docs say.
Ask: `if self.is_job_active(state):` here and `if
self.is_job_active(existing_job_run_state):` at 431, plus the history-scan
tests parametrized over `WAITING` and `STOPPING` (only the task-state-store
tests cover those today).
Different code path from my round-2 comment on line 487, so this is not
re-litigating it: that fix landed in `poll_until_complete`, this is the
reattach decision.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -384,6 +364,147 @@ def on_kill(self):
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
+ 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)
+ if self.durable:
+ 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
Review Comment:
Follow-up on my round-2 comment on line 470. The version boundary here is
off by three releases, and I should say plainly that my own round-2 wording
pointed you at the wrong one.
Pre-attempt XCom clearing shipped in 3.0.0, not 3.3. At tag 3.0.0 the
execution API builds `xcom_keys_to_clear` for any attempt with no
`next_method`, and the SDK deletes every one of those keys. Present and
unconditional at 3.0.0, 3.1.0 and 3.2.0.
Consequences: the XCom tier is dead on every Airflow 3 release, the `if
previous_job_run_id:` branch below is dead code there, and glue.rst:180-184
tells 3.0-3.2 users they get a cheap XCom lookup before the scan when in fact
every durable retry pays the full paginated history walk. The comment at line
463 ("it's also what feeds the XCom tier above") is false for the same reason.
Ask: reword to "Airflow 2.x only; every Airflow 3 release clears task XComs
before each non-deferral attempt", and fix glue.rst:182 to match. Since 2.11 is
the only beneficiary, dropping the XCom tier is worth considering. Note also
that `test_find_previous_job_run_reuses_from_xcom` and
`test_find_previous_job_run_does_not_fall_back_to_scan_on_xcom_state_mismatch`
pin a path unreachable on any Airflow 3 release, because they stub `xcom_pull`
to return a value the runtime never will.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -158,6 +203,16 @@ def __init__(
),
**kwargs,
):
+ if resume_glue_job_on_retry is not None:
+ # Kept as a real named parameter (not **kwargs) so `default_args`
still applies correctly.
Review Comment:
This comment is about `resume_glue_job_on_retry`, and it is accurate for
that parameter, but `durable` itself is not a named parameter on
`GlueJobOperator.__init__`. It reaches the mixin through `**kwargs` via
`kwargs.setdefault`. `_apply_defaults` only fills parameters present in the
decorated function's own signature
(`task-sdk/src/airflow/sdk/bases/operator.py:527-529`), and the pre-3.3
fallback stub's `__init__` carries no `_apply_defaults` at all.
Net effect on Airflow 2.11 through 3.2, which is inside this provider's
declared floor: `default_args={"durable": False}` leaves `durable` True with no
error, while `default_args={"resume_glue_job_on_retry": False}` works. On the
versions where the deprecation warning is deliberately suppressed, the
deprecated parameter is the only one that works at DAG level. That is not
cosmetic there, since `durable` still gates the task-UUID injection and the
retry scan below 3.3.
This matters mainly because the PR flips the default to True: the documented
way to opt out of a new default does not work on those versions. The skipif
reason on `test_default_args_durable_reaches_operator` records the mechanism,
but nothing user-facing does.
Ask: `durable: bool | None = None` on `GlueJobOperator.__init__`, forwarded
only when set, which preserves the mixin's own default. Fair caveat: the
sibling operators route `durable` through `**kwargs` the same way, so if you
would rather treat it as a cross-provider issue, then stating the limitation
beside the `durable=False` example in glue.rst is the minimum.
##########
providers/amazon/docs/operators/glue.rst:
##########
@@ -139,6 +139,86 @@ 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.
+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
+
+A stopped run is treated as a failure, not a success. 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. If the
stored state is already
+``STOPPED``, the operator submits fresh. If a reconnect finds the run still
stopping and it
+settles into ``STOPPED`` while polling, the operator raises instead of
returning a result -- the
+task fails and a normal retry resubmits. Either way, a self-inflicted stop
never gets silently
+reported as a false success.
+
+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. On earlier
+Airflow versions, or if a prior run was never recorded to the task state
store, ``durable=True``
+still recovers a prior run 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.
+
+Like the persisted state itself, the stored run id isn't deleted
automatically, that only happens
+when someone runs ``airflow state-store clean``. 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
resubmits by
Review Comment:
This sentence is now backwards. With `durable=True` as the default, a retry
of a deferrable task scans for the tagged run first; resubmitting is what you
get when you opt out with `durable=False`. As written, "resubmits by default"
asserts the opposite of what line 447 of the operator does.
Ask: "would otherwise resubmit". (Separate sentence from the deferrable
point in my round-2 comment, so this is a distinct fix.)
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -384,6 +364,147 @@ def on_kill(self):
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
+ 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)
+ if self.durable:
Review Comment:
With `durable` defaulting to True, this injects `--airflow_task_uuid` into
the Arguments of every Glue job run. `initialize_job` passes the dict straight
to `start_job_run`
(`providers/amazon/src/airflow/providers/amazon/aws/hooks/glue.py:271`), and
Glue surfaces Arguments to the script as command-line arguments. At the merge
base this happened only under `resume_glue_job_on_retry=True`, which defaulted
to False, so a provider upgrade alone now changes the argv of unchanged
production scripts.
Hedging this deliberately, because it is not clear-cut:
`getResolvedOptions`, the documented Glue idiom, uses `parse_known_args` and
tolerates the extra argument, so scripts following the docs are unaffected. A
script doing its own strict parsing with
`argparse.ArgumentParser().parse_args()` exits 2 on an unrecognised argument.
Common, not universal.
Ask: either call this out in the docs as an upgrade note, or inject only
where the tag is actually the recovery mechanism (`self.deferrable or not
AIRFLOW_V_3_3_PLUS`), since on 3.3+ synchronous runs the task state store is
the primary tier and the tag only covers the crash-before-persist window the
mixin's docstring already declares unclosable.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py:
##########
@@ -35,17 +38,52 @@
)
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
+# ResumableJobMixin only exists on Airflow 3; this provider still targets
>=2.11. Drop this
+# fallback once the provider's minimum Airflow version is >=3.0.
Review Comment:
Threshold looks wrong. The mixin reads `context["task_state_store"]`, which
only exists on Airflow 3.3+, and the stub's own docstring below, the operator
docstring at line 186, and the test class gate all say 3.3. A maintainer who
bumps the floor to 3.0 and follows this comment would delete the fallback and
break every Glue DAG on 3.0 through 3.2 with a parse-time ImportError.
Ask: `>=3.3`. Worth noting none of the four shipped sibling stubs
(redshift_data, databricks, snowflake, bigquery) carries this comment at all.
##########
providers/amazon/docs/operators/glue.rst:
##########
@@ -139,6 +139,86 @@ 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.
+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
+
+A stopped run is treated as a failure, not a success. 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. If the
stored state is already
+``STOPPED``, the operator submits fresh. If a reconnect finds the run still
stopping and it
+settles into ``STOPPED`` while polling, the operator raises instead of
returning a result -- the
+task fails and a normal retry resubmits. Either way, a self-inflicted stop
never gets silently
+reported as a false success.
+
+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. On earlier
+Airflow versions, or if a prior run was never recorded to the task state
store, ``durable=True``
+still recovers a prior run 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.
+
+Like the persisted state itself, the stored run id isn't deleted
automatically, that only happens
Review Comment:
`airflow db clean` also targets this table:
`_TableConfig(table_name="task_state_store", recency_column_name="expires_at",
...)` in `airflow-core/src/airflow/utils/db_cleanup.py`. So "that only happens
when someone runs `airflow state-store clean`" understates when the stored id
can disappear.
Worth fixing because this paragraph's own mitigation advice, avoiding a
cleanup schedule shorter than the longest `retry_delay`, only lands if the
reader knows `db clean` counts too.
##########
providers/amazon/tests/unit/amazon/aws/operators/test_glue.py:
##########
@@ -163,6 +171,62 @@ def test_execute_deferrable(self, _, mock_initialize_job):
assert defer.value.trigger.attempts == 75
assert defer.value.trigger.aws_conn_id == "aws_default"
+ @mock.patch.object(GlueJobHook, "initialize_job")
+ @mock.patch.object(GlueJobHook, "get_conn")
+ def test_deferrable_first_attempt_injects_task_uuid_but_skips_scan(
Review Comment:
The test asserts the `TASK_UUID_ARG` half of its name but nothing for
"skips_scan", so the part most likely to regress is unguarded.
`test_first_attempt_skips_the_retry_lookup_entirely` at line 1097 already does
this properly.
Ask: add `mock_get_conn.return_value.get_job_runs.assert_not_called()`.
Worth more than a naming nit here, because the scan has no page cap and an
unspecced conn mock returns a truthy `NextToken`, so if the `try_number > 1`
gate is ever dropped this test hangs rather than fails.
--
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]
