Repository navigation
Task management issue: EmrServerlessStartJobOperator task fails randomly for few tasks in 20-21s even though job is submitted and succeeds fine in emr serverless in background #67272
Description
Under which category would you file this issue?
Providers
Apache Airflow version
3.0.6
What happened and how to reproduce it?
We upgraded aws mwaa airflow from 2.7.2 to 3.0.6 and we noticed 1 random issue. While submitting jobs to emr serverless from our dags i.e. via EmrServerlessStartJobOperator, we see jobs are submitted fine to emr serverless and are finished in emr but task status is marked as failure in airflow dag's task. Out of 100 tasks, 98-99 proceed fine but we see random failures for 1 or 2 tasks. We saw a pattern, it fails in 20-21seconds. Its completely random, not for particular task.
Something is wrong with new version of airflow or might be some configuration is missing from our end
Requirements.txt for airflow of both versions
Airflow 3.0.6
--constraint "/usr/local/airflow/dags/constraints-3.11_spark_trino.txt"
apache-airflow-providers-apache-spark==5.3.2
apache-airflow-providers-amazon==9.12.0
apache-airflow-providers-ssh==4.1.3
types-paramiko==3.5.0.20250801
sshtunnel==0.4.0
requests==2.32.5
orjson==3.11.2
cachetools==5.5.2
Authlib==1.6.2
apache-airflow-providers-apache-livy==4.4.2
apache-airflow-providers-http==5.3.3
confluent-kafka==2.11.1
apache-airflow-providers-apache-kafka==1.10.2
fastavro==1.12.0
Airflow 2.7.2
--constraint "/usr/local/airflow/dags/constraints-3.7_spark_trino.txt"
apache-airflow-providers-apache-spark==3.0.0
apache-airflow-providers-amazon==6.0.0
apache-airflow-providers-ssh==3.2.0
types-paramiko==2.11.6
sshtunnel==0.4.0
requests==2.28.1
apache-airflow-providers-apache-livy==3.1.0
apache-airflow-providers-http==4.0.0
Following are the logs of the task which fails randomly
Reading remote log from Cloudwatch log_group: arn:aws:logs:xxxxx:log-group:airflow-abc-MwaaEnvironment-Task log_stream: dag_id=xxx/run_id=manual__2026-05-19T10_35_27.159729+00_00/task_id=mytaskid/attempt=1.log
An error occurred (ResourceNotFoundException) when calling the GetLogEvents operation: The specified log stream does not exist.
Ideally this error log should be printed for other tasks as well but I dont think its failing due to missing log stream in the cloud-watch. It even didnt print that job was submitted to EMR successfully as other tasks are doing.
I logged similar issue, airflow team fixed waiter error codes for throttling and asked to log a separate issue since its a task management issue.
Reference to the original issue: #67178
I still faced the same issue with the fix airflow team provided in #67178 . Additional to the same issue I mentioned above, there is another task where we see some task logs, sharing task logs where it worked vs where it failed for the same table after the fix was applied. For the failed one, job was submitted and succeeded fine in EMR.
Passed:
Reading remote log from Cloudwatch log_group: arn:aws:logs:us-west-2:xxx:log-group:airflow-abc-MwaaEnvironment-Task log_stream: dag_id=mynamespace_xxxxx/run_id=manual__2026-05-20T06_53_06.800846+00_00/task_id=KP.mynamespace_csv_ingest_mytable/attempt=1.log
[2026-05-20, 12:35:55] WARNING - /usr/local/airflow/.local/lib/python3.12/site-packages/flask_sqlalchemy/model.py:121: SAWarning: This declarative base already contains a class with the same class name and module name as iam.MWAASession, and will be replaced in the string-lookup table. super(BindMetaMixin, cls).__init__(name, bases, d): source="py.warnings"
[2026-05-20, 12:35:55] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2026-05-20, 12:35:55] INFO - Filling up the DagBag from /usr/local/airflow/dags/mynamespace_ns/csv_load_dags/xxxxx.py: source="airflow.models.dagbag.DagBag"
[2026-05-20, 12:35:55] WARNING - /usr/local/airflow/.local/lib/python3.12/site-packages/airflow/models/connection.py:471: DeprecationWarning: Using Connection.get_connection_from_secrets from `airflow.models` is deprecated.Please use `get` on Connection from sdk(`airflow.sdk.Connection`) instead warnings.warn(: source="py.warnings"
[2026-05-20, 12:35:56] INFO - Connection Retrieved 'aws_default': source="airflow.hooks.base"
[2026-05-20, 12:35:56] INFO - Starting job on Application: myappid: source="airflow.task.operators.edfx_emr_serverless_operator.EdfxEmrServerlessStartJobOperator"
[2026-05-20, 12:35:56] INFO - EMR serverless job started: 00g5ql0rdccnpg0n: source="airflow.task.operators.edfx_emr_serverless_operator.EdfxEmrServerlessStartJobOperator"
[2026-05-20, 12:35:56] INFO - Serverless Job status is: SUBMITTED - SUBMITTED: source="waiter_with_logging"
[2026-05-20, 12:36:56] INFO - Serverless Job status is: RUNNING: source="waiter_with_logging"
[2026-05-20, 12:37:56] INFO - Pushing xcom: ti="RuntimeTaskInstance(id=UUID('019cc78ed-941e-7f0f656105c6'), task_id='KP.mynamespace_csv_ingest_mytable', dag_id='mynamespace_xxxxx', run_id='manual__2026-05-20T06:53:06.800846+00:00', try_number=1, map_index=-1, hostname='ip-10-151-47-166.us-west-2.compute.internal', context_carrier={}, task=<Task(EdfxEmrServerlessStartJobOperator): KP.mynamespace_csv_ingest_mytable>, bundle_instance=LocalDagBundle(name=dags-folder), max_tries=0, start_date=datetime.datetime(2026, 5, 20, 7, 5, 55, 318443, tzinfo=datetime.timezone.utc), end_date=None, state=<TaskInstanceState.RUNNING: 'running'>, is_mapped=False, rendered_map_index=None, log_url='https://a5cca3ac-1398-448f-a42f-1e87b05867a4-vpce.c29.airflow.us-west-2.on.awsdags/mynamespace_xxxxx/runs/manual__2026-05-20T06%3A53%3A06.800846%2B00%3A00/tasks/KP.mynamespace_csv_ingest_mytable?try_number=1%27)%22: source="task"
[2026-05-20, 12:37:56] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_s3_logs": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_mytable": run_id="manual__2026-05-20T06:53:06.800846+00:00": map_index=-1: source="task"
[2026-05-20, 12:37:56] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_cloudwatch_logs": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_mytable": run_id="manual__2026-05-20T06:53:06.800846+00:00": map_index=-1: source="task"
[2026-05-20, 12:37:56] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_dashboard": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_mytable": run_id="manual__2026-05-20T06:53:06.800846+00:00": map_index=-1: source="task"
[2026-05-20, 12:37:56] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_logs": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_mytable": run_id="manual__2026-05-20T06:53:06.800846+00:00": map_index=-1: source="task"
Failed:
[2026-05-20, 16:01:25] INFO - Starting job on Application: myappid: source="airflow.task.operators.edfx_emr_serverless_operator.EdfxEmrServerlessStartJobOperator"
[2026-05-20, 16:01:25] INFO - EMR serverless job started: jobid: source="airflow.task.operators.edfx_emr_serverless_operator.EdfxEmrServerlessStartJobOperator"
[2026-05-20, 16:01:25] INFO - Using backported waiter_with_logging.wait (module=waiter_with_logging, file=/usr/local/airflow/dags/mynamespace_ns/_commonutil/waiter_with_logging.py, max_attempts=480, delay=60s, args={'applicationId': 'myappid', 'jobRunId': 'jobid'}): source="waiter_with_logging"
[2026-05-20, 16:01:25] INFO - Serverless Job status is [attempt 1/480]: SUBMITTED - SUBMITTED: source="waiter_with_logging"
[2026-05-20, 16:01:41] ERROR - Server indicated the task shouldn't be running anymore. Terminating process: detail={"detail":{"reason":"not_running","message":"TI is no longer in the running state and task should terminate","current_state":"failed"}}: source="task"
[2026-05-20, 16:01:41] INFO - Stopping job run with jobId - jobid: source="airflow.task.operators.edfx_emr_serverless_operator.EdfxEmrServerlessStartJobOperator"
[2026-05-20, 16:01:41] ERROR - Task failed with exception: source="task"ClientError: An error occurred (AccessDeniedException) when calling the CancelJobRun operation: User: arn:aws:sts::accid:assumed-role/abc-MwaaEnvRole/AmazonMWAA-iamrole is not authorized to perform: emr-serverless:CancelJobRun on resource: arn:aws:emr-serverless:us-west-2:accid:/applications/myappid/jobruns/jobid because no identity-based policy allows the emr-serverless:CancelJobRun action
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 920 in run
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1215 in _execute_task
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 397 in wrapper
File "/usr/local/airflow/dags/mynamespace_ns/_commonutil/edfx_emr_serverless_operator.py", line 101 in execute
File "/usr/local/airflow/dags/mynamespace_ns/_commonutil/waiter_with_logging.py", line 101 in wait
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 891 in _on_term
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/amazon/aws/operators/emr.py", line 1294 in on_kill
File "/usr/local/airflow/.local/lib/python3.12/site-packages/botocore/client.py", line 601 in _api_call
File "/usr/local/airflow/.local/lib/python3.12/site-packages/botocore/context.py", line 123 in wrapper
File "/usr/local/airflow/.local/lib/python3.12/site-packages/botocore/client.py", line 1074 in _make_api_call
[2026-05-20, 16:01:41] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_s3_logs": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_endispositionreason": run_id="manual__2026-05-20T09:55:43.497627+00:00": map_index=-1: source="task"
[2026-05-20, 16:01:41] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_cloudwatch_logs": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_endispositionreason": run_id="manual__2026-05-20T09:55:43.497627+00:00": map_index=-1: source="task"
[2026-05-20, 16:01:41] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_dashboard": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_endispositionreason": run_id="manual__2026-05-20T09:55:43.497627+00:00": map_index=-1: source="task"
[2026-05-20, 16:01:41] WARNING - No XCom value found; defaulting to None.: key="emr_serverless_logs": dag_id="mynamespace_xxxxx": task_id="KP.mynamespace_csv_ingest_endispositionreason": run_id="manual__2026-05-20T09:55:43.497627+00:00": map_index=-1: source="task"
[2026-05-20, 16:01:41] ERROR - Top level error: source="task"UndefinedError: 'airflow.sdk.execution_time.task_runner.RuntimeTaskInstance object' has no attribute 'mark_success_url'
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1353 in main
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1330 in finalize
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1161 in _send_task_error_email
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 411 in _get_email_subject_content
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 408 in render
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/utils/helpers.py", line 244 in render_template_to_string
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/utils/helpers.py", line 239 in render_template
File "<template>", line 26 in root
File "/usr/local/airflow/.local/lib/python3.12/site-packages/jinja2/runtime.py", line 859 in _fail_with_undefined_error
[2026-05-20, 16:01:41] WARNING - Process exited abnormally: exit_code=1: source="task"
[2026-05-20, 16:01:41] ERROR - Task killed!: source="task"
What you think should happen instead?
If job was submitted to emr successfully, task should reflect it and should proceed fine without any failure.
Operating System
No response
Deployment
Amazon (AWS) MWAA
Apache Airflow Provider(s)
amazon
Versions of Apache Airflow Providers
apache-airflow-providers-amazon==9.12.0
Official Helm Chart version
Not Applicable
Kubernetes Version
No response
Helm Chart configuration
No response
Docker Image customizations
No response
Anything else?
Tried using waiter max attempts=500 and waiter delay =60s but nothing helped
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
Activity
Subham-KRLX commentedon May 21, 2026
PR #67222 fixes the waiter throttling issue However your real problem is the TaskInstance being marked FAILED at 16 seconds while the job still runs in EMR this needs investigation there is also a cascade bug with mark_success_url undefined in task_runner.py Can you share your execution_timeout and deferrable settings on EmrServerlessStartJobOperator.
ROOBALJINDAL commentedon May 21, 2026
deferrable=false and we are using airflow pool=4
Subham-KRLX commentedon May 21, 2026
Are you using deferrable=True by any chance on some tasks? Also can you check your Airflow logs (scheduler and worker) at the exact timestamps when tasks fail to see if there are timeout or heartbeat messages? The task seems to be killed by the executor at 16 seconds even though execution_timeout is 2 hours 15 minutes. This suggests an issue with task heartbeat/status synchronization rather than a timeout configuration issue.
ROOBALJINDAL commentedon May 21, 2026
No, we are not using deferrable property.
fyi..
Update:
We logged a ticket with AWS guys too and they responded
We tried this but still faced the same issue
Subham-KRLX commentedon May 21, 2026
Thanks for confirming with AWS this is a Celery fork bug affecting all Airflow 3.0.X versions the worker autoscale workaround didn't help because it doesn't fix the underlying fork deadlock issue. The waiter throttling fix in PR 67222 won't solve this either your jobs fail because Celery kills the process mid execution due to thread deadlock from fork your best options are to switch to deferrable tasks wait for Airflow 3.1 to see if they fixed the Celery fork issue, or stay on 2.7.2 until 3.1 is stable this is an Airflow core issue not an EMR provider issue so you should escalate to Airflow maintainers to prioritize the Celery fork fix for 3.1 or a patch release.
eladkal commentedon May 22, 2026
@ROOBALJINDAL all indications you provided suggesting your issue is MWAA specific thus Airflow open source can't really do anything about it.
eladkal commentedon May 22, 2026
I am closing this issue as MWAA specific and advise you to contact AWS for this.
If any new information comes to light please comment and we can revisit.