<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.0 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Returning a list from a <code class="notranslate">PythonOperator</code> and passing it on to a <code class="notranslate">PostgresOperator</code> using <code class="notranslate">expand</code> causes the scheduler to crash:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="scheduler | [2022-05-06 15:44:47,317] {dag.py:2915} INFO - Setting next_dagrun for sum-dag to 2022-05-06T00:00:00+00:00, run_after=2022-05-07T00:00:00+00:00
scheduler | [2022-05-06 15:44:47,384] {scheduler_job.py:753} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
scheduler | Traceback (most recent call last):
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 736, in _execute
scheduler | self._run_scheduler_loop()
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 824, in _run_scheduler_loop
scheduler | num_queued_tis = self._do_scheduling(session)
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 906, in _do_scheduling
scheduler | callback_to_run = self._schedule_dag_run(dag_run, session)
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1148, in _schedule_dag_run
scheduler | schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
scheduler | return func(*args, **kwargs)
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 522, in update_state
scheduler | info = self.task_instance_scheduling_decisions(session)
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
scheduler | return func(*args, **kwargs)
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 658, in task_instance_scheduling_decisions
scheduler | schedulable_tis, changed_tis, expansion_happened = self._get_ready_tis(
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 714, in _get_ready_tis
scheduler | expanded_tis, _ = schedulable.task.expand_mapped_task(self.run_id, session=session)
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/mappedoperator.py&quot;, line 609, in expand_mapped_task
scheduler | operator.mul, self._resolve_map_lengths(run_id, session=session).values()
scheduler | File &quot;/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/mappedoperator.py&quot;, line 595, in _resolve_map_lengths
scheduler | raise RuntimeError(f&quot;Failed to populate all mapping metadata; missing: {keys}&quot;)
scheduler | RuntimeError: Failed to populate all mapping metadata; missing: 'parameters'
scheduler | [2022-05-06 15:44:48,410] {process_utils.py:125} INFO - Sending Signals.SIGTERM to group 92610. PIDs of all processes in the group: [92610]
scheduler | [2022-05-06 15:44:48,412] {process_utils.py:80} INFO - Sending the signal Signals.SIGTERM to group 92610
scheduler | [2022-05-06 15:44:48,684] {process_utils.py:75} INFO - Process psutil.Process(pid=92610, status='terminated', exitcode=0, started='15:44:25') (92610) terminated with exit code 0
scheduler | [2022-05-06 15:44:48,685] {scheduler_job.py:765} INFO - Exited execute loop"><pre class="notranslate"><code class="notranslate">scheduler | [2022-05-06 15:44:47,317] {dag.py:2915} INFO - Setting next_dagrun for sum-dag to 2022-05-06T00:00:00+00:00, run_after=2022-05-07T00:00:00+00:00
scheduler | [2022-05-06 15:44:47,384] {scheduler_job.py:753} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
scheduler | Traceback (most recent call last):
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 736, in _execute
scheduler | self._run_scheduler_loop()
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 824, in _run_scheduler_loop
scheduler | num_queued_tis = self._do_scheduling(session)
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 906, in _do_scheduling
scheduler | callback_to_run = self._schedule_dag_run(dag_run, session)
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 1148, in _schedule_dag_run
scheduler | schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
scheduler | return func(*args, **kwargs)
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/dagrun.py", line 522, in update_state
scheduler | info = self.task_instance_scheduling_decisions(session)
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
scheduler | return func(*args, **kwargs)
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/dagrun.py", line 658, in task_instance_scheduling_decisions
scheduler | schedulable_tis, changed_tis, expansion_happened = self._get_ready_tis(
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/dagrun.py", line 714, in _get_ready_tis
scheduler | expanded_tis, _ = schedulable.task.expand_mapped_task(self.run_id, session=session)
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/mappedoperator.py", line 609, in expand_mapped_task
scheduler | operator.mul, self._resolve_map_lengths(run_id, session=session).values()
scheduler | File "/Users/example/aaa/vvv/lib/python3.9/site-packages/airflow/models/mappedoperator.py", line 595, in _resolve_map_lengths
scheduler | raise RuntimeError(f"Failed to populate all mapping metadata; missing: {keys}")
scheduler | RuntimeError: Failed to populate all mapping metadata; missing: 'parameters'
scheduler | [2022-05-06 15:44:48,410] {process_utils.py:125} INFO - Sending Signals.SIGTERM to group 92610. PIDs of all processes in the group: [92610]
scheduler | [2022-05-06 15:44:48,412] {process_utils.py:80} INFO - Sending the signal Signals.SIGTERM to group 92610
scheduler | [2022-05-06 15:44:48,684] {process_utils.py:75} INFO - Process psutil.Process(pid=92610, status='terminated', exitcode=0, started='15:44:25') (92610) terminated with exit code 0
scheduler | [2022-05-06 15:44:48,685] {scheduler_job.py:765} INFO - Exited execute loop
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The scheduler should not crash and the <code class="notranslate">PostgresOperator</code> should get executed.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">The execution of this DAG fails with the exception posted above:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="import pendulum
from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.decorators import task


with DAG(
    dag_id=&quot;sum-dag&quot;,
    start_date=pendulum.datetime(2021, 11, 19, tz=&quot;UTC&quot;),
    schedule_interval=&quot;@daily&quot;,
    catchup=False,
) as dag:


    @task
    def input_list():
        return [{&quot;v&quot;: 1}]


    PostgresOperator.partial(
        task_id=&quot;test2&quot;,
        sql=&quot;SELECT %(v)s&quot;,
    ).expand(parameters=input_list())"><pre class="notranslate"><span class="pl-k">import</span> <span class="pl-s1">pendulum</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">providers</span>.<span class="pl-s1">postgres</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">postgres</span> <span class="pl-k">import</span> <span class="pl-v">PostgresOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">decorators</span> <span class="pl-k">import</span> <span class="pl-s1">task</span>


<span class="pl-k">with</span> <span class="pl-v">DAG</span>(
    <span class="pl-s1">dag_id</span><span class="pl-c1">=</span><span class="pl-s">"sum-dag"</span>,
    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-s1">pendulum</span>.<span class="pl-en">datetime</span>(<span class="pl-c1">2021</span>, <span class="pl-c1">11</span>, <span class="pl-c1">19</span>, <span class="pl-s1">tz</span><span class="pl-c1">=</span><span class="pl-s">"UTC"</span>),
    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-s">"@daily"</span>,
    <span class="pl-s1">catchup</span><span class="pl-c1">=</span><span class="pl-c1">False</span>,
) <span class="pl-k">as</span> <span class="pl-s1">dag</span>:


    <span class="pl-en">@<span class="pl-s1">task</span></span>
    <span class="pl-k">def</span> <span class="pl-en">input_list</span>():
        <span class="pl-k">return</span> [{<span class="pl-s">"v"</span>: <span class="pl-c1">1</span>}]


    <span class="pl-v">PostgresOperator</span>.<span class="pl-en">partial</span>(
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"test2"</span>,
        <span class="pl-s1">sql</span><span class="pl-c1">=</span><span class="pl-s">"SELECT %(v)s"</span>,
    ).<span class="pl-en">expand</span>(<span class="pl-s1">parameters</span><span class="pl-c1">=</span><span class="pl-en">input_list</span>())</pre></div>
<p dir="auto">Passing a <code class="notranslate">list</code> directly to the <code class="notranslate">PostgresOperator</code> (instead of having it returned from a <code class="notranslate">PythonOperator</code>) does work successfully:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="import pendulum
from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.decorators import task


with DAG(
    dag_id=&quot;sum-dag&quot;,
    start_date=pendulum.datetime(2021, 11, 19, tz=&quot;UTC&quot;),
    schedule_interval=&quot;@daily&quot;,
    catchup=False,
) as dag:


    input_list_simple = [{&quot;v&quot;: 1}]
    PostgresOperator.partial(
        task_id=&quot;test2&quot;,
        sql=&quot;SELECT %(v)s&quot;,
    ).expand(parameters=input_list_simple)"><pre class="notranslate"><span class="pl-k">import</span> <span class="pl-s1">pendulum</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">providers</span>.<span class="pl-s1">postgres</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">postgres</span> <span class="pl-k">import</span> <span class="pl-v">PostgresOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">decorators</span> <span class="pl-k">import</span> <span class="pl-s1">task</span>


<span class="pl-k">with</span> <span class="pl-v">DAG</span>(
    <span class="pl-s1">dag_id</span><span class="pl-c1">=</span><span class="pl-s">"sum-dag"</span>,
    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-s1">pendulum</span>.<span class="pl-en">datetime</span>(<span class="pl-c1">2021</span>, <span class="pl-c1">11</span>, <span class="pl-c1">19</span>, <span class="pl-s1">tz</span><span class="pl-c1">=</span><span class="pl-s">"UTC"</span>),
    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-s">"@daily"</span>,
    <span class="pl-s1">catchup</span><span class="pl-c1">=</span><span class="pl-c1">False</span>,
) <span class="pl-k">as</span> <span class="pl-s1">dag</span>:


    <span class="pl-s1">input_list_simple</span> <span class="pl-c1">=</span> [{<span class="pl-s">"v"</span>: <span class="pl-c1">1</span>}]
    <span class="pl-v">PostgresOperator</span>.<span class="pl-en">partial</span>(
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"test2"</span>,
        <span class="pl-s1">sql</span><span class="pl-c1">=</span><span class="pl-s">"SELECT %(v)s"</span>,
    ).<span class="pl-en">expand</span>(<span class="pl-s1">parameters</span><span class="pl-c1">=</span><span class="pl-s1">input_list_simple</span>)</pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">macOS 12.3.1</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><code class="notranslate">apache-airflow-providers-postgres==4.1.0</code></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Virtualenv installation</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"> 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>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.0 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I have a dag that looks like this.<br>
When I uncomment <code class="notranslate">py_job</code>(Dynamically mapped PythonOperator) it works well with <code class="notranslate">pull_messages</code> (Taskflow API).<br>
When I try to do the same with <code class="notranslate">DatabricksRunNowOperator</code> it crashes the scheduler with error</p>
<p dir="auto">Related issues <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1225684729" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/23486" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/23486/hovercard" href="https://github.com/apache/airflow/pull/23486">#23486</a></p>
<h3 dir="auto">Sample DAG</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="import json
import pendulum
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator

@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2021, 1, 1, tz=&quot;UTC&quot;),
    catchup=False,
    tags=['example'],
)
def tutorial_taskflow_api_etl():
    def random(*args, **kwargs):
        print (&quot;==== kwargs inside random ====&quot;, args, kwargs)
        print (&quot;I'm random&quot;)
        return 49

    @task
    def pull_messages():
        return [[&quot;hi&quot;], [&quot;hello&quot;]]

    op = DatabricksRunNowOperator.partial(
        task_id = &quot;new_job&quot;,
        job_id=42,
        notebook_params={&quot;dry-run&quot;: &quot;true&quot;},
        python_params=[&quot;douglas adams&quot;, &quot;42&quot;],
        spark_submit_params=[&quot;--class&quot;, &quot;org.apache.spark.examples.SparkPi&quot;]
    ).expand(jar_params=pull_messages())

    # py_job = PythonOperator.partial(
    #     task_id = 'py_job',
    #     python_callable=random
    # ).expand(op_args= pull_messages())


tutorial_etl_dag = tutorial_taskflow_api_etl()"><pre class="notranslate"><code class="notranslate">import json
import pendulum
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator

@dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=['example'],
)
def tutorial_taskflow_api_etl():
    def random(*args, **kwargs):
        print ("==== kwargs inside random ====", args, kwargs)
        print ("I'm random")
        return 49

    @task
    def pull_messages():
        return [["hi"], ["hello"]]

    op = DatabricksRunNowOperator.partial(
        task_id = "new_job",
        job_id=42,
        notebook_params={"dry-run": "true"},
        python_params=["douglas adams", "42"],
        spark_submit_params=["--class", "org.apache.spark.examples.SparkPi"]
    ).expand(jar_params=pull_messages())

    # py_job = PythonOperator.partial(
    #     task_id = 'py_job',
    #     python_callable=random
    # ).expand(op_args= pull_messages())


