diff --git a/airbyte_cdk/entrypoint.py b/airbyte_cdk/entrypoint.py
index 57820f005..3ee0b68d1 100644
--- a/airbyte_cdk/entrypoint.py
+++ b/airbyte_cdk/entrypoint.py
@@ -430,10 +430,13 @@ def filtered_send(self: Any, request: PreparedRequest, **kwargs: Any) -> Respons
                     message="Invalid URL endpoint: The endpoint that data is being requested from belongs to a private network. Source connectors only support requesting data from public API endpoints.",
                 )
         except socket.gaierror as exception:
-            # This is a special case where the developer specifies an IP address string that is not formatted correctly like trailing
-            # whitespace which will fail the socket IP lookup. This only happens when using IP addresses and not text hostnames.
-            # Knowing that this is a request using the requests library, we will mock the exception without calling the lib
-            raise requests.exceptions.InvalidURL(f"Invalid URL {parsed_url}: {exception}")
+            # socket.gaierror fires on transient DNS failures (EAI_AGAIN / errno -3)
+            # for valid hostnames, not just malformed IPs.  Re-raise as ConnectionError
+            # so every caller (CDK streams *and* external SDKs) treats it as retryable.
+            hostname = parsed_url.hostname or ""
+            raise requests.ConnectionError(
+                f"DNS resolution failed for {str(hostname)}: {str(exception)}"
+            )
 
         return wrapped_fn(self, request, **kwargs)
 
diff --git a/unit_tests/connector_builder/test_connector_builder_handler.py b/unit_tests/connector_builder/test_connector_builder_handler.py
index 000699609..b99834b54 100644
--- a/unit_tests/connector_builder/test_connector_builder_handler.py
+++ b/unit_tests/connector_builder/test_connector_builder_handler.py
@@ -1220,7 +1220,7 @@ def test_read_source_single_page_single_slice(mock_http_stream):
         pytest.param(
             "CLOUD",
             "https://domainwithoutextension",
-            "Invalid URL",
+            "DNS resolution failed for domainwithoutextension",
             id="test_cloud_read_with_invalid_url_endpoint",
         ),
         pytest.param(
@@ -1300,7 +1300,7 @@ def test_handle_read_external_requests(deployment_mode, url_base, expected_error
         pytest.param(
             "CLOUD",
             "https://domainwithoutextension",
-            "Invalid URL",
+            "DNS resolution failed for domainwithoutextension",
             id="test_cloud_read_with_invalid_url_endpoint",
         ),
         pytest.param(
diff --git a/unit_tests/test_entrypoint.py b/unit_tests/test_entrypoint.py
index fcfb44915..4a3125211 100644
--- a/unit_tests/test_entrypoint.py
+++ b/unit_tests/test_entrypoint.py
@@ -3,6 +3,8 @@
 #
 
 import os
+import socket
+import sys
 from argparse import Namespace
 from collections import defaultdict
 from copy import deepcopy
@@ -559,7 +561,7 @@ def test_invalid_command(entrypoint: AirbyteEntrypoint, config_mock):
         pytest.param(
             "CLOUD",
             "https://192.168.27.30    ",
-            ValueError,
+            requests.ConnectionError,
             id="test_cloud_incorrect_ip_format_is_rejected",
         ),
         pytest.param(
@@ -944,3 +946,57 @@ def test_memory_failfast_flushes_queued_state_before_raising(mocker):
     with pytest.raises(AirbyteTracedException) as exc_info:
         next(gen)
     assert exc_info.value is fail_fast_exc
+
+
+@pytest.mark.parametrize(
+    "gaierror_errno,gaierror_msg",
+    [
+        pytest.param(
+            socket.EAI_AGAIN, "Temporary failure in name resolution", id="transient_dns_EAI_AGAIN"
+        ),
+        pytest.param(socket.EAI_NONAME, "Name or service not known", id="unknown_host_EAI_NONAME"),
+        pytest.param(
+            socket.EAI_FAIL,
+            "Non-recoverable failure in name resolution",
+            id="permanent_dns_EAI_FAIL",
+        ),
+    ],
+)
+def test_filtered_send_raises_connection_error_on_dns_failure(gaierror_errno, gaierror_msg):
+    """filtered_send must re-raise socket.gaierror as ConnectionError, not InvalidURL."""
+    entrypoint_module._init_internal_request_filter()
+
+    prepared = requests.PreparedRequest()
+    prepared.prepare_url("https://graph.facebook.com/v25.0/", None)
+
+    dns_error = socket.gaierror(gaierror_errno, gaierror_msg)
+
+    with patch.object(entrypoint_module, "_is_private_url", side_effect=dns_error):
+        with pytest.raises(
+            requests.ConnectionError, match="DNS resolution failed for graph.facebook.com"
+        ):
+            requests.Session().send(prepared)
+
+
+def test_filtered_send_dns_failure_is_not_invalid_url():
+    """Verify DNS errors are raised as ConnectionError, not InvalidURL."""
+    entrypoint_module._init_internal_request_filter()
+
+    prepared = requests.PreparedRequest()
+    prepared.prepare_url("https://graph.facebook.com/v25.0/", None)
+
+    dns_error = socket.gaierror(socket.EAI_AGAIN, "Temporary failure in name resolution")
+
+    with patch.object(entrypoint_module, "_is_private_url", side_effect=dns_error):
+        with pytest.raises(requests.ConnectionError):
+            requests.Session().send(prepared)
+
+    # The raised exception must not be a subclass of InvalidURL
+    with patch.object(entrypoint_module, "_is_private_url", side_effect=dns_error):
+        try:
+            requests.Session().send(prepared)
+            pytest.fail("Expected ConnectionError to be raised")
+        except requests.ConnectionError:
+            assert not isinstance(sys.exc_info()[1], requests.exceptions.InvalidURL), (
+                "DNS failure should raise ConnectionError, not InvalidURL"
+            )