Fixes #24348: Strip URL scheme from hostPort to prevent ValueError

Author: RajdeepKushwaha5

ingestion/src/metadata/ingestion/connections/builders.py

@@ -12,6 +12,7 @@
 """
 Get and test connection utilities
 """
+
 from functools import partial
 from typing import Any, Callable, Dict, Optional
 from urllib.parse import quote_plus
@@ -34,6 +35,7 @@
 from metadata.ingestion.connections.query_logger import attach_query_tracker
 from metadata.ingestion.connections.secrets import connection_with_options_secrets
 from metadata.utils.constants import BUILDER_PASSWORD_ATTR
+from metadata.utils.db_utils import clean_host_port
 from metadata.utils.logger import cli_logger
 
 logger = cli_logger()
@@ -152,7 +154,13 @@ def get_password_secret(connection) -> SecretStr:
                 aws_client = AWSClient(
                     config=connection.authType.awsConfig
                 ).get_rds_client()
-                host, port = connection.hostPort.split(":")
+                cleaned = clean_host_port(connection.hostPort)
+                if ":" not in cleaned:
+                    raise ValueError(
+                        f"hostPort '{connection.hostPort}' must include a port "
+                        f"for IAM authentication (expected 'hostname:port')."
+                    )
+                host, port = cleaned.split(":", 1)
                 password = SecretStr(
                     aws_client.generate_db_auth_token(
                         DBHostname=host,
@@ -188,7 +196,7 @@ def get_connection_url_common(connection) -> str:
         url = _add_password(url, connection)
         url += "@"
 
-    url += connection.hostPort
+    url += clean_host_port(connection.hostPort)
     if hasattr(connection, "database"):
         url += f"/{connection.database}" if connection.database else ""
 

ingestion/src/metadata/ingestion/source/database/cassandra/connection.py

@@ -12,6 +12,7 @@
 """
 Source connection handler
 """
+
 from functools import partial
 from typing import Optional
 
@@ -44,6 +45,7 @@
     CASSANDRA_GET_RELEASE_VERSION,
 )
 from metadata.utils.constants import THREE_MIN
+from metadata.utils.db_utils import clean_host_port
 
 
 def get_connection(connection: CassandraConnection):
@@ -70,7 +72,7 @@ def get_connection(connection: CassandraConnection):
             }
         )
     else:
