<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.5.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">when using a dynamically generated task that gets the expand data from xcom after a branch_task the graph doesn't render.<br>
It reappears once the dag run is finished.<br>
tried with BashOperator and a KubernetesPodOperator.</p>
<p dir="auto">the developer console in the browser shows the error:<br>
<code class="notranslate">Uncaught TypeError: Cannot read properties of undefined (reading 'length') at z (graph.1c0596dfced26c638bfe.js:2:17499) at graph.1c0596dfced26c638bfe.js:2:17654 at Array.map (&lt;anonymous&gt;) at z (graph.1c0596dfced26c638bfe.js:2:17646) at graph.1c0596dfced26c638bfe.js:2:26602 at graph.1c0596dfced26c638bfe.js:2:26655 at graph.1c0596dfced26c638bfe.js:2:26661 at graph.1c0596dfced26c638bfe.js:2:222 at graph.1c0596dfced26c638bfe.js:2:227 z @ graph.1c0596dfced26c638bfe.js:2 (anonymous) @ graph.1c0596dfced26c638bfe.js:2 z @ graph.1c0596dfced26c638bfe.js:2 (anonymous) @ graph.1c0596dfced26c638bfe.js:2 (anonymous) @ graph.1c0596dfced26c638bfe.js:2 (anonymous) @ graph.1c0596dfced26c638bfe.js:2 (anonymous) @ graph.1c0596dfced26c638bfe.js:2 (anonymous) @ graph.1c0596dfced26c638bfe.js:2 </code></p>
<p dir="auto">grid view renders fine.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">graph should be rendered.</p>
<h3 dir="auto">How to reproduce</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="def branch_dynamic_flow():
    @branch_task
    def choose_path():
        return 'b'


    @task
    def a():
        print('a')

    @task
    def get_args():
        return ['echo 1', 'echo 2']

    b = BashOperator.partial(task_id=&quot;b&quot;).expand(bash_command=get_args())
    path = choose_path()
    path &gt;&gt; a()
    path &gt;&gt; b
