fix(ingestion): cap OM and Snowflake socket waits to stop silent ingestion hangs

ingestion/Dockerfile

@@ -102,6 +102,12 @@ RUN [ $(uname -m) = "x86_64" ] \
   && pip install "openmetadata-ingestion[db2]~=${RI_VERSION}" \
   || echo "DB2 not supported on ARM architectures."
 
+# Ship py-spy so a hung worker can be sampled in place
+# (`py-spy dump --pid <pid>`) without first installing anything in the pod.
+# Container-only — kept out of setup.py to avoid forcing a native binary on
+# dev laptops / CI / non-container installs.
+RUN pip install "py-spy>=0.3.14"
+
 # bump python-daemon for https://github.com/apache/airflow/pull/29916
 RUN pip install "python-daemon>=3.0.0"
 # remove all airflow providers except for docker, cncf kubernetes, and standard (required in Airflow 3.x)

ingestion/src/metadata/ingestion/ometa/client.py

@@ -115,11 +115,14 @@
     raw_data: Optional[bool] = False  # noqa: UP045
     allow_redirects: Optional[bool] = False  # noqa: UP045
     auth_token_mode: Optional[str] = "Bearer"  # noqa: UP045
     verify: Optional[Union[bool, str]] = None  # noqa: UP007, UP045
     cookies: Optional[Any] = None  # noqa: UP045
     ttl_cache: int = 60
-    timeout: Optional[int] = None  # noqa: UP045
+    # (connect, read) seconds. Default prevents indefinite hangs when a pooled
+    # socket is silently severed (NAT/LB idle reaping). Override with None to
+    # disable, or pass a single int to use the same value for both.
+    timeout: Optional[int | tuple[int, int]] = (10, 300)  # noqa: UP045
     cert: Optional[Union[str, tuple]] = None  # noqa: UP007, UP045
 
 
 # pylint: disable=too-many-instance-attributes
@@ -175,7 +178,7 @@
             and self._auth_token
         ):
             self.config.access_token, expiry = self._auth_token()
             if not self.config.access_token == "no_token":  # noqa: SIM201
                 if isinstance(expiry, datetime):
                     self.config.expires_in = expiry.timestamp() - 120
                 else:

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

@@ -205,6 +205,13 @@ def get_connection(self) -> Engine:
         if keep_alive := self._get_client_session_keep_alive():
             connection.connectionArguments.root["client_session_keep_alive"] = keep_alive
 
+        # Bound the Snowflake socket so a silently-severed TCP connection
+        # (NAT/LB idle reaping in K8s/hybrid runners) surfaces as a network
+        # error within 10 minutes instead of hanging the worker indefinitely.
+        # User-supplied connectionArguments win via setdefault.
+        if connection.connectionArguments.root is not None:
+            connection.connectionArguments.root.setdefault("network_timeout", 600)
+
         engine = create_generic_db_connection(
             connection=connection,
             get_connection_url_fn=self.get_connection_url,