Deferred Glue operators generate a new glue job ID upon task failure even existing glue job ID still running

[wilsonhooi86]

Apache Airflow Provider(s)

amazon

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==9.0.0

Apache Airflow version

2.10.3

Operating System

Amazon Linux 2023

Deployment

Amazon (AWS) MWAA

Deployment details

MWAA in production and mwaa-local-runner for local testing. Bug exists on both.

What happened

When a Deferred GlueJobOperator tasks failed with internal errors , the glueJobOperator task is marked as failed. However the existing glue job_run_id still running.

When the GlueJobOperator task retry 2nd time, it will automatically create a new glue job_run_id while the existing glue job_run_id still running. This caused duplicate run glue jobs at AWS Glue side.

Example of internal errors:

[2025-12-05, 01:10:33 UTC] {baseoperator.py:1798} ERROR - Trigger failed:
Traceback (most recent call last):

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/jobs/triggerer_job_runner.py", line 558, in cleanup_finished_triggers
    result = details["task"].result()
             ^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/jobs/triggerer_job_runner.py", line 630, in run_trigger
    async for event in trigger.run():

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/triggers/glue.py", line 73, in run
    await hook.async_job_completion(self.job_name, self.run_id, self.verbose)

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/hooks/glue.py", line 315, in async_job_completion
    ret = self._handle_state(job_run_state, job_name, run_id, verbose, next_log_tokens)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/hooks/glue.py", line 334, in _handle_state
    self.print_job_logs(

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/hooks/glue.py", line 278, in print_job_logs
    continuation_tokens.output_stream_continuation = display_logs_from(
                                                     ^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/hooks/glue.py", line 245, in display_logs_from
    for response in paginator.paginate(

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/botocore/paginate.py", line 269, in __iter__
    response = self._make_request(current_kwargs)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/botocore/paginate.py", line 357, in _make_request
    return self._method(**current_kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/botocore/client.py", line 569, in _api_call
    return self._make_api_call(operation_name, kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/botocore/client.py", line 1023, in _make_api_call
    raise error_class(parsed_response, operation_name)

botocore.errorfactory.ThrottlingException: An error occurred (ThrottlingException) when calling the FilterLogEvents operation (reached max retries: 4): Rate exceeded

[2025-12-05, 01:10:33 UTC] {taskinstance.py:3311} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 767, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 733, in _execute_callable
    return ExecutionCallableRunner(
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/utils/operator_helpers.py", line 252, in run
    return self.func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 1799, in resume_execution
    raise TaskDeferralError(next_kwargs.get("error", "Unknown"))
airflow.exceptions.TaskDeferralError: Trigger failure

or

The above exception was the direct cause of the following exception:


Traceback (most recent call last):

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/aiobotocore/httpsession.py", line 222, in send
    response = await session.request(
               ^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/aiohttp/client.py", line 667, in _request
    raise ConnectionTimeoutError(

aiohttp.client_exceptions.ConnectionTimeoutError: Connection timeout to host http://169.254.170.2/v2/credentials/c9ab61ee-ea14-42a2-9a16-0b4aebdedad7


During handling of the above exception, another exception occurred:


Traceback (most recent call last):

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/aiobotocore/utils.py", line 722, in _get_response
    response = await session.send(request.prepare())
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/aiobotocore/httpsession.py", line 259, in send
    raise ConnectTimeoutError(endpoint_url=request.url, error=e)

botocore.exceptions.ConnectTimeoutError: Connect timeout on endpoint URL: "[http://169.254.170.2/v2/credentials/c9ab61ee-ea14-42a2-9a16-0b4aebdedad7"](http://169.254.170.2/v2/credentials/c9ab61ee-ea14-42a2-9a16-0b4aebdedad7%22)

What you think should happen instead

I would propose to include a logic for GlueJobOperator during next retry, the GlueJobOperator would check if there is a previous glue job_run_id , it would check the the glue STATE,

If the glue STATE = in progress / completed, GlueJobOperator should not create a new glue job_run_id
If the glue STATE = failed, should create a new glue job_run_id

How to reproduce

1. Create a GlueJobOperator task.
2. Set deferrable=True and verbose=False
3. Run the task and watch the Airflow logs while it is in a deferred state.

Sample of the GlueJobOperator:

        company = GlueJobOperator(
            task_id="company",
            job_name=var_glue_job_name,
            verbose=False,
            stop_job_run_on_kill=True,
            deferrable=True,
            pool=var_glue_pool,
            priority_weight=3,
            job_poll_interval=random.randint(30, 120)
        )

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[jroachgolf84]

Just my two cents; I think that could get a bit weird. Let's say a Task fails and a user sets retries to 2. My guess is that the user would expect that a new Glue job would be spun up with that retry. While it's sometimes the case that the status of an Airflow Task does not match the status of the job that it's orchestrating, I think shrouding that could be dangerous.

I know that this has come up quite a bit with KPO in the past.

[wilsonhooi86]

Hi @jroachgolf84 , thank you for your comments, to explain better:

Our current Scenario:

1. GlueJobOperator task 1 runs glue_job_name_1. In AWS Glue UI, we can see glue_job_name_1 running
2. Suddenly GlueJobOperator faced internal error as below. Failed the airflow task but however glue_job_name_1 still running at AWS Glue

botocore.errorfactory.ThrottlingException: An error occurred (ThrottlingException) when calling the FilterLogEvents operation (reached max retries: 4): Rate exceeded

This error is due to the AWS hard limit when we have too many concurrent glue jobs running.

3. GlueJobOperator task 1 retry 2nd time and runs glue_job_name_1. AWS Glue calls a new glue_job_name_1
4. In AWS Glue, there are 2 glue_job_name_1 running

This would cause double glue job running and may cause complication as we only need 1 glue job name running at a time. However GlueJobOperator does not know the 1st glue job was still running.

Proposed solution:
We are hoping if we can have option for the GlueJobOperator to check previous glue job ID state with following conditions:

Example: check_previous_job_id_run: True

During next retry, GlueJobOperator can retrieve the previous_glue_job_id from XCOM
If the glue STATE = in progress, GlueJobOperator should not create a new glue job_run_id
If the glue STATE = failed/completed, should create a new glue job_run_id

Or if you have other alternatives, we are happy to hear as well =)

[jroachgolf84]

Got it, that makes sense - thanks for the explanation! For your use-case, something like this makes sense. That being said, I'm not sure if it's a common pattern to "look back" into XComs for a previous value. I need to look into that a bit more.

If we do go ahead with something like this, I would actually name the parameter something different, maybe like resume_glue_job_on_retry? Only reason I say that is that check_previous_job_id_run makes me think that a Task would always do this for a new Task run (before it fails).