Repository: apache/airflow
Pull request: Local kubernetes executor
LocalKubernetesExecutor similar to CeleryKubernetesExecutor.
Tested by running webserver, scheduler and connecting to minikube k8s cluster.
Note: This PR does not update the helm charts.

closes: #16185
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.

Feel free to ping committers for the review!

In case of existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).


Problem statement:
Combined Executor - KubernetesExecutor + LocalExecutor
**Description**
In short: There exists already combined executor support for Celery + Kubernetes, I think that also adding Local + Kubernetes
would be warmly welcomed.

**Use case / motivation**

Many tasks are low-resource intensive tasks on the airflow pod, It makes sense to keep things running on LocalExecutor, 0 overhead. However I also have a large number of dags, that have some long running tasks which are somewhat resource hungry, for these I currently use KubernetesPodOperator, this makes the Dags more scalable and overall workload elastic. Being able to easily switch between easily between KubernetesExecutor and LocalExecutor, just feels natural in this sense.


**Related Issues**

N/A
@potiuk

Hints from issue discussion:
Thanks for opening your first issue here! Be sure to follow the issue template!

Worth looking at creating it !
@potiuk  Can I grab this issue? If yes, where I can start looking?
Take a look at ``CeleryKubernetesExecutor``. Although this one would require a little more intrinsic knowledge of Airflow - so please take a little more knowledge of Airflow  - like integration with Helm Chart etc. Maybe marking it as ``first-good-issue`` was not the best idea :) . Just take a look at the C-K executor and see if this is something you would like to take on.
> Take a look at `CeleryKubernetesExecutor`. Although this one would require a little more intrinsic knowledge of Airflow - so please take a little more knowledge of Airflow - like integration with Helm Chart etc. Maybe marking it as `first-good-issue` was not the best idea :) . Just take a look at the C-K executor and see if this is something you would like to take on.

Is this being worked on, Im trying to find an example DAG of how to mark the tasks as kubernetes_queue, would be helpful if you can point to it - @potiuk 
Actually @potiuk can u assign this to me please, if no one else is working.
Feel free!

Commit message: [16185] Added LocalKubernetesExecutor to breeze supported executors

Patch to review:
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index 0e56875738a5c..253f5bcf3fa22 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -1446,6 +1446,21 @@
       type: string
       example: ~
       default: ~
+- name: local_kubernetes_executor
+  description: |
+    This section only applies if you are using the ``LocalKubernetesExecutor`` in
+    ``[core]`` section above
+  options:
+    - name: kubernetes_queue
+      description: |
+        Define when to send a task to ``KubernetesExecutor`` when using ``LocalKubernetesExecutor``.
+        When the queue of a task is the value of ``kubernetes_queue`` (default ``kubernetes``),
+        the task is executed via ``KubernetesExecutor``,
+        otherwise via ``LocalExecutor``
+      version_added: 2.2.0
+      type: string
+      example: ~
+      default: "kubernetes"
 - name: celery_kubernetes_executor
   description: |
     This section only applies if you are using the ``CeleryKubernetesExecutor`` in
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index f733c2db0ef9b..d47226c0d24c7 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -714,6 +714,16 @@ sentry_dsn =
 # Dotted path to a before_send function that the sentry SDK should be configured to use.
 # before_send =
 
+[local_kubernetes_executor]
+
+# This section only applies if you are using the ``LocalKubernetesExecutor`` in
+# ``[core]`` section above
+# Define when to send a task to ``KubernetesExecutor`` when using ``LocalKubernetesExecutor``.
+# When the queue of a task is the value of ``kubernetes_queue`` (default ``kubernetes``),
+# the task is executed via ``KubernetesExecutor``,
+# otherwise via ``LocalExecutor``
+kubernetes_queue = kubernetes
+
 [celery_kubernetes_executor]
 
 # This section only applies if you are using the ``CeleryKubernetesExecutor`` in
