DAG processor retry surfaces PendingRollbackError and hides the original IntegrityError from dag_tag unique violation
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.1.8
### What happened and how to reproduce it?
The DAG processor crashed with a `sqlalchemy.exc.PendingRollbackError.` The "Original exception was:" tail of the formatted error shows the actual cause: a `psycopg2.errors.UniqueViolation` on `dag_tag_pkey` during `bulk_write_to_db`.
```
Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/airflow/jobs/dag_processor_job_runner.py", line 61, in _execute
self.processor.run()
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/manager.py", line 297, in run
return self._run_parsing_loop()
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/manager.py", line 401, in _run_parsing_loop
self._collect_results()
File "/usr/local/lib/python3.11/site-packages/airflow/utils/session.py", line 100, in wrapper
return func(*args, session=session, **kwargs)
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/manager.py", line 882, in _collect_results
self._file_stats[file] = process_parse_results(
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/manager.py", line 1276, in process_parse_results
update_dag_parsing_results_in_db(
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/collection.py", line 400, in update_dag_parsing_results_in_db
for attempt in run_with_db_retries(logger=log):
File "/usr/local/lib/python3.11/site-packages/tenacity/__init__.py", line 438, in __iter__
do = self.iter(retry_state=retry_state)
File "/usr/local/lib/python3.11/site-packages/tenacity/__init__.py", line 371, in iter
result = action(retry_state)
File "/usr/local/lib/python3.11/site-packages/tenacity/__init__.py", line 393, in
self._add_action_func(lambda rs: rs.outcome.result())
File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 449, in result
return self.__get_result()
File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
raise self._exception
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/collection.py", line 410, in update_dag_parsing_results_in_db
SerializedDAG.bulk_write_to_db(
File "/usr/local/lib/python3.11/site-packages/airflow/utils/session.py", line 98, in wrapper
return func(*args, **kwargs)
File "/usr/local/lib/python3.11/site-packages/airflow/serialization/serialized_objects.py", line 2956, in bulk_write_to_db
orm_dags = dag_op.add_dags(session=session)
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/collection.py", line 470, in add_dags
orm_dags = self.find_orm_dags(session=session)
File "/usr/local/lib/python3.11/site-packages/airflow/dag_processing/collection.py", line 467, in find_orm_dags
return {dm.dag_id: dm for dm in session.scalars(stmt).unique()}
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1778, in scalars
return self.execute(
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1716, in execute
conn = self._connection_for_bind(bind)
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1555, in _connection_for_bind
return self._transaction._connection_for_bind(
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 724, in _connection_for_bind
self._assert_active()
File "/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 604, in _assert_active
raise sa_exc.PendingRollbackError(
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_tag_pkey"
DETAIL: Key (name, dag_id)=(is_gdpr=False, featuregen_dag_generator) already exists.
[SQL: INSERT INTO dag_tag (name, dag_id) VALUES (%(name)s, %(dag_id)s)]
[parameters: {'name': 'is_gdpr=False', 'dag_id': 'featuregen_dag_generator'}]
(Background on this error at: https://sqlalche.me/e/14/gkpj) (Background on this error at: https://sqlalche.me/e/14/7s2a)
```
Happened transiently; not sure how to reproduce.
### What you think should happen instead?
_No response_
### Operating System
debian
### Deployment
Astronomer
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
Deployment is running in HA mode with two DAG processors.
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Assessment
This issue has not been assessed yet.