AttributeError: 'S3FileSystem' object has no attribute '_loop'

So here is my code to store a xarray.Dataset as zarr in s3 bucket

def store_xarray_as_zarr(
        dataset: xarray.Dataset,
        bucket_name: str,
        dataset_name: str,
        data_type_mapping: Dict[str, str],
        append_dim: str = "dt_calc",
        time_dimensions: Optional[List[str]] = None,
        storage_class: str = "GLACIER_IR",
) -> None:
    """
    Takes xarray.Dataset object and stores it in the given s3 bucket


    Args:
        dataset: xarray.Dataset objects contains the data
        bucket_name: name of the bucket on s3
        dataset_name: refined name of the dataset, can be a path too!
        data_type_mapping: is the final mapping to the dtype for the given data
        append_dim: dimension to append data along zarr archive
        time_dimensions: all time dimensions will be stored as float64 and has to be encoded additionally
        storage_class: 'STANDARD'|'REDUCED_REDUNDANCY'|'STANDARD_IA'|'ONEZONE_IA'|'INTELLIGENT_TIERING'|
                       'GLACIER'|'DEEP_ARCHIVE'|'OUTPOSTS'|'GLACIER_IR'

    Returns:
        None, stores data in s3
    """
    if time_dimensions is None:
        time_dimensions = ["dt_calc", "dt_fore"]
    check_aws_env_vars()

    s3_out = s3fs.S3FileSystem(
        anon=False, s3_additional_kwargs={"StorageClass": storage_class}
    )
    store_out = s3fs.S3Map(
        root=f"s3:///{bucket_name}/{dataset_name}.zarr", s3=s3_out, check=False
    )

    try:
        create_bucket(bucket_name)
        compressor = zarr.Blosc(cname="zstd", clevel=6, shuffle=2)
        encoding = {
            key: {"dtype": data_type_mapping[key], "compressor": compressor}
            for key in list(dataset.keys())
            if key not in time_dimensions
        }
        for time_dimension in time_dimensions:
            encoding.update({time_dimension: {"dtype": "float64"}})
        dataset.to_zarr(
            store_out, mode="w-", encoding=encoding, compute=True, consolidated=True
        )
    except zarr.errors.ContainsGroupError:
        dataset.to_zarr(
            store_out, mode="a", append_dim=append_dim, compute=True, consolidated=True
        )

Actually the script fails at this point:

        dataset.to_zarr(
            store_out, mode="w-", encoding=encoding, compute=True, consolidated=True
        )

With this error message:

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
/app/src/radar_data/german_radar_data_download.py in <module>
----> 1 process_radoland_data_upload(datetime(2022, 1, 1), datetime(2022, 2, 1))

/app/src/radar_data/german_radar_data_download.py in process_radoland_data_upload(dt_start, dt_end)
    113                 DwdRadarParameter.RADOLAN_CDC
    114             )
