Change Fivetran Airflow operators to run synchronously so the task concurrency limit is respected. (#5144)

bigquery_etl/query_scheduling/templates/airflow_dag.j2

@@ -159,6 +159,8 @@ with DAG('{{ name }}', default_args=default_args{%+ if schedule_interval != None
     {{ fivetran_task.task_id }}_sync_start = FivetranOperator(
         connector_id='{% raw %}{{{% endraw %} var.value.{{ fivetran_task.task_id }}_connector_id {% raw %}}}{% endraw %}',
         task_id='{{ fivetran_task.task_id }}_task',
+        {# Have the operator run synchronously so the task concurrency limit is respected. -#}
+        deferrable=False,
         task_concurrency=1,
     )
 

tests/data/dags/simple_test_dag

@@ -51,12 +51,14 @@ with DAG(
     fivetran_import_1_sync_start = FivetranOperator(
         connector_id="{{ var.value.fivetran_import_1_connector_id }}",
         task_id="fivetran_import_1_task",
+        deferrable=False,
         task_concurrency=1,
     )
 
     fivetran_import_2_sync_start = FivetranOperator(
         connector_id="{{ var.value.fivetran_import_2_connector_id }}",
         task_id="fivetran_import_2_task",
+        deferrable=False,
         task_concurrency=1,
     )