fix: Missing 'slots_occupied' in CeleryKubernetesExecutor and `Loca…

…lKubernetesExecutor` (apache#41602)

Each time the base executor changes, often those need to be copied into
the old (soon to be deprecated) hybrid executors. This was missed for
the new slots_occupied property.

airflow/providers/celery/executors/celery_kubernetes_executor.py

@@ -116,6 +116,11 @@ def slots_available(self) -> int:
         """Number of new tasks this executor instance can accept."""
         return self.celery_executor.slots_available
 
+    @property
+    def slots_occupied(self):
+        """Number of tasks this executor instance is currently managing."""
+        return len(self.running) + len(self.queued_tasks)
+
     def queue_command(
         self,
         task_instance: TaskInstance,

airflow/providers/cncf/kubernetes/executors/local_kubernetes_executor.py

@@ -103,6 +103,11 @@ def slots_available(self) -> int:
         """Number of new tasks this executor instance can accept."""
         return self.local_executor.slots_available
 
+    @property
+    def slots_occupied(self):
+        """Number of tasks this executor instance is currently managing."""
+        return len(self.running) + len(self.queued_tasks)
+
     def queue_command(
         self,
         task_instance: TaskInstance,