Repository: ray-project/ray
File: python/ray/data/_internal/datasource/iceberg_datasink.py (Python)

Code before:
            if self._overwrite_kwargs.pop(invalid_param, None) is not None:
                logger.warning(
                    f"Removed '{invalid_param}' from overwrite_kwargs: {reason}"
                )

        if "name" in self._catalog_kwargs:
            self._catalog_name = self._catalog_kwargs.pop("name")
        else:
            self._catalog_name = "default"

        self._table: "Table" = None
        self._io: "FileIO" = None
        self._table_metadata: "TableMetadata" = None
        self._data_context = DataContext.get_current()

    def __getstate__(self) -> dict:
        """Exclude `_table` during pickling."""
        state = self.__dict__.copy()
        state.pop("_table", None)
        return state

    def __setstate__(self, state: dict) -> None:
        self.__dict__.update(state)
        self._table = None

    def _with_catalog_retry(self, func: Callable, description: str) -> Any:
        """Execute a function with catalog retry logic.

        This helper encapsulates the common retry pattern for Iceberg catalog
        operations, using the configured retry parameters from DataContext.

        Args:
            func: The callable to execute with retry logic.
            description: Human-readable description for logging/error messages.

        Returns:
            The result of calling func.
        """
        iceberg_config = self._data_context.iceberg_config
        return call_with_retry(
            func,
            description=description,
            match=iceberg_config.catalog_retried_errors,
            max_attempts=iceberg_config.catalog_max_attempts,
            max_backoff_s=iceberg_config.catalog_retry_max_backoff_s,
        )

    def _get_catalog(self) -> "Catalog":
        from pyiceberg import catalog

        return self._with_catalog_retry(

Review comment (suggestion):
```suggestion
    def _with_retry(self, func: Callable, description: str) -> Any:
```

Code after:
            if self._overwrite_kwargs.pop(invalid_param, None) is not None:
                logger.warning(
                    f"Removed '{invalid_param}' from overwrite_kwargs: {reason}"
                )

        if "name" in self._catalog_kwargs:
            self._catalog_name = self._catalog_kwargs.pop("name")
        else:
            self._catalog_name = "default"

        self._table: "Table" = None
        self._io: "FileIO" = None
        self._table_metadata: "TableMetadata" = None
        self._data_context = DataContext.get_current()

    def __getstate__(self) -> dict:
        """Exclude `_table` during pickling."""
        state = self.__dict__.copy()
        state.pop("_table", None)
        return state

    def __setstate__(self, state: dict) -> None:
        self.__dict__.update(state)
        self._table = None

    def _with_retry(self, func: Callable, description: str) -> Any:
        """Execute a function with retry logic.

        This helper encapsulates the common retry pattern for Iceberg catalog
        operations, using the configured retry parameters from DataContext.

        Args:
            func: The callable to execute with retry logic.
            description: Human-readable description for logging/error messages.

        Returns:
            The result of calling func.
        """
        iceberg_config = self._data_context.iceberg_config
        return call_with_retry(
            func,
            description=description,
            match=iceberg_config.catalog_retried_errors,
            max_attempts=iceberg_config.catalog_max_attempts,
            max_backoff_s=iceberg_config.catalog_retry_max_backoff_s,
        )

    def _get_catalog(self) -> "Catalog":
        from pyiceberg import catalog

        return self._with_retry(