Struggling with unique key constrain violations

[ginwakeup]

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

Airflow keeps polluting the postgres DB. I cannot understand what causes this but it happens in our cluster every 3/4 weeks.
At some point the triggerer stops working and throw errors such as these:

Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/__main__.py", line 39, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/cli_parser.py", line 52, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py", line 108, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/triggerer_command.py", line 61, in triggerer
    job.run()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 247, in run
    self._execute()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/triggerer_job.py", line 106, in _execute
    self._run_trigger_loop()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/triggerer_job.py", line 135, in _run_trigger_loop
    self.heartbeat(only_if_necessary=True)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 205, in heartbeat
    previous_heartbeat = self.latest_heartbeat
  File "/usr/local/lib/python3.7/contextlib.py", line 119, in __exit__
    next(self.gen)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 36, in create_session
    session.commit()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 1451, in commit
    self._transaction.commit(_to_root=self.future)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 829, in commit
    self._prepare_impl()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl
    self.session.flush()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 3444, in flush
    self._flush(objects)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 3584, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/langhelpers.py", line 72, in __exit__
    with_traceback=exc_tb,
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 210, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 3544, in _flush
    flush_context.execute()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py", line 456, in execute
    rec.execute(self)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py", line 633, in execute
    uow,
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py", line 250, in save_obj
    insert,
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py", line 1098, in _emit_insert_statements
    statement, multiparams, execution_options=execution_options
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/sql/elements.py", line 335, in _execute_on_connection
    self, multiparams, params, execution_options
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1582, in _execute_clauseelement
    cache_hit=cache_hit,
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1944, in _execute_context
    e, statement, parameters, cursor, context
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 2125, in _handle_dbapi_exception
    sqlalchemy_exception, with_traceback=exc_info[2], from_=e
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 210, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1901, in _execute_context
    cursor, statement, parameters, context
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "job_pkey"
DETAIL:  Key (id)=(46430) already exists.

I cannot understand what is causing it and it forces me to erase the DB every time to make it work again.

Nothing changed on the DAGs at all, this was all working since 3 days ago and it just started failing on DB data. Any idea?

What you think should happen instead

No response

How to reproduce

I don't know.

Operating System

Ubuntu 20.04

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

Deployed on Kubernetes, version 2.5
6 replicas for each component
db backend: postgres 15

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

[jscheffl]

Can you please add some details which Airflow version you are actually using? Else it is a bit shooting in the dark, what the problem might be.

Taking a look to the stack trace it seems that you call some job / trigger a DAG via CLI. Would be very important to classify the bug or understand to know which CLI params you are using.

[ginwakeup]

@jens-scheffler-bosch 2.5.0

[ginwakeup]

@jens-scheffler-bosch More details on the setup:

• gitSync pulls DAGS and it runs on a separate pod which mounts a PVC
• The PVC is mounted on all the airflow components
• the AIRFLOW_DAGS_FOLDER is set to point to the PVC location
• Airflow is executed using the official HELM chart

DAGS are executed using Airflow API calls by external apps.

[ginwakeup]

I've also just tried to run airflow db reset but it didn't help.

[ginwakeup]

@jens-scheffler-bosch forgive me, that's not the scheduler logs, it's the triggerer.

The scheduler errors like this:

[2023-06-16T22:01:08.483+0000] {settings.py:407} DEBUG - Disposing DB connection pool (PID 6602)
Process DagFileProcessor910-Process:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "/usr/local/lib/python3.7/multiprocessing/process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/dag_processing/processor.py", line 174, in _run_file_processor
    _handle_dag_file_processing()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/dag_processing/processor.py", line 158, in _handle_dag_file_processing
    callback_requests=callback_requests,
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/dag_processing/processor.py", line 768, in process_file
    dagbag.sync_to_db(processor_subdir=self._dag_directory, session=session)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 72, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 645, in sync_to_db
    for attempt in run_with_db_retries(logger=self.log):
  File "/home/airflow/.local/lib/python3.7/site-packages/tenacity/__init__.py", line 384, in __iter__
    do = self.iter(retry_state=retry_state)
  File "/home/airflow/.local/lib/python3.7/site-packages/tenacity/__init__.py", line 351, in iter
    return fut.result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 428, in result
    return self.__get_result()
  File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagbag.py", line 660, in sync_to_db
    self.dags.values(), processor_subdir=processor_subdir, session=session
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py", line 72, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dag.py", line 2687, in bulk_write_to_db
    orm_dags: list[DagModel] = with_row_locks(query, of=DagModel, session=session).all()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 2772, in all
    return self._iter().all()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 2918, in _iter
    execution_options={"_sa_orm_load_options": self.load_options},
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 1713, in execute
    conn = self._connection_for_bind(bind)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 1553, in _connection_for_bind
    engine, execution_options
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 721, in _connection_for_bind
    self._assert_active()
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 608, in _assert_active
    code="7s2a",
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 "serialized_dag_pkey"
DETAIL:  Key (dag_id)=(***) already exists.

