Issue 1:
We are using EC2 auto-scaling. It turns out that our airflow machines are on the autoscaled EC2 instances, resulting in multiple machines launching schedulers with LocalExecutors. Is this a design pattern that is dangerous?

We originally worked around this using EC2 "user-data" scripts that fire after an EC2 auto-scaled instances is launched -- the "user-data" script kills the scheduler processes. However, we suspect that some of these schedulers pick up duplicate work and execute them before they are killed.

Hence, I am wondering how the scheduler with local executor works in a topology with schedulers running on multiple machines.

Issue 2:
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