Pub/Sub appears to duplicate messages

I am trying to use Pub/Sub to propagate log messages to the client. In particular propagating job-specific log messages to the client. I have a sort of working, but something is quite strange where the log messages get duplicated on multiple executions of the script.

Here is a script:

```py
import os
import dask
import logging
import dask.distributed
from dask.delayed import Delayed

def verbose_to_loglevel(c: int) -> int:
loglevel = logging.WARNING
if c == 1:
loglevel = logging.INFO
elif c >= 2:
loglevel = logging.DEBUG
return loglevel

def main():

verbosity = 1
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()

client = dask.distributed.Client(os.environ.get("PO_LATTICE_SCHEDULER"))

class ClusterLoggingHandler(logging.Handler):
def __init__(self, topic: str, level: int = logging.NOTSET) -> None:
super().__init__(level)
self._topic = topic
self._pub = None

def emit(self, record: logging.LogRecord) -> None:
if self._pub is None:
worker = dask.distributed.get_worker()
self._pub = dask.distributed.Pub(self._topic, worker=worker)
self._pub.put(record)

class TestGraph:

def __init__(self):
self.logger = logging.getLogger('TestGraph')

def graph(self):
logger = self.logger.getChild("graph")
return dask.delayed(self.dodelayed)(1)

def get_logger(self, name):
logger = self.logger.getChild(name)
logger.setLevel('DEBUG')
logger.addHandler(ClusterLoggingHandler('testtopic', logging.DEBUG))
return logger

def dodelayed(self, input_val: int) -> int:
logger = self.get_logger("dodelayed")
logger.info("LOGGING FROM DO DELAYED")
return input_val + 1

pub = dask.distributed.Pub('testtopic', client=client)
sub = dask.distributed.Sub('testtopic', client=client)

tg = TestGraph()
g = tg.graph()

fs = client.compute(g)
rs = client.gather(fs)
print(rs)

pub.put(None)

for record in sub:
if record is None:
break
print(record)

main()
```

If the above script was called `test.py`, and you set the `PO_LATTICE_SCHEDULER` to the dask distributed scheduler. On the first time it runs:

```
> python ./test.py
2

```

On the second time:

```
> python ./test.py
2

```

Then it just keeps building up. There are just more log records...