[EWT-569] Airflow Upgrade to 1.10.14 [CP] from 1.10.4+twtr : 9b58c88

[CP][EWT-302]Patch Pool.DEFAULT_POOL_NAME in BaseOperator (apache#8587) (twitter-forks#49) Open source commit id: b37ce29 Co-authored-by: Vishesh Jain <[email protected]>
ayushSethi22 · Dec 20, 2020 · 1698a7a · 1698a7a
1 parent c54061a
commit 1698a7a
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 2 deletions.
diff --git a/UPDATING.md b/UPDATING.md
@@ -21,6 +21,7 @@
 This file documents any backwards-incompatible changes in Airflow and
 assists users migrating to a new version.
 
+
 <!-- START doctoc generated TOC please keep comment here to allow auto update -->
 <!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
 **Table of contents**
@@ -196,6 +197,20 @@ Users can now offer a path to a yaml for the KubernetesPodOperator using the `po
 
 Now use NULL as default value for dag.description in dag table
 
+## CP
+
+### Ability to patch Pool.DEFAULT_POOL_NAME in BaseOperator
+It was not possible to patch pool in BaseOperator as the signature sets the default value of pool
+as Pool.DEFAULT_POOL_NAME.
+While using subdagoperator in unittest(without initializing the sqlite db), it was throwing the
+following error:
+```
+sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: slot_pool.
+```
+Fix for this, https://github.com/apache/airflow/pull/8587
+
+## Airflow 1.10.4
+
 ### Restrict editing DagRun State in the old UI (Flask-admin based UI)
 
 Before 1.10.11 it was possible to edit DagRun State in the `/admin/dagrun/` page

diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
@@ -309,7 +309,7 @@ def __init__(
         priority_weight=1,  # type: int
         weight_rule=WeightRule.DOWNSTREAM,  # type: str
         queue=conf.get('celery', 'default_queue'),  # type: str
-        pool=Pool.DEFAULT_POOL_NAME,  # type: str
+        pool=None,  # type: str
         pool_slots=1,  # type: int
         sla=None,  # type: Optional[timedelta]
         execution_timeout=None,  # type: Optional[timedelta]
@@ -380,7 +380,7 @@ def __init__(
         self.retries = retries if retries is not None else \
             int(configuration.conf.get('core', 'default_task_retries', fallback=0))
         self.queue = queue
-        self.pool = pool
+        self.pool = Pool.DEFAULT_POOL_NAME if pool is None else pool
         self.pool_slots = pool_slots
         if self.pool_slots < 1:
             raise AirflowException("pool slots for %s in dag %s cannot be less than 1"