tutorial_etl_dag = tutorial_taskflow_api_etl()
</code></pre></div>
<h3 dir="auto">Error</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2022-05-11 11:46:30 +0000] [40] [INFO] Worker exiting (pid: 40)
    return f(*args, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 75, in scheduler
    _run_scheduler_job(args=args)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 46, in _run_scheduler_job
    job.run()
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/base_job.py&quot;, line 244, in run
    self._execute()
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 736, in _execute
    self._run_scheduler_loop()
  File &quot;/usr/local/lib/python3.9/site-packages/astronomer/airflow/version_check/plugin.py&quot;, line 29, in run_before
    fn(*args, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 824, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 906, in _do_scheduling
    callback_to_run = self._schedule_dag_run(dag_run, session)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py&quot;, line 1148, in _schedule_dag_run
    schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 522, in update_state
    info = self.task_instance_scheduling_decisions(session)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 658, in task_instance_scheduling_decisions
    schedulable_tis, changed_tis, expansion_happened = self._get_ready_tis(
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 714, in _get_ready_tis
    expanded_tis, _ = schedulable.task.expand_mapped_task(self.run_id, session=session)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/models/mappedoperator.py&quot;, line 609, in expand_mapped_task
    operator.mul, self._resolve_map_lengths(run_id, session=session).values()
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/models/mappedoperator.py&quot;, line 595, in _resolve_map_lengths
    raise RuntimeError(f&quot;Failed to populate all mapping metadata; missing: {keys}&quot;)
RuntimeError: Failed to populate all mapping metadata; missing: 'jar_params'
[2022-05-11 11:46:30 +0000] [31] [INFO] Shutting down: Master"><pre class="notranslate"><code class="notranslate">[2022-05-11 11:46:30 +0000] [40] [INFO] Worker exiting (pid: 40)
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 75, in scheduler
    _run_scheduler_job(args=args)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/scheduler_command.py", line 46, in _run_scheduler_job
    job.run()
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/base_job.py", line 244, in run
    self._execute()
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 736, in _execute
    self._run_scheduler_loop()
  File "/usr/local/lib/python3.9/site-packages/astronomer/airflow/version_check/plugin.py", line 29, in run_before
    fn(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 824, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 906, in _do_scheduling
    callback_to_run = self._schedule_dag_run(dag_run, session)
  File "/usr/local/lib/python3.9/site-packages/airflow/jobs/scheduler_job.py", line 1148, in _schedule_dag_run
    schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py", line 522, in update_state
    info = self.task_instance_scheduling_decisions(session)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py", line 658, in task_instance_scheduling_decisions
    schedulable_tis, changed_tis, expansion_happened = self._get_ready_tis(
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagrun.py", line 714, in _get_ready_tis
    expanded_tis, _ = schedulable.task.expand_mapped_task(self.run_id, session=session)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/mappedoperator.py", line 609, in expand_mapped_task
    operator.mul, self._resolve_map_lengths(run_id, session=session).values()
  File "/usr/local/lib/python3.9/site-packages/airflow/models/mappedoperator.py", line 595, in _resolve_map_lengths
    raise RuntimeError(f"Failed to populate all mapping metadata; missing: {keys}")
RuntimeError: Failed to populate all mapping metadata; missing: 'jar_params'
[2022-05-11 11:46:30 +0000] [31] [INFO] Shutting down: Master
</code></pre></div>
<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"><em>No response</em></p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 10 (buster)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-databricks</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Astronomer</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"> 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>