Worker.close() leaves RPC channels dangling #6409
Description
Follow-up from #6371 and #6385.
CC @gjoseph92
test_deadlock_cancelled_after_inflight_before_gather_from_worker removes a worker while another worker has a task in flight from it.
The test calls
distributed/distributed/tests/test_worker.py
Lines 3282 to 3284 in 9bb999d
which in turn calls
distributed/distributed/scheduler.py
Lines 4172 to 4174 in 9bb999d
which in turrn calls
distributed/distributed/worker.py
Line 1576 in 9bb999d
Expected result
The RPC channel is explicitly shut down. The waiting gather_dep call on the other worker raises OSError within milliseconds.
Actual result
The RPC channel is left dangling until the TCP timeout kicks in (5s by default; in the test it's been shortened to 0.5s).