Edge case in log event

[abergou]

What happened:
This is a funny edge case that someone ran into. For various reasons the user connected two clients to the same dask cluster from one process. He had one of these clients subscribed to an event topic and the other one was not. Because the client id is dervied from the machine and pid, two clients from the same process will end up with the same id. When topics are logged the dask scheduler the scheduler sent messages about the topic two both clients (since they have the same id). The the client that wasn't subscribed to the message blew up with a ValueError.

What you expected to happen:
For the client that didn't subscribe to topic to to ignore the message or not receive it.

Minimal Complete Verifiable Example:

>>> from dask import delayed
>>> from distributed import Client, LocalCluster, get_worker
>>> @delayed
>>> def test():
>>> 	worker = get_worker()
>>> 	worker.log_event("test", "this is a test")
>>> cluster = LocalCluster(n_workers=1, threads_per_worker=1)
>>> client1 = Client(cluster)
>>> client1.subscribe_topic("test", print)
>>> client1.compute(test(), sync=True)
>>> client2 = Client(cluster)
>>> client2.compute(test(), sync=True)
(1648836108.3501477, 'this is a test')
(1648836108.375489, 'this is a test')
No event handler known for topic test.
Traceback (most recent call last):
  File "/usr/local/python/python-3.9/std/lib64/python3.9/site-packages/distributed/client.py", line 1224, in _handle_report
    await result
  File "/usr/local/python/python-3.9/std/lib64/python3.9/site-packages/distributed/client.py", line 3778, in _handle_event
    self.unsubscribe_topic(topic)
  File "/usr/local/python/python-3.9/std/lib64/python3.9/site-packages/distributed/client.py", line 3829, in unsubscribe_topic
    raise ValueError(f"No event handler known for topic {topic}.")
ValueError: No event handler known for topic test.

Anything else we need to know?:
Looking through the code there's two potential issues that I can spot:

1. two clients from the same process can end up with the same id -- this seems intentional though I'm not sure why the client isn't simply a singleton from within a process if this is intended.
2. from the logic of client.py:_handle_event it's not clear whether the ValueError is intentional or just a bug. It looks like the client that receives the unintended message tries to ignore the message, and it tries to unsubscribe itself from receiving further messages, but since it never subscribed and doesn't have an active handler it raises a ValueError.

Environment:

• Dask version: 2022.03.0
• Python version: 3.9
• Operating System: Linux
• Install method (conda, pip, source): pip