[ginwakeup]

Triggerer exception as soon as it starts:

[2023-06-16T22:08:22.757+0000] {triggerer_job.py:108} ERROR - Exception when executing TriggererJob._run_trigger_loop
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1901, in _execute_context
    cursor, statement, parameters, context
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "job_pkey"
DETAIL:  Key (id)=(39) already exists.

[ginwakeup]

Same with scheduler:

[2023-06-16T22:14:40.368+0000] {scheduler_job.py:776} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1901, in _execute_context
    cursor, statement, parameters, context
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "job_pkey"
DETAIL:  Key (id)=(3) already exists.

I do not understand how this is possible after the DB has been completely cleared.

[ginwakeup]

Seems like these are dags executed by the scheduler..

[SQL: INSERT INTO job (id, dag_id, state, job_type, start_date, end_date, latest_heartbeat, executor_class, hostname, unixname) VALUES (%(id)s, %(dag_id)s, %(state)s, %(job_type)s, %(start_date)s, %(end_date)s, %(latest_heartbeat)s, %(executor_class)s, %(hostname)s, %(unixname)s)]

[parameters: {'id': 3, 'dag_id': None, 'state': <TaskInstanceState.RUNNING: 'running'>, 'job_type': 'SchedulerJob', 'start_date': datetime.datetime(2023, 6, 16, 22, 14, 9, 709334, tzinfo=Timezone('***')), 'end_date': None, 'latest_heartbeat': datetime.datetime(2023, 6, 16, 22, 14, 9, 709348, tzinfo=Timezone('***')), 'executor_class': 'CeleryKubernetesExecutor', 'hostname': 'airflow-scheduler-***-*****', 'unixname': 'airflow'}]

[potiuk]

The way it looks - it's a problem of your Database. You likely try to use some active-active DB (not standard Postgres, but something that builds on top of Postgres).

Likely this is some non-standard active-active setup with some load balancing to route your DB connection between those two. - i.e. multiple DB instances, each serving the traffic, each trying to insert its own keys.

Airflow (and generally any application that is not build to run in such setup, and specifically any application that is build using SQLAlchemy without specifically accounting for such active-active setup) will fail because different replicas of the DB will attempt to insert the same unique keys when attempting to create a record in a table with automated unique key generated by SqlAlchemy.

Airflow will only work with Active-Passive high availability DB - i.e one that you have only one node handling traffic at a time and the second is a replica, which can be used as a failover.

The solution to your problem is to have a database that does not have active-active setup.

BTW. Using Airflow 2.5.0 is a bad idea, there were probably few dosens bugfixes in 2.5 line alone (2.5.3 was release quite a few months ago) so using 2.5.0 when those fixes are out there makes very little sense, because you do not give yourself a chance to make use of those bug-fixes that were implemented (even if you do not want to switch to latest 2.6.* which I woudl heartily recommend).

Converting it into discussion to continue discussion there because it is unlikely to be an airflow issue.

@ginwakeup -> I am curious to see later what kind of database solution you use, almost for certain this is the reason of your problem.

[ginwakeup]

@potiuk thank you for the detailed reply, I was wondering if this was caused by postgres setup indeed.
I will switch to what you suggested and I will update the discussion.

[potiuk]

what DB solution did you use ? Curious if the claim for it is that it will "just work" same as standard Postgres DB? I am really interested in hearing that

[potiuk]

If only to warn our users explicitely - that might help others not to make the same mistake.

[potiuk]

@ginwakeup -> would be great to undedstand it :). I looked through the issue raised for help, and it woudl be great to know how to avoid such requests in the future and warn our users. Was it something you setup in a custom way (what?) or was it some external DB ? - that would help us to describe explicit warning in our docs for the future users.