@@ -1,3 +1,4 @@
+import json
 import logging
 import time
 from functools import wraps
@@ -67,15 +68,27 @@ def wrapper(*args, **kwargs):
 
 
 def _get_exception_data(except_str) -> str:
-    substring = "Flight error: "
+    if not isinstance(except_str, str):
+        return ""
 
-    # Find the starting index of the substring
+    substring = "Flight error: "
     position = except_str.find(substring)
-    end_json_index = except_str.find("}")
-
-    if position != -1 and end_json_index != -1:
-        # Extract the part of the string after the substring
-        result = except_str[position + len(substring) : end_json_index + 1]
-        return result
-
-    return ""
+    if position == -1:
+        return ""
+
+    search_start = position + len(substring)
+    json_start = except_str.find("{", search_start)
+    if json_start == -1:
+        return ""
+
+    search_pos = json_start
+    while True:
+        json_end = except_str.find("}", search_pos + 1)
+        if json_end == -1:
+            return ""
+        candidate = except_str[json_start : json_end + 1]
+        try:
+            json.loads(candidate)
+            return candidate
+        except (json.JSONDecodeError, ValueError):
+            search_pos = json_end
@@ -15,6 +15,7 @@
 from feast import FeatureStore, FeatureView, utils
 from feast.arrow_error_handler import arrow_server_error_handling_decorator
 from feast.data_source import DataSource
+from feast.errors import FeatureViewNotFoundException
 from feast.feature_logging import FeatureServiceLoggingSource
 from feast.feature_view import DUMMY_ENTITY_NAME
 from feast.infra.offline_stores.offline_utils import get_offline_store_from_config
@@ -199,7 +200,7 @@ def get_feature_view_by_name(
                             )
                             fv = fv.with_projection(p)
             return fv
-        except Exception:
+        except FeatureViewNotFoundException:
             try:
                 return self.store.registry.get_stream_feature_view(
                     name=fv_name, project=project
 
@@ -4,8 +4,11 @@
 import pytest
 from pydantic import ValidationError
 
-from feast.arrow_error_handler import arrow_client_error_handling_decorator
-from feast.errors import PermissionNotFoundException
+from feast.arrow_error_handler import (
+    _get_exception_data,
+    arrow_client_error_handling_decorator,
+)
+from feast.errors import FeatureViewNotFoundException, PermissionNotFoundException
 from feast.infra.offline_stores.remote import RemoteOfflineStoreConfig
 
 permissionError = PermissionNotFoundException("dummy_name", "dummy_project")
@@ -179,3 +182,50 @@ def method_on_client(self_arg):
     def test_config_rejects_negative_connection_retries(self):
         with pytest.raises(ValidationError):
             RemoteOfflineStoreConfig(host="localhost", connection_retries=-1)
+
+
+class TestGetExceptionData:
+    def test_non_string_input_returns_empty(self):
+        assert _get_exception_data(12345) == ""
+        assert _get_exception_data(None) == ""
+        assert _get_exception_data(b"bytes") == ""
+
+    def test_no_flight_error_prefix_returns_empty(self):
+        assert _get_exception_data("some random error") == ""
+
+    def test_flight_error_prefix_without_json_returns_empty(self):
+        assert _get_exception_data("Flight error: no json here") == ""
+
+    def test_extracts_json_from_flight_error(self):
+        fv_error = FeatureViewNotFoundException("my_view", "my_project")
+        error_str = f"Flight error: {fv_error.to_error_detail()}"
+        result = _get_exception_data(error_str)
+        assert '"class": "FeatureViewNotFoundException"' in result
+        assert '"module": "feast.errors"' in result
+
+    def test_extracts_json_with_trailing_text(self):
+        fv_error = FeatureViewNotFoundException("my_view", "my_project")
+        error_str = (
+            f"Flight error: {fv_error.to_error_detail()}. "
+            "gRPC client debug context: some extra info"
+        )
+        result = _get_exception_data(error_str)
+        assert '"class": "FeatureViewNotFoundException"' in result
+        assert '"module": "feast.errors"' in result
+
+    def test_extracts_json_with_grpc_debug_context_containing_braces(self):
+        fv_error = FeatureViewNotFoundException("my_view", "my_project")
+        error_str = (
+            f"Flight error: {fv_error.to_error_detail()}. "
+            "gRPC client debug context: UNKNOWN:Error received from peer "
+            'ipv4:127.0.0.1:59930 {grpc_message:"Flight error: ...", '
+            'grpc_status:2, created_time:"2026-03-17T17:32:07"}'
+        )
+        result = _get_exception_data(error_str)
+        assert '"class": "FeatureViewNotFoundException"' in result
+        assert '"module": "feast.errors"' in result
+        from feast.errors import FeastError
+
+        reconstructed = FeastError.from_error_detail(result)
+        assert reconstructed is not None
+        assert "my_view" in str(reconstructed)
@@ -1,6 +1,7 @@
 import os
 import tempfile
 from datetime import datetime, timedelta
+from unittest.mock import patch
 
 import assertpy
 import pandas as pd
@@ -345,3 +346,28 @@ def _test_pull_all_from_table_or_query(temp_dir, fs: FeatureStore):
         start_date=start_date,
         end_date=end_date,
     ).to_df()
+
+
+def test_get_feature_view_by_name_propagates_transient_errors():
+    """Transient registry errors must not be swallowed and misreported as
+    FeatureViewNotFoundException."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        store = default_store(str(temp_dir))
+        location = "grpc+tcp://localhost:0"
+
+        _init_auth_manager(store=store)
+        server = OfflineServer(store=store, location=location)
+
+        transient_error = ConnectionError("registry temporarily unavailable")
+
+        with patch.object(
+            server.store.registry,
+            "get_feature_view",
+            side_effect=transient_error,
+        ):
+            with pytest.raises(ConnectionError, match="registry temporarily"):
+                server.get_feature_view_by_name(
+                    fv_name="driver_hourly_stats",
+                    name_alias=None,
+                    project=PROJECT_NAME,
+                )