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

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 <lambda>
    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)