Allow hyphens in pod id used by k8s executor
diff --git a/airflow/kubernetes/kubernetes_helper_functions.py b/airflow/kubernetes/kubernetes_helper_functions.py
index f1c0e0fb63dd7..1f0c809cf94bd 100644
--- a/airflow/kubernetes/kubernetes_helper_functions.py
+++ b/airflow/kubernetes/kubernetes_helper_functions.py
@@ -26,22 +26,7 @@
 log = logging.getLogger(__name__)
 
 
-def _strip_unsafe_kubernetes_special_chars(string: str) -> str:
-    """
-    Kubernetes only supports lowercase alphanumeric characters, "-" and "." in
-    the pod name.
-    However, there are special rules about how "-" and "." can be used so let's
-    only keep
-    alphanumeric chars  see here for detail:
-    https://kubernetes.io/docs/concepts/overview/working-with-objects/names/
-
-    :param string: The requested Pod name
-    :return: Pod name stripped of any unsafe characters
-    """
-    return slugify(string, separator="", lowercase=True)
-
-
-def create_pod_id(dag_id: str, task_id: str) -> str:
+def create_pod_id(dag_id: str | None = None, task_id: str | None = None) -> str:
     """
     Generates the kubernetes safe pod_id. Note that this is
     NOT the full ID that will be launched to k8s. We will add a uuid
@@ -51,9 +36,14 @@ def create_pod_id(dag_id: str, task_id: str) -> str:
     :param task_id: Task ID
     :return: The non-unique pod_id for this task/DAG pairing
     """
-    safe_dag_id = _strip_unsafe_kubernetes_special_chars(dag_id)
-    safe_task_id = _strip_unsafe_kubernetes_special_chars(task_id)
-    return safe_dag_id + safe_task_id
+    name = ""
+    if dag_id:
+        name += dag_id
+    if task_id:
+        if name:
+            name += "-"
+        name += task_id
+    return slugify(name, lowercase=True)[:253].strip("-.")
 
 
 def annotations_to_key(annotations: dict[str, str]) -> TaskInstanceKey:
diff --git a/tests/kubernetes/test_kubernetes_helper_functions.py b/tests/kubernetes/test_kubernetes_helper_functions.py
new file mode 100644
index 0000000000000..b90125aeb8fed
--- /dev/null
+++ b/tests/kubernetes/test_kubernetes_helper_functions.py
@@ -0,0 +1,86 @@
+# 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 __future__ import annotations
+
+import re
+
+import pytest
+
+from airflow.kubernetes.kubernetes_helper_functions import create_pod_id
+
+pod_name_regex = r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$"
+
+
+@pytest.mark.parametrize(
+    "val, expected",
+    [
+        ("task-id", "task-id"),  # no problem
+        ("task_id", "task-id"),  # underscores
+        ("---task.id---", "task-id"),  # dots
+        (".task.id", "task-id"),  # leading dot invalid
+        ("**task.id", "task-id"),  # leading dot invalid
+        ("-90Abc*&", "90abc"),  # invalid ends
+        ("90AçLbˆˆç˙ßß˜˜˙c*a", "90aclb-c-ssss-c-a"),  # weird unicode
+    ],
+)
+def test_create_pod_id_task_only(val, expected):
+    actual = create_pod_id(task_id=val)
+    assert actual == expected
+    assert re.match(pod_name_regex, actual)
+
+
+@pytest.mark.parametrize(
+    "val, expected",
+    [
+        ("dag-id", "dag-id"),  # no problem
+        ("dag_id", "dag-id"),  # underscores
+        ("---dag.id---", "dag-id"),  # dots
+        (".dag.id", "dag-id"),  # leading dot invalid
+        ("**dag.id", "dag-id"),  # leading dot invalid
+        ("-90Abc*&", "90abc"),  # invalid ends
+        ("90AçLbˆˆç˙ßß˜˜˙c*a", "90aclb-c-ssss-c-a"),  # weird unicode
+    ],
+)
+def test_create_pod_id_dag_only(val, expected):
+    actual = create_pod_id(dag_id=val)
+    assert actual == expected
+    assert re.match(pod_name_regex, actual)
+
+
+@pytest.mark.parametrize(
+    "dag_id, task_id, expected",
+    [
+        ("dag-id", "task-id", "dag-id-task-id"),  # no problem
+        ("dag_id", "task_id", "dag-id-task-id"),  # underscores
+        ("dag.id", "task.id", "dag-id-task-id"),  # dots
+        (".dag.id", ".---task.id", "dag-id-task-id"),  # leading dot invalid
+        ("**dag.id", "**task.id", "dag-id-task-id"),  # leading dot invalid
+        ("-90Abc*&", "-90Abc*&", "90abc-90abc"),  # invalid ends
+        ("90AçLbˆˆç˙ßß˜˜˙c*a", "90AçLbˆˆç˙ßß˜˜˙c*a", "90aclb-c-ssss-c-a-90aclb-c-ssss-c-a"),  # ugly
+    ],
+)
+def test_create_pod_id_dag_and_task(dag_id, task_id, expected):
+    actual = create_pod_id(dag_id=dag_id, task_id=task_id)
+    assert actual == expected
+    assert re.match(pod_name_regex, actual)
+
+
+def test_create_pod_id_dag_too_long():
+    actual = create_pod_id("0" * 254)
+    assert actual == "0" * 253
+    assert re.match(pod_name_regex, actual)

Makes things more readable.  E.g. my-dag-my-task instead of mydagmytask.
