fix(ingestion): make strip_hostport_scheme public and validate JDBC port

- Renames _strip_hostport_scheme to public strip_hostport_scheme so cross-module imports (db_utils.py) no longer depend on a private symbol; private alias kept for back-compat.

- Validates the port in the JDBC-style fallback branch so inputs like 'jdbc:postgresql://host:abc/db' raise ValueError with a clear message, matching the docstring contract instead of silently passing a broken hostPort downstream.

- Extends test_clean_host_port with the new fallback raise cases.

Author: RajdeepKushwaha5

ingestion/src/metadata/ingestion/models/custom_pydantic.py

@@ -35,7 +35,7 @@
 JSON_ENCODERS = "json_encoders"
 
 
-def _strip_hostport_scheme(raw: str) -> str:
+def strip_hostport_scheme(raw: str) -> str:
     """
     Strip an accidental URL scheme from a hostPort string.
 
@@ -44,7 +44,9 @@ def _strip_hostport_scheme(raw: str) -> str:
     bootstrap path of every generated Connection class.
 
     Raises ValueError if the scheme carries a non-numeric port so the user
-    gets a clear error instead of a silently broken hostPort.
+    gets a clear error instead of a silently broken hostPort. This applies to
+    both standard URLs handled by ``urlparse`` and to JDBC-style URLs handled
+    by the fallback branch (e.g. ``jdbc:postgresql://host:abc/db``).
     """
     value = raw.strip()
     if "://" not in value:
@@ -78,12 +80,32 @@ def _strip_hostport_scheme(raw: str) -> str:
             tail = tail.split(sep, 1)[0]
         if "@" in tail:
             tail = tail.rsplit("@", 1)[-1]
+
+        # Validate the port in the fallback path so the same ValueError
+        # contract holds for JDBC-style URLs (e.g. 'jdbc:postgresql://host:abc').
+        fallback_port: Optional[str] = None
+        if tail.startswith("["):
+            closing = tail.find("]")
+            if closing != -1 and len(tail) > closing + 1 and tail[closing + 1] == ":":
+                fallback_port = tail[closing + 2 :]
+        elif ":" in tail:
+            fallback_port = tail.rsplit(":", 1)[1]
+        if fallback_port and not fallback_port.isdigit():
+            raise ValueError(
+                f"Invalid hostPort '{safe_label}'. Expected format is "
+                "'hostname[:port]' (e.g. 'localhost:3306')."
+            )
         return tail
 
     host = f"[{hostname}]" if ":" in hostname else hostname
     return f"{host}:{port}" if port is not None else host
 
 
+# Backwards-compatible private alias retained for any internal callers that
+# pinned to the original underscored symbol while the helper was private.
+_strip_hostport_scheme = strip_hostport_scheme
+
+
 class BaseModel(PydanticBaseModel):
     """
     Base model for OpenMetadata generated models.
@@ -108,7 +130,7 @@ def model_post_init(self, context: Any, /):
                 # Let ValueError propagate: if the hostPort cannot be parsed
                 # (e.g. non-numeric port), the user must fix their config
                 # rather than silently getting a broken hostPort.
-                object.__setattr__(self, "hostPort", _strip_hostport_scheme(raw))
+                object.__setattr__(self, "hostPort", strip_hostport_scheme(raw))
 
         try:
             for field in self.__pydantic_fields__:

ingestion/src/metadata/utils/db_utils.py

@@ -33,7 +33,7 @@
     get_lineage_by_query,
     get_lineage_via_table_entity,
 )
-from metadata.ingestion.models.custom_pydantic import _strip_hostport_scheme
+from metadata.ingestion.models.custom_pydantic import strip_hostport_scheme
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.models import TableView
 from metadata.utils import fqn
@@ -59,7 +59,7 @@ def clean_host_port(host_port: str) -> str:
     value = host_port.strip()
     if "://" not in value:
         return value.rstrip("/")
-    return _strip_hostport_scheme(value)
+    return strip_hostport_scheme(value)
 
 
 def get_host_from_host_port(uri: str) -> str:

ingestion/tests/unit/test_db_utils.py

@@ -201,6 +201,12 @@ def test_clean_host_port(self):
         with self.assertRaises(ValueError):
             clean_host_port("http://localhost:abc")
 
+        # Non-numeric port in JDBC-style fallback also raises ValueError
+        with self.assertRaises(ValueError):
+            clean_host_port("jdbc:postgresql://host:abc/db")
+        with self.assertRaises(ValueError):
+            clean_host_port("jdbc:postgresql://[::1]:abc")
+
     @patch("metadata.utils.db_utils.ConnectionTypeDialectMapper")
     @patch("metadata.utils.db_utils.fqn")
     def test_get_view_lineage_success_with_lineage_parser(