FileExpired exception when reading parquet from a Minio bucket using Dask

**Describe the issue**:

I have a list of dataframes in a Minio bucket that are updated every 15 minutes. My script runs inside a Docker container in a loop and every 15 minutes a list of futures is created to read and preprocess every dataframe in the bucket. When computing the result it happens sometimes that the following exception is triggered:
```
s3fs.utils.FileExpired: [Errno 16] The remote file corresponding to filename filename.part.0.parquet and Etag "76b9a0ed3044b29e8a326d6f4ade2036" no longer exists
```

and triggers `TypeError: __init__() missing 1 required positional argument: 'e_tag'`. Even catching the exception does not solve the problem since during the next iteration it is triggered again. I checked on Minio and the Etag effectively does not correspond to that in the exception but I do not know how to solve this problem. The code to read data is this

```
df = dd.read_parquet('s3://{}/{}/{}'.format(bucket, path, filename),
storage_options={
"key": key,
"secret": secret,
"client_kwargs": {'endpoint_url': 'http://{}'.format(minio_endpoint)}
})
```

**Minimal Complete Verifiable Example**:
Providing a verifiable example is difficult since it runs on Docker and it is the result of various interacting scripts. I tried to replicate it running a simple script outside of docker but the problem does not appear. This is the script I used that is similar to what the original script does.

```
def load_user(bucket, path, file, minio_endpoint, key, secret):
print(f'Reading {file}...')
df = dd.read_parquet('s3://{}/{}/{}'.format(bucket, path, hash_filename(file)),
storage_options={
"key": key,
"secret": secret,
"client_kwargs": {'endpoint_url': 'http://{}'.format(minio_endpoint)}
},
ignore_metadata_file=True)

return df

if __name__ == "__main__":
bucket = bucket name
path = path
filenames = list of filenames

cluster = LocalCluster(silence_logs=logging.CRITICAL)
dask_client = Client(cluster)
minio_client = Minio(endpoint, key, secret, secure=False)

minio_endpoint = minio_client._base_url.host
key = minio_client._provider._credentials.access_key
secret = minio_client._provider._credentials.secret_key

while True:
futures = []
for file in filenames:
future = dask_client.submit(load_user,
bucket,
path,
file,
minio_endpoint,
key,
secret)
futures.append(future)

for user, future in zip(filenames, futures):
df = future.result()
```

**Anything else we need to know?**:

I have tried to use the function `invalidate_cache()` of s3fs and using `ignore_metadata_file=True` when reading data but it didn't worked. Catching the exception works but the problem is not solved during the following iteration.

Here is the complete traceback if you find it useful

