asyncio exception when monitoring distributed compute

[data2code]

Describe the issue:
In the example code below, I try to monitor the progress of a cluster compute.
It does return the right results, but there is always an exception as shown below. This exception cannot be suppressed by try/except wrapper I add.

(1, 1, 1)
2023-11-30 21:27:09,396 - distributed.scheduler - ERROR -
Traceback (most recent call last):
File "/miniconda3/envs/X/lib/python3.9/site-packages/distributed/utils.py", line 832, in wrapper
return await func(*args, **kwargs)
File "/miniconda3/envs/X/lib/python3.9/site-packages/distributed/scheduler.py", line 7246, in feed
await asyncio.sleep(interval)
File "/miniconda3/envs/X/lib/python3.9/asyncio/tasks.py", line 652, in sleep
return await future
asyncio.exceptions.CancelledError

Minimal Complete Verifiable Example:

from time import sleep
import dask
from dask.distributed import Client,progress
from dask_jobqueue import SGECluster
from dask import compute, persist, delayed

@dask.delayed
def f():
    sleep(1)
    return 1

cluster = SGECluster(cores=1, memory='1GB')
cluster.scale(1)
c = Client(cluster)
out=[ f() ]*3

try:
    x=persist(*out)
    progress(x)
    out=compute(x)
except:
    print("error")

c.close()
print(out[0])

Anything else we need to know?:

Environment:

• Dask version: 2023.11.0
• Python version: 3.9.18
• Operating System: Red Hat Enterprise Linux release 8.5 (Ootpa)
• Install method (conda, pip, source):
  mamba install -c conda-forge dask
  mamba install -c conda-forge dask-jobqueue

[data2code]

I tried this barebone version, still got the exception that I cannot catch and cannot suppress. I have also reinstalled dask:
pip3 install "dask[complete]" --upgrade --force-reinstall

Really appreciate your insights.

from time import sleep
from dask.distributed import Client,progress
from dask_jobqueue import SGECluster

def f(x):
    sleep(1)
    return 1

cluster = SGECluster(cores=1, memory='1GB')
cluster.scale(3)
c = Client(cluster)

x=c.map(f, range(3))
progress(x)
out=c.gather(x)

c.close()
print(out)

Output:

[1, 1, 1]
2023-12-01 14:14:52,476 - distributed.scheduler - ERROR -
Traceback (most recent call last):
  File "/miniconda3/envs/X/lib/python3.9/site-packages/distributed/utils.py", line 832, in wrapper
    return await func(*args, **kwargs)
  File "/miniconda3/envs/X/lib/python3.9/site-packages/distributed/scheduler.py", line 7246, in feed
    await asyncio.sleep(interval)
  File "/miniconda3/envs/X/lib/python3.9/asyncio/tasks.py", line 652, in sleep
    return await future
asyncio.exceptions.CancelledError