diff --git a/airflow/example_dags/example_local_kubernetes_executor.py b/airflow/example_dags/example_local_kubernetes_executor.py
new file mode 100644
index 0000000000000..3e86de3bd25e2
--- /dev/null
+++ b/airflow/example_dags/example_local_kubernetes_executor.py
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+This is an example dag for using a Kubernetes Executor Configuration.
+"""
+import logging
+from datetime import datetime
+
+from airflow import DAG
+from airflow.configuration import conf
+from airflow.example_dags.libs.helper import print_stuff
+from airflow.operators.python_operator import PythonOperator
+
+log = logging.getLogger(__name__)
+
+worker_container_repository = conf.get('kubernetes', 'worker_container_repository')
+worker_container_tag = conf.get('kubernetes', 'worker_container_tag')
+
+try:
+    from kubernetes.client import models as k8s
+except ImportError:
+    log.warning("Could not import DAGs in example_kubernetes_executor.py", exc_info=True)
+    log.warning("Install Kubernetes dependencies with: pip install apache-airflow[cncf.kubernetes]")
+
+with DAG(
+    dag_id='example_local_kubernetes_executor',
+    schedule_interval=None,
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
+    tags=['example3'],
+) as dag:
+    # You can use annotations on your kubernetes pods!
+    start_task_executor_config = {
+        "pod_override": k8s.V1Pod(metadata=k8s.V1ObjectMeta(annotations={"test": "annotation"}))
+    }
+
+    task_with_template = PythonOperator(
+        task_id="task_with_kubernetes_executor",
+        python_callable=print_stuff,
+        pool='kubernetes',
+        queue='kubernetes',
+        executor_config=start_task_executor_config,
+    )
+
+    def print_context(ds, **kwargs):
+        """Print the Airflow context and ds variable from the context."""
+        print(kwargs)
+        print(ds)
+        return 'Whatever you return gets printed in the logs'
+
+    task_with_local = PythonOperator(
+        task_id="task_with_local_executor",
+        python_callable=print_context,
+    )
+
+    task_with_local >> task_with_template
diff --git a/airflow/executors/executor_constants.py b/airflow/executors/executor_constants.py
index fed8e27b839bc..55a3a7f766301 100644
--- a/airflow/executors/executor_constants.py
+++ b/airflow/executors/executor_constants.py
@@ -16,6 +16,7 @@
 # under the License.
 
 LOCAL_EXECUTOR = "LocalExecutor"
+LOCAL_KUBERNETES_EXECUTOR = "LocalKubernetesExecutor"
 SEQUENTIAL_EXECUTOR = "SequentialExecutor"
 CELERY_EXECUTOR = "CeleryExecutor"
 CELERY_KUBERNETES_EXECUTOR = "CeleryKubernetesExecutor"
diff --git a/airflow/executors/executor_loader.py b/airflow/executors/executor_loader.py
index 676bf3582c98a..b0cb182453502 100644
--- a/airflow/executors/executor_loader.py
+++ b/airflow/executors/executor_loader.py
@@ -28,6 +28,7 @@
     DEBUG_EXECUTOR,
     KUBERNETES_EXECUTOR,
     LOCAL_EXECUTOR,
+    LOCAL_KUBERNETES_EXECUTOR,
     SEQUENTIAL_EXECUTOR,
 )
 from airflow.utils.module_loading import import_string
@@ -41,6 +42,7 @@ class ExecutorLoader:
     _default_executor: Optional[BaseExecutor] = None
     executors = {
         LOCAL_EXECUTOR: 'airflow.executors.local_executor.LocalExecutor',
+        LOCAL_KUBERNETES_EXECUTOR: 'airflow.executors.local_kubernetes_executor.LocalKubernetesExecutor',
         SEQUENTIAL_EXECUTOR: 'airflow.executors.sequential_executor.SequentialExecutor',
         CELERY_EXECUTOR: 'airflow.executors.celery_executor.CeleryExecutor',
         CELERY_KUBERNETES_EXECUTOR: 'airflow.executors.celery_kubernetes_executor.CeleryKubernetesExecutor',
@@ -77,6 +79,8 @@ def load_executor(cls, executor_name: str) -> BaseExecutor:
         """
         if executor_name == CELERY_KUBERNETES_EXECUTOR:
             return cls.__load_celery_kubernetes_executor()
+        elif executor_name == LOCAL_KUBERNETES_EXECUTOR:
+            return cls.__load_local_kubernetes_executor()
 
         if executor_name in cls.executors:
             log.debug("Loading core executor: %s", executor_name)
