Added additionall Error Messaging for Lake Formation and AccessDenied

exceptions. New helped method to get if a FG is lake formation governed.
Also added testing for these features.
---
X-AI-Prompt: Add error checking in feature_group_manager.py to differentiate whether a Glue permissions error (AccessDeniedException) during iceberg properties operations is related to Lake Formation governance or regular IAM. Check the
  feature group's describe response for LakeFormationConfig before the call, and surface a targeted error message accordingly.
X-AI-Tool: Kiro CLI sisyphus

Author: alexyoung13

sagemaker-mlops/src/sagemaker/mlops/feature_store/feature_group_manager.py

@@ -25,6 +25,7 @@
 from sagemaker.core.s3.utils import parse_s3_url
 from sagemaker.core.common_utils import aws_partition
 from boto3 import Session
+from botocore.exceptions import ClientError
 from pyiceberg.catalog import load_catalog
 from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
 from sagemaker.mlops.feature_store.feature_utils import _APPROVED_ICEBERG_PROPERTIES
@@ -76,6 +77,16 @@ class FeatureGroupManager(FeatureGroup):
     if FeatureGroup.__doc__ and __doc__:
         __doc__ = FeatureGroup.__doc__
 
+    def _has_lake_formation_config(self) -> bool:
+        """Check if this feature group was created with Lake Formation governance.
+
+        Note: Returns False if the object was not hydrated via get()/describe
+        (i.e., constructed directly with just a name). In that case, the caller
+        falls back to the generic IAM error message, which is still valid advice.
+        """
+        lf_config = getattr(self, "lake_formation_config", None)
+        return lf_config is not None and lf_config != Unassigned()
+
     def _validate_table_ownership(self, table, database_name: str, table_name: str):
         """Validate that the Iceberg table belongs to this feature group by checking S3 location."""
         table_location = table.metadata.location if table.metadata else None
@@ -174,9 +185,26 @@ def _get_iceberg_properties(
                 "properties": dict(table.properties),
             }
 
