Add deprecation warning re unset namespace in k8s hook
diff --git a/airflow/providers/cncf/kubernetes/CHANGELOG.rst b/airflow/providers/cncf/kubernetes/CHANGELOG.rst
index 6a0c5edb2962d..b379dc45be43f 100644
--- a/airflow/providers/cncf/kubernetes/CHANGELOG.rst
+++ b/airflow/providers/cncf/kubernetes/CHANGELOG.rst
@@ -39,6 +39,13 @@ Features
 * KubernetsPodOperator argument ``namespace`` is now optional.  If not supplied via KPO param or pod template file or full pod spec, then we'll check the airflow conn,
   then if in a k8s pod, try to infer the namespace from the container, then finally will use the ``default`` namespace.
 
+Deprecations
+~~~~~~~~~~~~
+
+* In ``KubernetesHook.get_namespace``, if a connection is defined but a namespace isn't set, we currently return 'default'; this behavior is deprecated. In the next release, we'll return ``None``.
+
+
+
 4.4.0
 .....
 
diff --git a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index b22cf58085004..27064e6b5abe3 100644
--- a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -72,6 +72,8 @@ class KubernetesHook(BaseHook):
     conn_type = "kubernetes"
     hook_name = "Kubernetes Cluster Connection"
 
+    DEFAULT_NAMESPACE = "default"
+
     @staticmethod
     def get_connection_form_widgets() -> dict[str, Any]:
         """Returns connection widgets to add to connection form"""
@@ -268,8 +270,7 @@ def create_custom_object(
         :param namespace: kubernetes namespace
         """
         api = client.CustomObjectsApi(self.api_client)
-        if namespace is None:
-            namespace = self.get_namespace()
+        namespace = namespace or self._get_namespace() or self.DEFAULT_NAMESPACE
         if isinstance(body, str):
             body_dict = _load_body_to_dict(body)
         else:
@@ -308,8 +309,7 @@ def get_custom_object(
         :param namespace: kubernetes namespace
         """
         api = client.CustomObjectsApi(self.api_client)
-        if namespace is None:
-            namespace = self.get_namespace()
+        namespace = namespace or self._get_namespace() or self.DEFAULT_NAMESPACE
         try:
             response = api.get_namespaced_custom_object(
                 group=group, version=version, namespace=namespace, plural=plural, name=name
@@ -319,9 +319,32 @@ def get_custom_object(
             raise AirflowException(f"Exception when calling -> get_custom_object: {e}\n")
 
     def get_namespace(self) -> str | None:
-        """Returns the namespace that defined in the connection"""
+        """
+        Returns the namespace defined in the connection or 'default'.
+
+        TODO: in provider version 6.0, return None when namespace not defined in connection
+        """
+        namespace = self._get_namespace()
+        if self.conn_id and not namespace:
+            warnings.warn(
+                "Airflow connection defined but namespace is not set; returning 'default'.  In "
+                "cncf.kubernetes provider version 6.0 we will return None when namespace is "
+                "not defined in the connection so that it's clear whether user intends 'default' or "
+                "whether namespace is unset (which is required in order to apply precedence logic in "
+                "KubernetesPodOperator).",
+                DeprecationWarning,
+            )
+            return "default"
+        return namespace
+
+    def _get_namespace(self) -> str | None:
+        """
+        Returns the namespace that defined in the connection
+
+        TODO: in provider version 6.0, get rid of this method and make it the behavior of get_namespace.
+        """
         if self.conn_id:
-            return self._get_field("namespace") or "default"
+            return self._get_field("namespace")
         return None
 
     def get_pod_log_stream(
@@ -344,7 +367,7 @@ def get_pod_log_stream(
                 self.core_v1_client.read_namespaced_pod_log,
                 name=pod_name,
                 container=container,
-                namespace=namespace if namespace else self.get_namespace(),
+                namespace=namespace or self._get_namespace() or self.DEFAULT_NAMESPACE,
             ),
         )
 
@@ -365,7 +388,7 @@ def get_pod_logs(
             name=pod_name,
             container=container,
             _preload_content=False,
-            namespace=namespace if namespace else self.get_namespace(),
+            namespace=namespace or self._get_namespace() or self.DEFAULT_NAMESPACE,
         )
 
 
diff --git a/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
index c8bbef106fc80..e15648a5e9816 100644
--- a/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
+++ b/tests/providers/cncf/kubernetes/hooks/test_kubernetes.py
@@ -32,11 +32,16 @@
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
 from airflow.utils import db
 from tests.test_utils.db import clear_db_connections
+from tests.test_utils.providers import get_provider_min_airflow_version
 
 KUBE_CONFIG_PATH = os.getenv("KUBECONFIG", "~/.kube/config")
 HOOK_MODULE = "airflow.providers.cncf.kubernetes.hooks.kubernetes"
 
 
+class DeprecationRemovalRequired(AirflowException):
+    ...
+
+
 class TestKubernetesHook:
     @classmethod
     def setup_class(cls) -> None:
@@ -304,6 +309,12 @@ def test_default_kube_config_connection(self, mock_kube_config_merger, mock_kube
     def test_get_namespace(self, conn_id, expected):
         hook = KubernetesHook(conn_id=conn_id)
         assert hook.get_namespace() == expected
+        if get_provider_min_airflow_version("apache-airflow-providers-cncf-kubernetes") >= (6, 0):
+            raise DeprecationRemovalRequired(
+                "You must update get_namespace so that if namespace not set "
+                "in the connection, then None is returned. To do so, remove get_namespace "
+                "and rename _get_namespace to get_namespace."
+            )
 
     @patch("kubernetes.config.kube_config.KubeConfigLoader")
     @patch("kubernetes.config.kube_config.KubeConfigMerger")

Currently when k8s conn defined but namespace not set, we return 'default'.  This is not good behavior because e.g. in KPO we aren't able to tell whether user explicitly set namespace or whether it's just inserted as a fallback -- something that's important for applying precedence.

So we deprecate the return of 'default'.  In next major, we'll just return None.
