code: def test_parquet_deserialize_fragments_with_retry(
      ray_start_regular_shared, fs, data_path, monkeypatch
  ):
      setup_data_path = _unwrap_protocol(data_path)
      df1 = pd.DataFrame({"one": [1, 2, 3], "two": ["a", "b", "c"]})
      table = pa.Table.from_pandas(df1)
      path1 = os.path.join(setup_data_path, "test1.parquet")
      pq.write_table(table, path1, filesystem=fs)
      df2 = pd.DataFrame({"one": [4, 5, 6], "two": ["e", "f", "g"]})
      table = pa.Table.from_pandas(df2)
      path2 = os.path.join(setup_data_path, "test2.parquet")
      pq.write_table(table, path2, filesystem=fs)
  
      dataset_kwargs = {}
      pq_ds = pq.ParquetDataset(
          data_path,
          **dataset_kwargs,
          filesystem=fs,
          use_legacy_dataset=False,
      )
      serialized_fragments = [
          _SerializedFragment(p) for p in pq_ds.fragments
      ]
  
      # test 1st attempt succeed
      fragments = _deserialize_fragments_with_retry(
          serialized_fragments
      )
      assert "test1.parquet" in fragments[0].path
      assert "test2.parquet" in fragments[1].path
  
      # test the 3rd attempt succeed with a mock function constructed
      # to throw in the first two attempts
      class MockDeserializer:
          def __init__(self, planned_exp_or_return):
              self.planned_exp_or_return = planned_exp_or_return
              self.cur_index = 0
  
          def __call__(self, *args: Any, **kwds: Any) -> Any:
              exp_or_ret = self.planned_exp_or_return[self.cur_index]
              self.cur_index += 1
              if type(exp_or_ret) == Exception:
                  raise exp_or_ret
              else:
                  return exp_or_ret
  
      mock_deserializer = MockDeserializer(
          [
              Exception("1st mock failed attempt"),
              Exception("2nd mock failed attempt"),
              fragments,
          ]
      )
      monkeypatch.setattr(
          ray.data.datasource.parquet_datasource,
          "_deserialize_fragments",
          mock_deserializer,
      )
      retried_fragments = _deserialize_fragments_with_retry(
          serialized_fragments
      )
      assert "test1.parquet" in retried_fragments[0].path
      assert "test2.parquet" in retried_fragments[1].path
  
imports: import os
  import shutil
  from typing import Any
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  from pytest_lazyfixture import lazy_fixture
  import ray
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.datasource import DefaultFileMetadataProvider, DefaultParquetMetadataProvider
  from ray.data.datasource.parquet_base_datasource import ParquetBaseDatasource
  from ray.data.datasource.parquet_datasource import NUM_CPUS_FOR_META_FETCH_TASK, PARALLELIZE_META_FETCH_THRESHOLD, RETRY_EXCEPTIONS_FOR_META_FETCH_TASK, ParquetDatasource, _deserialize_fragments_with_retry, _SerializedFragment
  from ray.data.datasource.path_util import _unwrap_protocol
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.tests.conftest import *
label_desc: Use image node_id 3 for calling a global function with example usage: test_parquet_deserialize_fragments_with_retry(ray_start_regular_shared, fs, data_path, monkeypatch) and returns: exp_or_ret