--> 115             store_xarray_as_zarr(
    116                 ds,
    117                 'alitiq-radolan-hourly',

/usr/local/lib/python3.9/site-packages/alitiq_db/utils/aws/s3.py in store_xarray_as_zarr(dataset, bucket_name, dataset_name, data_type_mapping, append_dim, time_dimensions, storage_class)
    255         for time_dimension in time_dimensions:
    256             encoding.update({time_dimension: {"dtype": "float64"}})
--> 257         dataset.to_zarr(
    258             store_out, mode="w-", encoding=encoding, compute=True, consolidated=True
    259         )

/usr/local/lib/python3.9/site-packages/xarray/core/dataset.py in to_zarr(self, store, chunk_store, mode, synchronizer, group, encoding, compute, consolidated, append_dim, region, safe_chunks, storage_options)
   2034             encoding = {}
   2035 
-> 2036         return to_zarr(
   2037             self,
   2038             store=store,

/usr/local/lib/python3.9/site-packages/xarray/backends/api.py in to_zarr(dataset, store, chunk_store, mode, synchronizer, group, encoding, compute, consolidated, append_dim, region, safe_chunks, storage_options)
   1389         already_consolidated = False
   1390         consolidate_on_close = consolidated or consolidated is None
-> 1391     zstore = backends.ZarrStore.open_group(
   1392         store=mapper,
   1393         mode=mode,

/usr/local/lib/python3.9/site-packages/xarray/backends/zarr.py in open_group(cls, store, mode, synchronizer, group, consolidated, consolidate_on_close, chunk_store, storage_options, append_dim, write_region, safe_chunks, stacklevel)
    368             zarr_group = zarr.open_consolidated(store, **open_kwargs)
    369         else:
--> 370             zarr_group = zarr.open_group(store, **open_kwargs)
    371         return cls(
    372             zarr_group,

/usr/local/lib/python3.9/site-packages/zarr/hierarchy.py in open_group(store, mode, cache_attrs, synchronizer, path, chunk_store, storage_options)
   1192 
   1193     elif mode in ['w-', 'x']:
-> 1194         if contains_array(store, path=path):
   1195             raise ContainsArrayError(path)
   1196         elif contains_group(store, path=path):

/usr/local/lib/python3.9/site-packages/zarr/storage.py in contains_array(store, path)
     94     prefix = _path_to_prefix(path)
     95     key = prefix + array_meta_key
---> 96     return key in store
     97 
     98 

/usr/local/lib/python3.9/_collections_abc.py in __contains__(self, key)
    682     def __contains__(self, key):
    683         try:
--> 684             self[key]
    685         except KeyError:
    686             return False

/usr/local/lib/python3.9/site-packages/zarr/storage.py in __getitem__(self, key)
    543 
    544     def __getitem__(self, key):
--> 545         return self._mutable_mapping[key]
    546 
    547     def __setitem__(self, key, value):

/usr/local/lib/python3.9/site-packages/fsspec/mapping.py in __getitem__(self, key, default)
    135         k = self._key_to_str(key)
    136         try:
--> 137             result = self.fs.cat(k)
    138         except self.missing_exceptions:
    139             if default is not None:

/usr/local/lib/python3.9/site-packages/fsspec/asyn.py in wrapper(*args, **kwargs)
     84     def wrapper(*args, **kwargs):
     85         self = obj or args[0]
---> 86         return sync(self.loop, func, *args, **kwargs)
     87 
     88     return wrapper

/usr/local/lib/python3.9/site-packages/fsspec/asyn.py in loop(self)
    299         if self._pid != os.getpid():
    300             raise RuntimeError("This class is not fork-safe")
--> 301         return self._loop
    302 
    303     async def _rm_file(self, path, **kwargs):

AttributeError: 'S3FileSystem' object has no attribute '_loop'

Unfortunately my test is running fine, so it is difficult to find out whats the issue:

    ds = xarray.Dataset(
        {'temp': (('dt_calc', 'y', 'x'), np.array([[[1., 2., 3., 4.], [3., 4., 5., 6.]]]))},
        coords={'lon': ('y', np.array([50., 51.])), 'lat': ('x', np.array([4., 5., 6., 7.])),
                'dt_calc': ('dt_calc', [datetime(2022, 1, 1)])}
    )
    ds_2 = xarray.Dataset(
        {'temp': (('dt_calc', 'y', 'x'), np.array([[[1., 2., 3., 4.], [3., 4., 5., 6.]]]))},
        coords={'lon': ('y', np.array([50., 51.])), 'lat': ('x', np.array([4., 5., 6., 7.])),
                'dt_calc': ('dt_calc', [datetime(2022, 1, 1, 1)])}
    )
    store_xarray_as_zarr(ds, 'alitiq-test-bucket-db', '202206/test', {'temp': ds.temp.dtype},
                         time_dimensions=['dt_calc'])
    store_xarray_as_zarr(ds_2, 'alitiq-test-bucket-db', '202206/test', {'temp': ds.temp.dtype},
                         time_dimensions=['dt_calc'])

This is the head of my dataset:

<xarray.Dataset>
Dimensions:        (time: 744, y: 900, x: 900)
Coordinates:
    lat            (y, x) float64 46.95 46.95 46.95 46.96 ... 54.73 54.73 54.73
    lon            (y, x) float64 3.589 3.601 3.613 3.625 ... 15.67 15.69 15.7
  * time           (time) datetime64[ns] 2022-01-01T00:50:00 ... 2022-01-31T2...
Dimensions without coordinates: y, x
Data variables:
    precipitation  (time, y, x) float64 nan nan nan nan nan ... nan nan nan nan

Thanks a lot for your support !

Here is my env:
boto3==1.21.10
aiobotocore==2.3.0
botocore==1.24.21
s3fs==2022.5.0
zarr==2.11.3
xarray==2022.3.0

Start with the traceback through fsspec/asyn.py and the S3FileSystem construction in the supplied function, then reproduce the minimal xarray-to-Zarr example using the listed boto3, aiobotocore, botocore, s3fs, zarr, and xarray versions. Compare the failing dataset path with the small passing test and verify that writing to S3 no longer raises AttributeError: 'S3FileSystem' object has no attribute '_loop'.