```
Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 113, in _error_wrapper return await func(*args, **kwargs) File "/usr/local/lib/python3.9/site-packages/aiobotocore/client.py", line 408, in _make_api_call raise error_class(parsed_response, operation_name)botocore.exceptions.ClientError: An error occurred (PreconditionFailed) when calling the GetObject operation: At least one of the pre-conditions you specified did not holdThe above exception was the direct cause of the following exception:Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 2300, in _fetch_range return _fetch_range( File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 2462, in _fetch_range return sync(fs.loop, _inner_fetch, fs, bucket, key, version_id, start, end, req_kw) File "/usr/local/lib/python3.9/site-packages/fsspec/asyn.py", line 103, in sync raise return_result File "/usr/local/lib/python3.9/site-packages/fsspec/asyn.py", line 56, in _runner result[0] = await coro File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 2480, in _inner_fetch return await _error_wrapper(_call_and_read, retries=fs.retries) File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 142, in _error_wrapper raise err File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 113, in _error_wrapper return await func(*args, **kwargs) File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 2467, in _call_and_read resp = await fs._call_s3( File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 362, in _call_s3 return await _error_wrapper( File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 142, in _error_wrapper raise errOSError: [Errno 22] At least one of the pre-conditions you specified did not holdThe above exception was the direct cause of the following exception:Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/dask/backends.py", line 141, in wrapper return func(*args, **kwargs) File "/usr/local/lib/python3.9/site-packages/dask/dataframe/io/parquet/core.py", line 529, in read_parquet read_metadata_result = engine.read_metadata( File "/usr/local/lib/python3.9/site-packages/dask/dataframe/io/parquet/arrow.py", line 546, in read_metadata dataset_info = cls._collect_dataset_info( File "/usr/local/lib/python3.9/site-packages/dask/dataframe/io/parquet/arrow.py", line 1061, in _collect_dataset_info ds = pa_ds.dataset( File "/usr/local/lib/python3.9/site-packages/pyarrow/dataset.py", line 785, in dataset return _filesystem_dataset(source, **kwargs) File "/usr/local/lib/python3.9/site-packages/pyarrow/dataset.py", line 475, in _filesystem_dataset return factory.finish(schema) File "pyarrow/_dataset.pyx", line 3025, in pyarrow._dataset.DatasetFactory.finish File "pyarrow/error.pxi", line 154, in pyarrow.lib.pyarrow_internal_check_status File "pyarrow/error.pxi", line 88, in pyarrow.lib.check_status File "/usr/local/lib/python3.9/site-packages/fsspec/spec.py", line 1846, in read out = self.cache._fetch(self.loc, self.loc + length) File "/usr/local/lib/python3.9/site-packages/fsspec/caching.py", line 189, in _fetch self.cache = self.fetcher(start, end) # new block replaces old File "/usr/local/lib/python3.9/site-packages/s3fs/core.py", line 2312, in _fetch_range raise FileExpired(s3fs.utils.FileExpired: [Errno 16] The remote file corresponding to filename 16 and Etag The remote file corresponding to filename filename.part.0.parquet and Etag "76b9a0ed3044b29e8a326d6f4ade2036" no longer exists. no longer exists.During handling of the above exception, another exception occurred:Traceback (most recent call last): File "/python-docker/run.py", line 19, in service.start() File "/python-docker/chimera_ml_app/services.py", line 130, in start self.channel.start_consuming() File "/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py", line 1883, in start_consuming self._process_data_events(time_limit=None) File "/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py", line 2044, in _process_data_events self.connection.process_data_events(time_limit=time_limit) File "/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py", line 851, in process_data_events self._dispatch_channel_events() File "/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py", line 567, in _dispatch_channel_events impl_channel._get_cookie()._dispatch_events() File "/usr/local/lib/python3.9/site-packages/pika/adapters/blocking_connection.py", line 1510, in _dispatch_events consumer_info.on_message_callback(self, evt.method, File "/python-docker/chimera_ml_app/services.py", line 157, in on_request response = chimera.ml.predict(body=request) File "src/dependency_injector/_cwiring.pyx", line 28, in dependency_injector._cwiring._get_sync_patched._patched File "/python-docker/chimera_ml_app/ml.py", line 20, in predict predictions = predictor.predict() File "/usr/local/lib/python3.9/site-packages/hijacking_detection/pipelines/PipelinePredictorKATANA.py", line 125, in predict preprocessing.preprocess(path, save_path) File "/usr/local/lib/python3.9/site-packages/hijacking_detection/preprocessing/PreprocessingTestResample.py", line 55, in preprocess dataframe = future.result() File "/usr/local/lib/python3.9/site-packages/distributed/client.py", line 323, in result return self.client.sync(self._result, callback_timeout=timeout) File "/usr/local/lib/python3.9/site-packages/hijacking_detection/preprocessing/PreprocessingTestResample.py", line 91, in load_user df = dd.read_parquet('s3://{}/{}'.format(bucket, path), File "/usr/local/lib/python3.9/site-packages/dask/backends.py", line 143, in wrapper raise type(e)(TypeError: __init__() missing 1 required positional argument: 'e_tag'
```

**Environment**:

- Dask version: 2024.2.1
- Python version: 3.9.13
- Operating System: Docker running [python:3.9.17-slim-buster
](https://hub.docker.com/layers/library/python/3.9.17-slim-buster/images/sha256-d5cca64dca485c37ccf06721e36a93bf4331b0404bfd3ef2c7873867965359b7?context=explore)
- Install method (conda, pip, source): pip