+        except ClientError as e:
+            if e.response["Error"]["Code"] == "AccessDeniedException":
+                if self._has_lake_formation_config():
+                    raise PermissionError(
+                        f"Access denied reading Iceberg properties for '{self.feature_group_name}'. "
+                        f"This feature group uses Lake Formation governance — ensure you have "
+                        f"SELECT and DESCRIBE permissions on the table in Lake Formation, "
+                        f"in addition to IAM permissions."
+                    ) from e
+                raise PermissionError(
+                    f"Access denied reading Iceberg properties for '{self.feature_group_name}'. "
+                    f"Ensure your role has glue:GetTable permission "
+                    f"on the feature group's Glue table."
+                ) from e
+            raise RuntimeError(
+                f"Failed to get Iceberg properties for '{self.feature_group_name}': {e}"
+            ) from e
         except Exception as e:
             raise RuntimeError(
-                f"Failed to get Iceberg properties for {self.feature_group_name}: {e}"
+                f"Failed to get Iceberg properties for '{self.feature_group_name}': {e}"
             ) from e
 
     @retry(
@@ -268,6 +296,27 @@ def _update_iceberg_properties(
                 "properties_updated": iceberg_properties.properties,
             }
 
+        except ClientError as e:
+            if e.response["Error"]["Code"] == "AccessDeniedException":
+                if self._has_lake_formation_config():
+                    raise PermissionError(
+                        f"Access denied updating Iceberg properties for '{self.feature_group_name}'. "
+                        f"This feature group uses Lake Formation governance — ensure you have "
+                        f"ALTER permission on the table in Lake Formation, "
+                        f"in addition to IAM permissions."
+                    ) from e
+                raise PermissionError(
+                    f"Access denied updating Iceberg properties for '{self.feature_group_name}'. "
+                    f"Ensure your role has glue:UpdateTable permission "
+                    f"on the feature group's Glue table."
+                ) from e
+            logger.error(
+                f"Failed to update Iceberg properties for feature group "
+                f"'{self.feature_group_name}'. Attempted changes: {changed}. Error: {e}"
+            )
+            raise RuntimeError(
+                f"Failed to update Iceberg properties for '{self.feature_group_name}': {e}"
+            ) from e
         except Exception as e:
             logger.error(
                 f"Failed to update Iceberg properties for feature group "

sagemaker-mlops/tests/unit/sagemaker/mlops/feature_store/test_iceberg_properties.py

@@ -993,3 +993,182 @@ def test_passes_session_and_region_to_get_iceberg_properties(self, mock_get_clie
         )
 
         mock_get_iceberg.assert_called_once_with(session=mock_session, region="us-east-1")
+
+
+class TestHasLakeFormationConfig:
+    """Tests for _has_lake_formation_config method."""
+
+    def test_returns_false_when_attribute_missing(self):
+        """Test returns False when lake_formation_config attribute does not exist."""
+        fg = MagicMock(spec=FeatureGroupManager)
+        fg._has_lake_formation_config = FeatureGroupManager._has_lake_formation_config.__get__(fg)
+        # MagicMock(spec=...) won't have lake_formation_config since it's not on the class
+        del fg.lake_formation_config
+        assert fg._has_lake_formation_config() is False
+
+    def test_returns_false_when_none(self):
+        """Test returns False when lake_formation_config is None."""
+        fg = MagicMock(spec=FeatureGroupManager)
+        fg._has_lake_formation_config = FeatureGroupManager._has_lake_formation_config.__get__(fg)
+        fg.lake_formation_config = None
+        assert fg._has_lake_formation_config() is False
+
+    def test_returns_false_when_unassigned(self):
+        """Test returns False when lake_formation_config is Unassigned()."""
+        from sagemaker.core.shapes import Unassigned
+
+        fg = MagicMock(spec=FeatureGroupManager)
+        fg._has_lake_formation_config = FeatureGroupManager._has_lake_formation_config.__get__(fg)
+        fg.lake_formation_config = Unassigned()
+        assert fg._has_lake_formation_config() is False
+
+    def test_returns_true_when_config_present(self):
+        """Test returns True when lake_formation_config has a real value."""
+        fg = MagicMock(spec=FeatureGroupManager)
+        fg._has_lake_formation_config = FeatureGroupManager._has_lake_formation_config.__get__(fg)
+        fg.lake_formation_config = MagicMock()
+        assert fg._has_lake_formation_config() is True
+
+
+class TestGetIcebergPropertiesAccessDenied:
+    """Tests for AccessDeniedException handling in _get_iceberg_properties."""
+
+    def setup_method(self):
+        from sagemaker.core.shapes import OfflineStoreConfig, S3StorageConfig, DataCatalogConfig
+
+        self.fg = MagicMock(spec=FeatureGroupManager)
+        self.fg._get_iceberg_properties = FeatureGroupManager._get_iceberg_properties.__get__(self.fg)
+        self.fg.feature_group_name = "test-fg"
+        self.fg.offline_store_config = OfflineStoreConfig(
+            s3_storage_config=S3StorageConfig(s3_uri="s3://test-bucket/path"),
+            data_catalog_config=DataCatalogConfig(
+                catalog="AwsDataCatalog", database="test_db", table_name="test_table"
+            ),
+            table_format="Iceberg",
+        )
+
+    def _make_client_error(self, code):
+        from botocore.exceptions import ClientError
+        return ClientError({"Error": {"Code": code, "Message": "denied"}}, "GetTable")
+
+    @patch("sagemaker.mlops.feature_store.feature_group_manager.load_catalog")
+    def test_access_denied_with_lake_formation_raises_permission_error(self, mock_load_catalog):
+        """Test PermissionError with LF message when AccessDenied and LF config present."""
+        mock_catalog = MagicMock()
+        mock_catalog.load_table.side_effect = self._make_client_error("AccessDeniedException")
+        mock_load_catalog.return_value = mock_catalog
+        self.fg._has_lake_formation_config.return_value = True
+
+        with pytest.raises(PermissionError, match="Lake Formation governance"):
+            self.fg._get_iceberg_properties(session=MagicMock())
+
+    @patch("sagemaker.mlops.feature_store.feature_group_manager.load_catalog")
+    def test_access_denied_without_lake_formation_raises_permission_error(self, mock_load_catalog):
+        """Test PermissionError with IAM message when AccessDenied and no LF config."""
+        mock_catalog = MagicMock()
+        mock_catalog.load_table.side_effect = self._make_client_error("AccessDeniedException")
+        mock_load_catalog.return_value = mock_catalog
+        self.fg._has_lake_formation_config.return_value = False
+
+        with pytest.raises(PermissionError, match="glue:GetTable"):
+            self.fg._get_iceberg_properties(session=MagicMock())
+
+    @patch("sagemaker.mlops.feature_store.feature_group_manager.load_catalog")
+    def test_non_access_denied_client_error_raises_runtime_error(self, mock_load_catalog):
+        """Test RuntimeError for non-AccessDenied ClientError."""
+        mock_catalog = MagicMock()
+        mock_catalog.load_table.side_effect = self._make_client_error("EntityNotFoundException")
+        mock_load_catalog.return_value = mock_catalog
+
+        with pytest.raises(RuntimeError, match="Failed to get Iceberg properties"):
+            self.fg._get_iceberg_properties(session=MagicMock())
+
+    @patch("sagemaker.mlops.feature_store.feature_group_manager.load_catalog")
+    def test_access_denied_permission_error_chains_original(self, mock_load_catalog):
+        """Test that PermissionError chains the original ClientError via __cause__."""
+        original = self._make_client_error("AccessDeniedException")
+        mock_catalog = MagicMock()
+        mock_catalog.load_table.side_effect = original
+        mock_load_catalog.return_value = mock_catalog
+        self.fg._has_lake_formation_config.return_value = False
+
+        with pytest.raises(PermissionError) as exc_info:
+            self.fg._get_iceberg_properties(session=MagicMock())
+        assert exc_info.value.__cause__ is original
+
+
+class TestUpdateIcebergPropertiesAccessDenied:
+    """Tests for AccessDeniedException handling in _update_iceberg_properties."""
+
+    def setup_method(self):
+        self.fg = MagicMock(spec=FeatureGroupManager)
+        self.fg._update_iceberg_properties = FeatureGroupManager._update_iceberg_properties.__get__(self.fg)
+        self.fg.feature_group_name = "test-fg"
+
+    def _make_client_error(self, code):
+        from botocore.exceptions import ClientError
+        return ClientError({"Error": {"Code": code, "Message": "denied"}}, "UpdateTable")
+
+    def _setup_get_result(self):
+        mock_table = MagicMock()
+        mock_table.transaction.return_value.__enter__ = MagicMock(return_value=MagicMock())
+        mock_table.transaction.return_value.__exit__ = MagicMock(return_value=False)
+        self.fg._get_iceberg_properties.return_value = {
+            "database_name": "test_db",
+            "table_name": "test_table",
+            "table": mock_table,
+            "properties": {},
+        }
+        return mock_table
+
+    def test_access_denied_with_lake_formation_raises_permission_error(self):
+        """Test PermissionError with LF message when AccessDenied and LF config present."""
+        mock_table = self._setup_get_result()
+        mock_table.transaction().__enter__().set_properties.side_effect = self._make_client_error("AccessDeniedException")
+        self.fg._has_lake_formation_config.return_value = True
+
+        props = IcebergProperties(properties={"write.target-file-size-bytes": "536870912"})
+        with pytest.raises(PermissionError, match="Lake Formation governance"):
+            self.fg._update_iceberg_properties(iceberg_properties=props)
+
+    def test_access_denied_without_lake_formation_raises_permission_error(self):
+        """Test PermissionError with IAM message when AccessDenied and no LF config."""
+        mock_table = self._setup_get_result()
+        mock_table.transaction().__enter__().set_properties.side_effect = self._make_client_error("AccessDeniedException")
+        self.fg._has_lake_formation_config.return_value = False
+
+        props = IcebergProperties(properties={"write.target-file-size-bytes": "536870912"})
+        with pytest.raises(PermissionError, match="glue:UpdateTable"):
+            self.fg._update_iceberg_properties(iceberg_properties=props)
+
+    def test_non_access_denied_client_error_raises_runtime_error(self):
+        """Test RuntimeError for non-AccessDenied ClientError."""
+        mock_table = self._setup_get_result()
+        mock_table.transaction().__enter__().set_properties.side_effect = self._make_client_error("InternalServiceException")
+        self.fg._has_lake_formation_config.return_value = False
+
+        props = IcebergProperties(properties={"write.target-file-size-bytes": "536870912"})
+        with pytest.raises(RuntimeError, match="Failed to update Iceberg properties"):
+            self.fg._update_iceberg_properties(iceberg_properties=props)
+
+    def test_access_denied_permission_error_chains_original(self):
+        """Test that PermissionError chains the original ClientError via __cause__."""
+        mock_table = self._setup_get_result()
+        original = self._make_client_error("AccessDeniedException")
+        mock_table.transaction().__enter__().set_properties.side_effect = original
+        self.fg._has_lake_formation_config.return_value = True
+
+        props = IcebergProperties(properties={"write.target-file-size-bytes": "536870912"})
+        with pytest.raises(PermissionError) as exc_info:
+            self.fg._update_iceberg_properties(iceberg_properties=props)
+        assert exc_info.value.__cause__ is original
+
+    def test_access_denied_lf_message_mentions_alter(self):
+        """Test that LF error message mentions ALTER permission."""
+        mock_table = self._setup_get_result()
+        mock_table.transaction().__enter__().set_properties.side_effect = self._make_client_error("AccessDeniedException")
+        self.fg._has_lake_formation_config.return_value = True
+
+        props = IcebergProperties(properties={"write.target-file-size-bytes": "536870912"})
+        with pytest.raises(PermissionError, match="ALTER permission"):
+            self.fg._update_iceberg_properties(iceberg_properties=props)