-        host, port = connection.hostPort.split(":")
+        host, port = clean_host_port(connection.hostPort).split(":")
         cluster_config.update({"contact_points": [host], "port": port})
         if connection.username and getattr(connection.authType, "password", None):
             cluster_config["auth_provider"] = PlainTextAuthProvider(

ingestion/src/metadata/ingestion/source/database/databricks/auth.py

@@ -12,6 +12,7 @@
 """
 This module provides authentication utilities for Databricks and Unity Catalog connections.
 """
+
 from typing import Union
 
 from databricks.sdk.core import Config, azure_service_principal, oauth_service_principal
@@ -31,6 +32,7 @@
 from metadata.generated.schema.entity.services.connections.database.unityCatalogConnection import (
     UnityCatalogConnection,
 )
+from metadata.utils.db_utils import clean_host_port
 
 
 def get_personal_access_token_auth(
@@ -50,7 +52,7 @@ def get_databricks_oauth_auth(
     """
 
     def credential_provider():
-        hostname = connection.hostPort.split(":")[0]
+        hostname = clean_host_port(connection.hostPort).split(":")[0]
         config = Config(
             host=f"https://{hostname}",
             client_id=connection.authType.clientId,
@@ -67,7 +69,7 @@ def get_azure_ad_auth(connection: Union[DatabricksConnection, UnityCatalogConnec
     """
 
     def credential_provider():
-        hostname = connection.hostPort.split(":")[0]
+        hostname = clean_host_port(connection.hostPort).split(":")[0]
         config = Config(
             host=f"https://{hostname}",
             azure_client_secret=connection.authType.azureClientSecret.get_secret_value(),

ingestion/src/metadata/ingestion/source/database/databricks/client.py

@@ -11,6 +11,7 @@
 """
 Client to interact with databricks apis
 """
+
 import base64
 import json
 import traceback
@@ -34,6 +35,7 @@
     DATABRICKS_GET_TABLE_LINEAGE,
 )
 from metadata.utils.constants import QUERY_WITH_DBT, QUERY_WITH_OM_VERSION
+from metadata.utils.db_utils import clean_host_port
 from metadata.utils.helpers import datetime_to_ts
 from metadata.utils.logger import ingestion_logger
 
@@ -62,7 +64,7 @@ def __init__(
         engine: Optional[Engine] = None,
     ):
         self.config = config
-        base_url, *_ = self.config.hostPort.split(":")
+        base_url, *_ = clean_host_port(self.config.hostPort).split(":")
         self.base_url = f"https://{base_url}{API_VERSION}"
         self.base_query_url = f"{self.base_url}{QUERIES_PATH}"
         self.base_job_url = f"https://{base_url}{JOB_API_VERSION}/jobs"

ingestion/src/metadata/ingestion/source/database/db2/connection.py

@@ -12,6 +12,7 @@
 """
 Source connection handler
 """
+
 import importlib
 import sys
 from pathlib import Path
@@ -43,6 +44,7 @@
     install_clidriver,
 )
 from metadata.utils.constants import THREE_MIN, UTF_8
+from metadata.utils.db_utils import clean_host_port
 from metadata.utils.logger import ingestion_logger
 from metadata.utils.ssl_manager import check_ssl_and_init
 
@@ -59,7 +61,7 @@ def _get_ibmi_connection_url(connection: Db2Connection) -> str:
 
     Port is passed separately via connect_args.
     """
-    hostname = connection.hostPort.split(":")[0]
+    hostname = clean_host_port(connection.hostPort).split(":")[0]
 
     url = f"{connection.scheme.value}://"
     if connection.username:
@@ -81,7 +83,7 @@ def _get_ibmi_connection_args(connection: Db2Connection) -> Dict[str, Any]:
     rejects it in the URL.
     """
     args = get_connection_args_common(connection)
-    host_port = connection.hostPort
+    host_port = clean_host_port(connection.hostPort)
     if ":" in host_port:
         port_str = host_port.split(":")[1]
         try:

ingestion/src/metadata/ingestion/source/database/redshift/connection.py

@@ -12,6 +12,7 @@
 """
 Source connection handler
 """
+
 from functools import partial
 from typing import Optional
 from urllib.parse import quote_plus
@@ -55,6 +56,7 @@
     REDSHIFT_TEST_GET_QUERIES_MAP,
 )
 from metadata.utils.constants import THREE_MIN
+from metadata.utils.db_utils import clean_host_port
 from metadata.utils.logger import ingestion_logger
 
 logger = ingestion_logger()
@@ -129,7 +131,7 @@ def _get_redshift_iam_credentials(connection: RedshiftConnection) -> tuple:
     Get temporary credentials for Redshift using IAM authentication.
     Detects Serverless vs Provisioned from the host and uses the appropriate API.
     """
-    host = connection.hostPort.split(":")[0]
+    host = clean_host_port(connection.hostPort).split(":")[0]
 
     if _is_serverless_host(host):
         return _get_serverless_iam_credentials(connection, host)
@@ -150,7 +152,7 @@ def get_redshift_connection_url(connection: RedshiftConnection) -> str:
 
         url = f"{connection.scheme.value}://"
         url += f"{quote_plus(username)}:{quote_plus(password)}@"
-        url += connection.hostPort
+        url += clean_host_port(connection.hostPort)
         url += f"/{connection.database}" if connection.database else ""
 
         options = get_connection_options_dict(connection)

ingestion/src/metadata/utils/db_utils.py

@@ -12,9 +12,11 @@
 """
 Helpers module for db sources
 """
+
 import time
 import traceback
 from typing import Iterable, List, Union
+from urllib.parse import urlparse
 
 from metadata.generated.schema.api.lineage.addLineage import AddLineageRequest
 from metadata.generated.schema.entity.data.table import Table
@@ -43,12 +45,43 @@
 PUBLIC_SCHEMA = "public"
 
 
+def clean_host_port(host_port: str) -> str:
+    """
+    Strip URL scheme prefixes from a hostPort string.
+
+    Users sometimes enter a full URL (e.g. 'http://localhost:3306')
+    instead of just 'localhost:3306'. This strips the scheme to avoid
+    ValueError when parsing host and port.
+    """
+    host_port = host_port.strip()
+    if "://" not in host_port:
+        return host_port.rstrip("/")
+
+    logger.warning(
+        "The hostPort '%s' contains a URL scheme. "
+        "Expected format is 'hostname[:port]' (e.g. 'localhost:3306'). "
+        "Stripping the scheme prefix.",
+        host_port,
+    )
+    parsed = urlparse(host_port)
+    hostname = parsed.hostname or ""
+    port = parsed.port
+
+    if not hostname:
+        # urlparse couldn't extract hostname (e.g. jdbc:postgresql://host:5432)
+        # Fall back to stripping everything before the last ://
+        raw = host_port.rsplit("://", 1)[-1].split("/")[0]
+        return raw
+
+    return f"{hostname}:{port}" if port else hostname
+
+
 def get_host_from_host_port(uri: str) -> str:
     """
     if uri is like "localhost:9000"
     then return the host "localhost"
     """
-    return uri.split(":")[0]
+    return clean_host_port(uri).split(":")[0]
 
 
 #  pylint: disable=too-many-locals

ingestion/tests/unit/test_build_connection_url.py

@@ -97,6 +97,34 @@ def test_get_connection_url_mysql(self):
                 "mysql+pymysql://openmetadata_user:mocked_token@localhost:3306/openmetadata_db",
             )
 
+    def test_get_connection_url_mysql_with_url_scheme(self):
+        """hostPort with http:// prefix should be cleaned automatically"""
+        connection = MysqlConnectionConfig(
+            username="openmetadata_user",
+            authType=BasicAuth(password="openmetadata_password"),
+            hostPort="http://localhost:3306",
+            databaseSchema="openmetadata_db",
+        )
+        engine_connection = MySQLConnection(connection).client
+        self.assertEqual(
+            engine_connection.url.render_as_string(hide_password=False),
+            "mysql+pymysql://openmetadata_user:openmetadata_password@localhost:3306/openmetadata_db",
+        )
+
+    def test_get_connection_url_postgres_with_url_scheme(self):
+        """hostPort with https:// prefix should be cleaned automatically"""
+        connection = PostgresConnectionConfig(
+            username="openmetadata_user",
+            authType=BasicAuth(password="openmetadata_password"),
+            hostPort="https://localhost:5432",
+            database="openmetadata_db",
+        )
+        engine_connection = PostgresConnection(connection).client
+        self.assertEqual(
+            engine_connection.url.render_as_string(hide_password=False),
+            "postgresql+psycopg2://openmetadata_user:openmetadata_password@localhost:5432/openmetadata_db",
+        )
+
     def test_get_connection_url_postgres(self):
         connection = PostgresConnectionConfig(
             username="openmetadata_user",

ingestion/tests/unit/test_db_utils.py

@@ -12,6 +12,7 @@
 """
 Unit tests for db_utils module
 """
+
 import uuid
 from copy import deepcopy
 from unittest import TestCase
@@ -36,7 +37,11 @@
 from metadata.ingestion.lineage.models import Dialect
 from metadata.ingestion.lineage.sql_lineage import search_cache
 from metadata.ingestion.source.models import TableView
-from metadata.utils.db_utils import get_host_from_host_port, get_view_lineage
+from metadata.utils.db_utils import (
+    clean_host_port,
+    get_host_from_host_port,
+    get_view_lineage,
+)
 
 
 # Mock LineageTable class to simulate collate_sqllineage.core.models.Table
@@ -118,6 +123,52 @@ def test_get_host_from_host_port(self):
         self.assertEqual(get_host_from_host_port("localhost"), "localhost")
         self.assertEqual(get_host_from_host_port("example.com"), "example.com")
 
+        # Test with URL scheme prefixes
+        self.assertEqual(get_host_from_host_port("http://localhost:3306"), "localhost")
+        self.assertEqual(
+            get_host_from_host_port("https://example.com:5432"), "example.com"
+        )
+        self.assertEqual(get_host_from_host_port("http://localhost"), "localhost")
+
+    def test_clean_host_port(self):
+        """Test clean_host_port strips URL scheme prefixes"""
+        # Already-clean values pass through unchanged
+        self.assertEqual(clean_host_port("localhost:3306"), "localhost:3306")
+        self.assertEqual(clean_host_port("127.0.0.1:5432"), "127.0.0.1:5432")
+        self.assertEqual(clean_host_port("example.com"), "example.com")
+
+        # HTTP prefix is stripped
+        self.assertEqual(clean_host_port("http://localhost:3306"), "localhost:3306")
+        self.assertEqual(clean_host_port("http://example.com:8080"), "example.com:8080")
+
+        # HTTPS prefix is stripped
+        self.assertEqual(clean_host_port("https://localhost:5432"), "localhost:5432")
+        self.assertEqual(
+            clean_host_port("https://mydb.example.com:3306"), "mydb.example.com:3306"
+        )
+
+        # Trailing slash is stripped
+        self.assertEqual(clean_host_port("http://localhost:3306/"), "localhost:3306")
+
+        # Host only with scheme
+        self.assertEqual(clean_host_port("http://localhost"), "localhost")
+        self.assertEqual(clean_host_port("https://example.com"), "example.com")
+
+        # URL with path is handled — path/query/fragment are discarded
+        self.assertEqual(clean_host_port("http://localhost:3306/db"), "localhost:3306")
+        self.assertEqual(
+            clean_host_port("https://example.com:5432/mydb?ssl=true"),
+            "example.com:5432",
+        )
+
+        # Whitespace is stripped
+        self.assertEqual(clean_host_port("  localhost:3306  "), "localhost:3306")
+        self.assertEqual(clean_host_port("  http://localhost:3306  "), "localhost:3306")
+
+        # JDBC-style URLs fall back to raw extraction
+        self.assertEqual(clean_host_port("jdbc:postgresql://host:5432"), "host:5432")
+        self.assertEqual(clean_host_port("jdbc:postgresql://host:5432/db"), "host:5432")
+
     @patch("metadata.utils.db_utils.ConnectionTypeDialectMapper")
     @patch("metadata.utils.db_utils.fqn")
     def test_get_view_lineage_success_with_lineage_parser(