botocore error when writing parquet to S3

I'm unable to write a particular dataframe to S3.

## Code overview
Read a parquet file from S3
```python
import dask.dataframe as dd
df = dd.read_parquet(path=[f's3://{bucket}/{path_to_file}'], # single file
kwargs={'dataset': {'schema': schema}}, # Tried with & without this
storage_options={'client_kwargs': s3_conf})
```
Extract rows with an na value to troubleshoot later
```python
null_df = df[df.isna().any(axis=1)]
```
Some columns that should be int have na values we can't directly cast. Find na values in columns that are supposed to be int, and replace them with a representative value (-1)
```python
null_df[col_names] = null_df[col_names].where(~null_df[col_names].isna(), -1.0)
```
Now types can be safely converted
```python
null_df = null_df.astype(dict_mapping_cols_to_np_types)
```
(I also replaced na values in float type columns with 0 to be safe).

Write the parquet file to a new S3 location
```python
# pyarrow_schema is the pyarrow equivalent to our numpy type mapping from earlier
null_df.to_parquet(path=f's3://{bucket}/{new_path}.parquet', engine='pyarrow', schema=pyarrow_schema, storage_options={'client_kwargs': s3_conf})
```

## The error
The primary resulting error:
```Traceback (most recent call last):
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/s3fs/core.py", line 248, in _call_s3
out = await method(**additional_kwargs)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/aiobotocore/client.py", line 155, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (MalformedACLError) when calling the CreateMultipartUpload operation: The XML you provided was not well-formed or did not validate against our published schema.
```

The full trace

Traceback (most recent call last):
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/s3fs/core.py", line 248, in _call_s3
out = await method(**additional_kwargs)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/aiobotocore/client.py", line 155, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (MalformedACLError) when calling the CreateMultipartUpload operation: The XML you provided was not well-formed or did not validate against our published schema.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/pyarrow/parquet.py", line 1987, in write_table
writer.write_table(table, row_group_size=row_group_size)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/pyarrow/parquet.py", line 693, in write_table
self.writer.write_table(table, row_group_size=row_group_size)
File "pyarrow/_parquet.pyx", line 1439, in pyarrow._parquet.ParquetWriter.write_table
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/fsspec/spec.py", line 1401, in write
self.flush()
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/fsspec/spec.py", line 1437, in flush
self._initiate_upload()
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/s3fs/core.py", line 1871, in _initiate_upload
self.mpu = self._call_s3(
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/s3fs/core.py", line 1863, in _call_s3
return self.fs.call_s3(method, self.s3_additional_kwargs, *kwarglist, **kwargs)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/fsspec/asyn.py", line 88, in wrapper
return sync(self.loop, func, *args, **kwargs)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/fsspec/asyn.py", line 69, in sync
raise result[0]
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/fsspec/asyn.py", line 25, in _runner
result[0] = await coro
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/s3fs/core.py", line 268, in _call_s3
raise err
OSError: [Errno 22] The XML you provided was not well-formed or did not validate against our published schema.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "src/main.py", line 379, in
handle_run_id(s3_client=s3_client,
File "/home/lz1sx9/project_repos/AVMMS_124381_Build_map-builder/features/post_processing/src/utils/decorators.py", line 35, in wrapper
return func(logger=logger, *args, **kwargs)
File "src/main.py", line 258, in handle_run_id
handle_geohash(s3_conf=s3_conf, run_id=run_id, bucket=bucket, geohash=geohash, all_files=all_files_carto,
File "/home/lz1sx9/project_repos/AVMMS_124381_Build_map-builder/features/post_processing/src/utils/decorators.py", line 35, in wrapper
return func(logger=logger, *args, **kwargs)
File "src/main.py", line 198, in handle_geohash
s3_utils.write_null_rows_to_s3(df_to_process=road_edges,
File "/home/lz1sx9/project_repos/AVMMS_124381_Build_map-builder/features/post_processing/src/utils/decorators.py", line 35, in wrapper
return func(logger=logger, *args, **kwargs)
File "/home/lz1sx9/project_repos/AVMMS_124381_Build_map-builder/features/post_processing/src/utils/s3_utils.py", line 167, in write_null_rows_to_s3
larger_df.to_parquet(path=f's3://{bucket}/feature_testing/cliff-testing/test.parquet',
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/dataframe/core.py", line 4540, in to_parquet
return to_parquet(self, path, *args, **kwargs)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/dataframe/io/parquet/core.py", line 725, in to_parquet
return compute_as_if_collection(
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/base.py", line 315, in compute_as_if_collection
return schedule(dsk2, keys, **kwargs)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/threaded.py", line 79, in get
results = get_async(
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/local.py", line 517, in get_async
raise_exception(exc, tb)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/local.py", line 325, in reraise
raise exc
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/local.py", line 223, in execute_task
result = _execute_task(task, data)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/core.py", line 121, in _execute_task
return func(*(_execute_task(a, cache) for a in args))
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/utils.py", line 35, in apply
return func(*args, **kwargs)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/dask/dataframe/io/parquet/arrow.py", line 946, in write_partition
pq.write_table(
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/pyarrow/parquet.py", line 1987, in write_table
writer.write_table(table, row_group_size=row_group_size)
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/pyarrow/parquet.py", line 678, in __exit__
self.close()
File "/home/lz1sx9/anaconda3/envs/dask/lib/python3.8/site-packages/pyarrow/parquet.py", line 700, in close
self._metadata_collector.append(self.writer.metadata)
File "pyarrow/_parquet.pyx", line 1453, in pyarrow._parquet.ParquetWriter.metadata.__get__
RuntimeError: file metadata is only available after writer close

## Other notes
I have no issues writing other dask dataframes to this location with the same approach. I can also split my dataframe with `null_df.random_split(frac=[.4, .6], random_state=1)`, and the smaller portion will write but not the larger. I've compared the smaller and larger by inspecting `df.columns`, `df.dtypes`, and `df.info` and see no differences.

The parquet file I originally read from S3 is 5.91 MB (not sure how that changes after reading to a local location), so I assume `null_df` is significantly smaller since it's a subset of the original.

## Environment

- Dask version: 2021.7.2, same result with 2021.09.1
- Python version: 3.8
- Operating System: Ubuntu 18.04.5 LTS
- Install method (conda, pip, source): pip from inside dedicated conda env