RuntimeError: Barrier task with key 'shuffle-barrier-379659673650beace1aa63b2956e7b58' does not exist. This may be caused by task fusion during graph generation. Please let us know that you ran into this by leaving a comment at distributed#7816.

**Describe the issue**:

**Minimal Complete Verifiable Example**:

```python
# Put your MCVE code here
```

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

**Environment**:

- Dask version:
- Python version:
- Operating System:
- Install method (conda, pip, source):