@@ -118,6 +122,15 @@ def __load_celery_kubernetes_executor(cls) -> BaseExecutor:
         celery_kubernetes_executor_cls = import_string(cls.executors[CELERY_KUBERNETES_EXECUTOR])
         return celery_kubernetes_executor_cls(celery_executor, kubernetes_executor)
 
+    @classmethod
+    def __load_local_kubernetes_executor(cls) -> BaseExecutor:
+        """:return: an instance of LocalKubernetesExecutor"""
+        local_executor = import_string(cls.executors[LOCAL_EXECUTOR])()
+        kubernetes_executor = import_string(cls.executors[KUBERNETES_EXECUTOR])()
+
+        local_kubernetes_executor_cls = import_string(cls.executors[LOCAL_KUBERNETES_EXECUTOR])
+        return local_kubernetes_executor_cls(local_executor, kubernetes_executor)
+
 
 UNPICKLEABLE_EXECUTORS = (
     LOCAL_EXECUTOR,
diff --git a/airflow/executors/local_kubernetes_executor.py b/airflow/executors/local_kubernetes_executor.py
new file mode 100644
index 0000000000000..4fab2d1efa913
--- /dev/null
+++ b/airflow/executors/local_kubernetes_executor.py
@@ -0,0 +1,204 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Dict, List, Optional, Set, Union
+
+from airflow.configuration import conf
+from airflow.executors.base_executor import CommandType, EventBufferValueType, QueuedTaskInstanceType
+from airflow.executors.kubernetes_executor import KubernetesExecutor
+from airflow.executors.local_executor import LocalExecutor
+from airflow.models.taskinstance import SimpleTaskInstance, TaskInstance, TaskInstanceKey
+from airflow.utils.log.logging_mixin import LoggingMixin
+
+
+class LocalKubernetesExecutor(LoggingMixin):
+    """
+    LocalKubernetesExecutor consists of LocalExecutor and KubernetesExecutor.
+    It chooses an executor to use based on the queue defined on the task.
+    When the queue is the value of ``kubernetes_queue`` in section ``[local_kubernetes_executor]``
+    of the configuration (default value: `kubernetes`), KubernetesExecutor is selected to run the task,
+    otherwise, LocalExecutor is used.
+    """
+
+    supports_ad_hoc_ti_run: bool = True
+
+    KUBERNETES_QUEUE = conf.get('local_kubernetes_executor', 'kubernetes_queue')
+
+    def __init__(self, local_executor: LocalExecutor, kubernetes_executor: KubernetesExecutor):
+        super().__init__()
+        self._job_id: Optional[str] = None
+        self.local_executor = local_executor
+        self.kubernetes_executor = kubernetes_executor
+
+    @property
+    def queued_tasks(self) -> Dict[TaskInstanceKey, QueuedTaskInstanceType]:
+        """Return queued tasks from local and kubernetes executor"""
+        queued_tasks = self.local_executor.queued_tasks.copy()
+        queued_tasks.update(self.kubernetes_executor.queued_tasks)
+
+        return queued_tasks
+
+    @property
+    def running(self) -> Set[TaskInstanceKey]:
+        """Return running tasks from local and kubernetes executor"""
+        return self.local_executor.running.union(self.kubernetes_executor.running)
+
+    @property
+    def job_id(self) -> Optional[str]:
+        """
+        This is a class attribute in BaseExecutor but since this is not really an executor, but a wrapper
+        of executors we implement as property so we can have custom setter.
+        """
+        return self._job_id
+
+    @job_id.setter
+    def job_id(self, value: Optional[str]) -> None:
+        """job_id is manipulated by SchedulerJob.  We must propagate the job_id to wrapped executors."""
+        self._job_id = value
+        self.kubernetes_executor.job_id = value
+        self.local_executor.job_id = value
+
+    def start(self) -> None:
+        self.log.info("Starting local and Kubernetes Executor")
+        """Start local and kubernetes executor"""
+        self.local_executor.start()
+        self.kubernetes_executor.start()
+
+    @property
+    def slots_available(self) -> int:
+        """Number of new tasks this executor instance can accept"""
+        return self.local_executor.slots_available
+
+    def queue_command(
+        self,
+        task_instance: TaskInstance,
+        command: CommandType,
+        priority: int = 1,
+        queue: Optional[str] = None,
+    ) -> None:
+        """Queues command via local or kubernetes executor"""
+        executor = self._router(task_instance)
+        self.log.debug("Using executor: %s for %s", executor.__class__.__name__, task_instance.key)
+        executor.queue_command(task_instance, command, priority, queue)
+
+    def queue_task_instance(
+        self,
+        task_instance: TaskInstance,
+        mark_success: bool = False,
+        pickle_id: Optional[str] = None,
+        ignore_all_deps: bool = False,
+        ignore_depends_on_past: bool = False,
+        ignore_task_deps: bool = False,
+        ignore_ti_state: bool = False,
+        pool: Optional[str] = None,
+        cfg_path: Optional[str] = None,
+    ) -> None:
+        """Queues task instance via local or kubernetes executor"""
+        executor = self._router(SimpleTaskInstance(task_instance))
+        self.log.debug(
+            "Using executor: %s to queue_task_instance for %s", executor.__class__.__name__, task_instance.key
+        )
+        executor.queue_task_instance(
+            task_instance,
+            mark_success,
+            pickle_id,
+            ignore_all_deps,
+            ignore_depends_on_past,
+            ignore_task_deps,
+            ignore_ti_state,
+            pool,
+            cfg_path,
+        )
+
+    def has_task(self, task_instance: TaskInstance) -> bool:
+        """
+        Checks if a task is either queued or running in either local or kubernetes executor.
+
+        :param task_instance: TaskInstance
+        :return: True if the task is known to this executor
+        """
+        return self.local_executor.has_task(task_instance) or self.kubernetes_executor.has_task(task_instance)
+
+    def heartbeat(self) -> None:
+        """Heartbeat sent to trigger new jobs in local and kubernetes executor"""
+        self.local_executor.heartbeat()
+        self.kubernetes_executor.heartbeat()
+
+    def get_event_buffer(
+        self, dag_ids: Optional[List[str]] = None
+    ) -> Dict[TaskInstanceKey, EventBufferValueType]:
+        """
+        Returns and flush the event buffer from local and kubernetes executor
+
+        :param dag_ids: dag_ids to return events for, if None returns all
+        :return: a dict of events
+        """
+        cleared_events_from_local = self.local_executor.get_event_buffer(dag_ids)
+        cleared_events_from_kubernetes = self.kubernetes_executor.get_event_buffer(dag_ids)
+
+        return {**cleared_events_from_local, **cleared_events_from_kubernetes}
+
+    def try_adopt_task_instances(self, tis: List[TaskInstance]) -> List[TaskInstance]:
+        """
+        Try to adopt running task instances that have been abandoned by a SchedulerJob dying.
+
+        Anything that is not adopted will be cleared by the scheduler (and then become eligible for
+        re-scheduling)
+
+        :return: any TaskInstances that were unable to be adopted
+        :rtype: list[airflow.models.TaskInstance]
+        """
+        local_tis = []
+        kubernetes_tis = []
+        abandoned_tis = []
+        for ti in tis:
+            if ti.queue == self.KUBERNETES_QUEUE:
+                kubernetes_tis.append(ti)
+            else:
+                local_tis.append(ti)
+        abandoned_tis.extend(self.local_executor.try_adopt_task_instances(local_tis))
+        abandoned_tis.extend(self.kubernetes_executor.try_adopt_task_instances(kubernetes_tis))
+        return abandoned_tis
+
+    def end(self) -> None:
+        """End local and kubernetes executor"""
+        self.local_executor.end()
+        self.kubernetes_executor.end()
+
+    def terminate(self) -> None:
+        """Terminate local and kubernetes executor"""
+        self.local_executor.terminate()
+        self.kubernetes_executor.terminate()
+
+    def _router(self, simple_task_instance: SimpleTaskInstance) -> Union[LocalExecutor, KubernetesExecutor]:
+        """
+        Return either local_executor or kubernetes_executor
+
+        :param simple_task_instance: SimpleTaskInstance
+        :return: local_executor or kubernetes_executor
+        :rtype: Union[LocalExecutor, KubernetesExecutor]
+        """
+        if simple_task_instance.queue == self.KUBERNETES_QUEUE:
+            return self.kubernetes_executor
+        return self.local_executor
+
+    def debug_dump(self) -> None:
+        """Called in response to SIGUSR2 by the scheduler"""
+        self.log.info("Dumping LocalExecutor state")
+        self.local_executor.debug_dump()
+        self.log.info("Dumping KubernetesExecutor state")
+        self.kubernetes_executor.debug_dump()
diff --git a/airflow/settings.py b/airflow/settings.py
index 2fb25f39df37d..51d17aa16f14b 100644
--- a/airflow/settings.py
+++ b/airflow/settings.py
@@ -540,6 +540,7 @@ def initialize():
 IS_K8S_OR_K8SCELERY_EXECUTOR = conf.get('core', 'EXECUTOR') in {
     executor_constants.KUBERNETES_EXECUTOR,
     executor_constants.CELERY_KUBERNETES_EXECUTOR,
+    executor_constants.LOCAL_KUBERNETES_EXECUTOR,
 }
 
 HIDE_SENSITIVE_VAR_CONN_FIELDS = conf.getboolean('core', 'hide_sensitive_var_conn_fields')
diff --git a/breeze-complete b/breeze-complete
index d03c5767819f6..7f24693811bd1 100644
--- a/breeze-complete
+++ b/breeze-complete
@@ -35,7 +35,7 @@ _breeze_allowed_mysql_versions="5.7 8"
 _breeze_allowed_mssql_versions="2017-latest 2019-latest"
 _breeze_allowed_postgres_versions="9.6 10 11 12 13"
 _breeze_allowed_kind_operations="start stop restart status deploy test shell k9s"
-_breeze_allowed_executors="KubernetesExecutor CeleryExecutor LocalExecutor CeleryKubernetesExecutor"
+_breeze_allowed_executors="KubernetesExecutor CeleryExecutor LocalExecutor CeleryKubernetesExecutor LocalKubernetesExecutor"
 _breeze_allowed_test_types="All Always Core Providers API CLI Integration Other WWW Postgres MySQL Helm Quarantined"
 _breeze_allowed_package_formats="both sdist wheel"
 _breeze_allowed_installation_methods=". apache-airflow"
diff --git a/docs/apache-airflow/executor/local_kubernetes.rst b/docs/apache-airflow/executor/local_kubernetes.rst
new file mode 100644
index 0000000000000..281620e910ab7
--- /dev/null
+++ b/docs/apache-airflow/executor/local_kubernetes.rst
@@ -0,0 +1,30 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+.. _executor:LocalKubernetesExecutor:
+
+LocalKubernetes Executor
+=========================
+
+The :class:`~airflow.executors.local_kubernetes_executor.LocalKubernetesExecutor` allows users
+to run simultaneously a ``LocalExecutor`` and a ``KubernetesExecutor``.
+An executor is chosen to run a task based on the task's queue.
+
+``LocalKubernetesExecutor`` provides the capability of running low intesive tasks locally
+using ``LocalExecutor`` and to handle the high load at the peak time and runtime isolation
+of the ``KubernetesExecutor``.
diff --git a/tests/core/test_config_templates.py b/tests/core/test_config_templates.py
index d2e26237de3ae..e69f38f75b215 100644
--- a/tests/core/test_config_templates.py
+++ b/tests/core/test_config_templates.py
@@ -40,6 +40,7 @@
     'email',
     'smtp',
     'sentry',
+    'local_kubernetes_executor',
     'celery_kubernetes_executor',
     'celery',
     'celery_broker_transport_options',
diff --git a/tests/executors/test_local_kubernetes_executor.py b/tests/executors/test_local_kubernetes_executor.py
new file mode 100644
index 0000000000000..eab4e2f0128e3
--- /dev/null
+++ b/tests/executors/test_local_kubernetes_executor.py
@@ -0,0 +1,64 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from unittest import mock
+
+from airflow.configuration import conf
+from airflow.executors.local_executor import LocalExecutor
+from airflow.executors.local_kubernetes_executor import LocalKubernetesExecutor
+
+KUBERNETES_QUEUE = LocalKubernetesExecutor.KUBERNETES_QUEUE
+
+
+class TestLocalKubernetesExecutor:
+    def test_queued_tasks(self):
+        local_executor_mock = mock.MagicMock()
+        k8s_executor_mock = mock.MagicMock()
+        local_kubernetes_executor = LocalKubernetesExecutor(local_executor_mock, k8s_executor_mock)
+
+        local_queued_tasks = {('dag_id', 'task_id', '2020-08-30', 1): 'queued_command'}
+        k8s_queued_tasks = {('dag_id_2', 'task_id_2', '2020-08-30', 2): 'queued_command'}
+
+        local_executor_mock.queued_tasks = local_queued_tasks
+        k8s_executor_mock.queued_tasks = k8s_queued_tasks
+
+        expected_queued_tasks = {**local_queued_tasks, **k8s_queued_tasks}
+
+        assert local_kubernetes_executor.queued_tasks == expected_queued_tasks
+        assert len(local_kubernetes_executor.queued_tasks) == 2
+
+    def test_running(self):
+        local_executor_mock = mock.MagicMock()
+        k8s_executor_mock = mock.MagicMock()
+        local_kubernetes_executor = LocalKubernetesExecutor(local_executor_mock, k8s_executor_mock)
+
+        local_running_tasks = {('dag_id', 'task_id', '2020-08-30', 1)}
+        k8s_running_tasks = {}
+
+        local_executor_mock.running = local_running_tasks
+        k8s_executor_mock.running = k8s_running_tasks
+
+        assert local_kubernetes_executor.running == local_running_tasks.union(k8s_running_tasks)
+        assert len(local_kubernetes_executor.running) == 1
+
+    def test_slots_available(self):
+        local_executor = LocalExecutor()
+        k8s_executor_mock = mock.MagicMock()
+        local_kubernetes_executor = LocalKubernetesExecutor(local_executor, k8s_executor_mock)
+
+        # Should be equal to Local Executor default parallelism.
+        assert local_kubernetes_executor.slots_available == conf.getint('core', 'PARALLELISM')


Code review comments:
diff_hunk: @@ -0,0 +1,30 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+.. _executor:LocalKubernetesExecutor:
+
+LocalKubernetes Executor
+=========================
+
+The :class:`~airflow.executors.local_kubernetes_executor.LocalKubernetesExecutor` allows users
+to run simultaneously a ``LocalExecutor`` and a ``KubernetesExecutor``.
+An executor is chosen to run a task based on the task's queue.
+
+``LocalKubernetesExecutor`` provides the capability of running low intesive tasks locally
path: docs/apache-airflow/executor/local_kubernetes.rst
text: @user1:
```suggestion
``LocalKubernetesExecutor`` provides the capability of running low-intensive tasks locally
```

diff_hunk: @@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+This is an example dag for using a Kubernetes Executor Configuration.
path: airflow/example_dags/example_local_kubernetes_executor.py
text: @user1:
I guess it's a copy-paste error here: `Kubernetes Executor Configuration` -> `LocalKubernetesExecutor ...`

@author:
Updated.

diff_hunk: @@ -1446,6 +1446,21 @@
       type: string
       example: ~
       default: ~
+- name: local_kubernetes_executor
+  description: |
+    This section only applies if you are using the ``LocalKubernetesExecutor`` in
+    ``[core]`` section above
+  options:
+    - name: kubernetes_queue
+      description: |
+        Define when to send a task to ``KubernetesExecutor`` when using ``LocalKubernetesExecutor``.
+        When the queue of a task is the value of ``kubernetes_queue`` (default ``kubernetes``),
+        the task is executed via ``KubernetesExecutor``,
+        otherwise via ``LocalExecutor``
+      version_added: 2.2.0
path: airflow/config_templates/config.yml
text: @user2:
I saw that @user1 has added milestone `2.3.0` for this, so let's update this line here as well?

@author:
Updated, thanks

diff_hunk: @@ -0,0 +1,30 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+.. _executor:LocalKubernetesExecutor:
+
+LocalKubernetes Executor
+=========================
+
+The :class:`~airflow.executors.local_kubernetes_executor.LocalKubernetesExecutor` allows users
+to run simultaneously a ``LocalExecutor`` and a ``KubernetesExecutor``.
path: docs/apache-airflow/executor/local_kubernetes.rst
text: @user1:
```suggestion
to simultaneously run a ``LocalExecutor`` and a ``KubernetesExecutor``.
```

diff_hunk: @@ -0,0 +1,64 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from unittest import mock
+
+from airflow.configuration import conf
+from airflow.executors.local_executor import LocalExecutor
+from airflow.executors.local_kubernetes_executor import LocalKubernetesExecutor
+
+KUBERNETES_QUEUE = LocalKubernetesExecutor.KUBERNETES_QUEUE
path: tests/executors/test_local_kubernetes_executor.py
text: @user1:
How is this variable being used?

@author:
Removed.

diff_hunk: @@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+This is an example dag for using a Kubernetes Executor Configuration.
+"""
+import logging
+from datetime import datetime
+
+from airflow import DAG
+from airflow.configuration import conf
+from airflow.example_dags.libs.helper import print_stuff
+from airflow.operators.python_operator import PythonOperator
+
+log = logging.getLogger(__name__)
+
+worker_container_repository = conf.get('kubernetes', 'worker_container_repository')
+worker_container_tag = conf.get('kubernetes', 'worker_container_tag')
+
+try:
+    from kubernetes.client import models as k8s
+except ImportError:
+    log.warning("Could not import DAGs in example_kubernetes_executor.py", exc_info=True)
+    log.warning("Install Kubernetes dependencies with: pip install apache-airflow[cncf.kubernetes]")
+
+with DAG(
+    dag_id='example_local_kubernetes_executor',
+    schedule_interval=None,
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
+    tags=['example3'],
+) as dag:
+    # You can use annotations on your kubernetes pods!
+    start_task_executor_config = {
+        "pod_override": k8s.V1Pod(metadata=k8s.V1ObjectMeta(annotations={"test": "annotation"}))
+    }
+
+    task_with_template = PythonOperator(
+        task_id="task_with_kubernetes_executor",
+        python_callable=print_stuff,
+        pool='kubernetes',
path: airflow/example_dags/example_local_kubernetes_executor.py
text: @user1:
May you please clarify why we need to specify `pool` here? If it's not necessary, we can remove it?

@author:
I was just trying something, will remove it.

@author:
Removed pool.

diff_hunk: @@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+This is an example dag for using a Kubernetes Executor Configuration.
+"""
+import logging
+from datetime import datetime
+
+from airflow import DAG
+from airflow.configuration import conf
+from airflow.example_dags.libs.helper import print_stuff
+from airflow.operators.python_operator import PythonOperator
+
+log = logging.getLogger(__name__)
+
+worker_container_repository = conf.get('kubernetes', 'worker_container_repository')
+worker_container_tag = conf.get('kubernetes', 'worker_container_tag')
+
+try:
+    from kubernetes.client import models as k8s
+except ImportError:
+    log.warning("Could not import DAGs in example_kubernetes_executor.py", exc_info=True)
+    log.warning("Install Kubernetes dependencies with: pip install apache-airflow[cncf.kubernetes]")
+
+with DAG(
+    dag_id='example_local_kubernetes_executor',
+    schedule_interval=None,
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
+    tags=['example3'],
+) as dag:
+    # You can use annotations on your kubernetes pods!
+    start_task_executor_config = {
+        "pod_override": k8s.V1Pod(metadata=k8s.V1ObjectMeta(annotations={"test": "annotation"}))
+    }
+
+    task_with_template = PythonOperator(
+        task_id="task_with_kubernetes_executor",
+        python_callable=print_stuff,
+        pool='kubernetes',
+        queue='kubernetes',
+        executor_config=start_task_executor_config,
+    )
+
+    def print_context(ds, **kwargs):
+        """Print the Airflow context and ds variable from the context."""
+        print(kwargs)
+        print(ds)
+        return 'Whatever you return gets printed in the logs'
+
+    task_with_local = PythonOperator(
+        task_id="task_with_local_executor",
+        python_callable=print_context,
+    )
path: airflow/example_dags/example_local_kubernetes_executor.py
text: @user1:
Let's use the TaskFlow API in the example DAGs rather than the `PythonOperator`. There was a substantial effort to update the example DAGs in this manner and we should try to keep this consistency.

The existing "example_kubernetes_executor" example DAG has some good pointers for how the TaskFlow API was used in a similar context.

@author:
I couldn't figure out on how to pass the queue='kubernetes', in the @user2 decorator. The example_kubernetes_executor doesn't do it. Let me try again. Thanks 

@user1:
You should be able to pass these types of `BaseOperator` args directly to the decorator like `@user2(task_id=..., pool=..., queue=..., executor_config=...)`.

@author:
Changed to use taskflow api.