"><pre lang="@dag('branch_dynamic'," class="notranslate"><code class="notranslate">def branch_dynamic_flow():
    @branch_task
    def choose_path():
        return 'b'


    @task
    def a():
        print('a')

    @task
    def get_args():
        return ['echo 1', 'echo 2']

    b = BashOperator.partial(task_id="b").expand(bash_command=get_args())
    path = choose_path()
    path &gt;&gt; a()
    path &gt;&gt; b

</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">red hat</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-cncf-kubernetes | 5.1.1 | Kubernetes</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></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" checked=""> 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>
<p dir="auto">I am trying to install three node airflow cluster. Each node has airflow scheduler, airflow worker, airflow webserver, also it has celery, RabbitMQ cluster and Postgres multi master cluster(implemented with Bucardo). Versions of software:</p>
<ul dir="auto">
<li>Airflow 2.0.1</li>
<li>Postregsql 13.2</li>
<li>Ubuntu 20.04</li>
<li>pyhton 3.8.5</li>
<li>celery 4.4.7</li>
<li>bucardo 5.6.0</li>
<li>RabbitMQ 3.8.2</li>
</ul>
<p dir="auto">And I occur the problem starting airflow scheduler.</p>
<p dir="auto">When I launch the first one(database is empty), it successfully starts. When scheduler is running on the machine, then when I launch another scheduler on another machine(I tried to launch on the same machine too), it fails with the following:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint &quot;job_pkey&quot;
DETAIL:  Key (id)=(25) already exists.

[SQL: INSERT INTO job (dag_id, state, job_type, start_date, end_date, latest_heartbeat, executor_class, hostname, unixname) VALUES (%(dag_id)s, %(state)s, %(job_type)s, %(start_date)s, %(end_date)s, %(latest_heartbeat)s, %(executor_class)s, %(hostname)s, %(unixname)s) RETURNING job.id]
[parameters: {'dag_id': None, 'state': 'running', 'job_type': 'SchedulerJob', 'start_date': datetime.datetime(2021, 4, 21, 7, 39, 20, 429478, tzinfo=Timezone('UTC')), 'end_date': None, 'latest_heartbeat': datetime.datetime(2021, 4, 21, 7, 39, 20, 429504, tzinfo=Timezone('UTC')), 'executor_class': 'CeleryExecutor', 'hostname': 'hostname', 'unixname': 'root'}]
(Background on this error at: http://sqlalche.me/e/13/gkpj)
"><pre class="notranslate"><code class="notranslate">sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "job_pkey"
DETAIL:  Key (id)=(25) already exists.

[SQL: INSERT INTO job (dag_id, state, job_type, start_date, end_date, latest_heartbeat, executor_class, hostname, unixname) VALUES (%(dag_id)s, %(state)s, %(job_type)s, %(start_date)s, %(end_date)s, %(latest_heartbeat)s, %(executor_class)s, %(hostname)s, %(unixname)s) RETURNING job.id]
[parameters: {'dag_id': None, 'state': 'running', 'job_type': 'SchedulerJob', 'start_date': datetime.datetime(2021, 4, 21, 7, 39, 20, 429478, tzinfo=Timezone('UTC')), 'end_date': None, 'latest_heartbeat': datetime.datetime(2021, 4, 21, 7, 39, 20, 429504, tzinfo=Timezone('UTC')), 'executor_class': 'CeleryExecutor', 'hostname': 'hostname', 'unixname': 'root'}]
(Background on this error at: http://sqlalche.me/e/13/gkpj)

</code></pre></div>
<p dir="auto">After trying to launch a few times eventually scheduler is working. I am assuming id is incremented and then data is successfully added into database:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="airflow=&gt; select * from job order by state;
 id | dag_id |  state  |   job_type   |          start_date           |           end_date            |       latest_heartbeat        | executor_class |           hostname           | unixname 
----+--------+---------+--------------+-------------------------------+-------------------------------+-------------------------------+----------------+------------------------------+----------
 26 |        | running | SchedulerJob | 2021-04-21 07:39:22.243721+00 |                               | 2021-04-21 07:39:22.243734+00 | CeleryExecutor |                machine name  | root
 25 |        | running | SchedulerJob | 2021-04-21 07:39:14.515009+00 |                               | 2021-04-21 07:39:19.632811+00 | CeleryExecutor |                machine name  | root 
"><pre class="notranslate"><code class="notranslate">airflow=&gt; select * from job order by state;
 id | dag_id |  state  |   job_type   |          start_date           |           end_date            |       latest_heartbeat        | executor_class |           hostname           | unixname 
----+--------+---------+--------------+-------------------------------+-------------------------------+-------------------------------+----------------+------------------------------+----------
 26 |        | running | SchedulerJob | 2021-04-21 07:39:22.243721+00 |                               | 2021-04-21 07:39:22.243734+00 | CeleryExecutor |                machine name  | root
 25 |        | running | SchedulerJob | 2021-04-21 07:39:14.515009+00 |                               | 2021-04-21 07:39:19.632811+00 | CeleryExecutor |                machine name  | root 

</code></pre></div>
<p dir="auto">There is a warning with log tables as well(If the second and subsequent schedulers successfully started):</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="WARNING - Failed to log action with (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint &quot;log_pkey&quot;
DETAIL:  Key (id)=(40) already exists."><pre class="notranslate"><code class="notranslate">WARNING - Failed to log action with (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "log_pkey"
DETAIL:  Key (id)=(40) already exists.
</code></pre></div>
<p dir="auto">If I succeeded in launching scheduler on another node(i.e. schedulers are working on two mahines), attempt to launch another instance on one of those machine will fail with the same error.</p>
<p dir="auto">I understand why scheduler cannot insert data into table, but I assume something's wrong with architecture or table structure.</p>