apache / apache/airflow

DAG processor retry surfaces PendingRollbackError and hides the original IntegrityError from dag_tag unique violation

Open
#68,263 1 comment 0 reactions 0 assignees View on GitHub
area:core kind:bug needs-triage
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.