<p dir="auto">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?</p>
<p dir="auto">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.</p>
<p dir="auto">Hence, I am wondering how the scheduler with local executor works in a topology with schedulers running on multiple machines.</p>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">Other Airflow 2 version (please specify below)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Airflow keeps polluting the postgres DB. I cannot understand what causes this but it happens in our cluster every 3/4 weeks.<br>
At some point the triggerer stops working and throw errors such as these:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/home/airflow/.local/bin/airflow&quot;, line 8, in &lt;module&gt;
    sys.exit(main())
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/__main__.py&quot;, line 39, in main
    args.func(args)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/cli_parser.py&quot;, line 52, in command
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/cli.py&quot;, line 108, in wrapper
    return f(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/cli/commands/triggerer_command.py&quot;, line 61, in triggerer
    job.run()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py&quot;, line 247, in run
    self._execute()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/triggerer_job.py&quot;, line 106, in _execute
    self._run_trigger_loop()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/triggerer_job.py&quot;, line 135, in _run_trigger_loop
    self.heartbeat(only_if_necessary=True)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/base_job.py&quot;, line 205, in heartbeat
    previous_heartbeat = self.latest_heartbeat
  File &quot;/usr/local/lib/python3.7/contextlib.py&quot;, line 119, in __exit__
    next(self.gen)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py&quot;, line 36, in create_session
    session.commit()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 1451, in commit
    self._transaction.commit(_to_root=self.future)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 829, in commit
    self._prepare_impl()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 808, in _prepare_impl
    self.session.flush()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 3444, in flush
    self._flush(objects)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 3584, in _flush
    transaction.rollback(_capture_exception=True)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/langhelpers.py&quot;, line 72, in __exit__
    with_traceback=exc_tb,
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/compat.py&quot;, line 210, in raise_
    raise exception
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/session.py&quot;, line 3544, in _flush
    flush_context.execute()
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py&quot;, line 456, in execute
    rec.execute(self)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/unitofwork.py&quot;, line 633, in execute
    uow,
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py&quot;, line 250, in save_obj
    insert,
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/persistence.py&quot;, line 1098, in _emit_insert_statements
    statement, multiparams, execution_options=execution_options
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1705, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/sql/elements.py&quot;, line 335, in _execute_on_connection
    self, multiparams, params, execution_options
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1582, in _execute_clauseelement
    cache_hit=cache_hit,
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1944, in _execute_context
    e, statement, parameters, cursor, context
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 2125, in _handle_dbapi_exception
    sqlalchemy_exception, with_traceback=exc_info[2], from_=e
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/compat.py&quot;, line 210, in raise_
    raise exception
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py&quot;, line 1901, in _execute_context
    cursor, statement, parameters, context
  File &quot;/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/default.py&quot;, line 736, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint &quot;job_pkey&quot;
DETAIL:  Key (id)=(46430) already exists."><pre class="notranslate"><code class="notranslate">Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in &lt;module&gt;
    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.
</code></pre></div>
<p dir="auto">I cannot understand what is causing it and it forces me to erase the DB every time to make it work again.</p>
<p dir="auto">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?</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">I don't know.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu 20.04</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Deployed on Kubernetes, version 2.5<br>
6 replicas for each component<br>
db backend: postgres  15</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>