dask distributed client.gather() loses reference to dataset across client sessions

Running from jupyterhub on dask distributed cluster generated with dask helm chart `dask-3.1.0` and app version `1.1.5`.

```
import dask.array as da
from distributed import Client

client = Client()
def get_array(arr_len):
return da.from_array([num for num in range(arr_len)], chunks=(arr_len))

future = client.submit(get_array, 10)
client.publish_dataset(**{'future': future})

print(client.list_datasets())

print(client.get_dataset('future').result())

client.gather(client.get_dataset('future'))
```

Above code works fine. Then kill the jupyterhub kernel and retrieve the dataset.

```
import dask.array as da
from distributed import Client

client = Client()
print(client.list_datasets())

print(client.get_dataset('future').result())

client.gather(client.get_dataset('future'))
```

`result()` returns the future result as expected, but the `gather` call results in exception. I found [this issue](https://github.com/dask/distributed/issues/2124) which has the same stack trace so I checked my worker, scheduler, and jupyterhub library versions and they are the same.

`client.get_versions(check=True)`

Here's the stack trace:
```
---------------------------------------------------------------------------
KeyError Traceback (most recent call last)
/opt/conda/lib/python3.7/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
1486 try:
-> 1487 st = self.futures[key]
1488 exception = st.exception

KeyError: 'get_array-719de5775564cba1d78cf935ef8a59f7'

During handling of the above exception, another exception occurred:

CancelledError Traceback (most recent call last)
in
----> 1 client.gather(client.get_dataset('future'))

/opt/conda/lib/python3.7/site-packages/distributed/client.py in gather(self, futures, errors, maxsize, direct, asynchronous)
1654 return self.sync(self._gather, futures, errors=errors,
1655 direct=direct, local_worker=local_worker,
-> 1656 asynchronous=asynchronous)
1657
1658 @gen.coroutine

/opt/conda/lib/python3.7/site-packages/distributed/client.py in sync(self, func, *args, **kwargs)
674 return future
675 else:
--> 676 return sync(self.loop, func, *args, **kwargs)
677
678 def __repr__(self):

/opt/conda/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs)
280 e.wait(10)
281 if error[0]:
--> 282 six.reraise(*error[0])
283 else:
284 return result[0]

/opt/conda/lib/python3.7/site-packages/six.py in reraise(tp, value, tb)
691 if value.__traceback__ is not tb:
692 raise value.with_traceback(tb)
--> 693 raise value
694 finally:
695 value = None

/opt/conda/lib/python3.7/site-packages/distributed/utils.py in f()
265 if timeout is not None:
266 future = gen.with_timeout(timedelta(seconds=timeout), future)
--> 267 result[0] = yield future
268 except Exception as exc:
269 error[0] = sys.exc_info()

/opt/conda/lib/python3.7/site-packages/tornado/gen.py in run(self)
727
728 try:
--> 729 value = future.result()
730 except Exception:
731 exc_info = sys.exc_info()

/opt/conda/lib/python3.7/site-packages/tornado/gen.py in run(self)
740 exc_info = None
741 else:
--> 742 yielded = self.gen.send(value)
743
744 except (StopIteration, Return) as e:

/opt/conda/lib/python3.7/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
1491 six.reraise(CancelledError,
1492 CancelledError(key),
-> 1493 None)
1494 else:
1495 six.reraise(type(exception),

/opt/conda/lib/python3.7/site-packages/six.py in reraise(tp, value, tb)
691 if value.__traceback__ is not tb:
692 raise value.with_traceback(tb)
--> 693 raise value
694 finally:
695 value = None

CancelledError: get_array-719de5775564cba1d78cf935ef8a59f7
```

I ran `client.get_versions(check=True)` with no Exception:
```
{'scheduler': {'host': (('python', '3.7.2.final.0'),
('python-bits', 64),
('OS', 'Linux'),
('OS-release', '4.14.138+'),
('machine', 'x86_64'),
('processor', ''),
('byteorder', 'little'),
('LC_ALL', 'C.UTF-8'),
('LANG', 'C.UTF-8'),
('LOCALE', 'en_US.UTF-8')),
'packages': {'required': (('dask', '1.1.5'),
('distributed', '1.26.1'),
('msgpack', '0.6.1'),
('cloudpickle', '0.8.1'),
('tornado', '6.0.2'),
('toolz', '0.9.0')),
'optional': (('numpy', '1.16.2'),
('pandas', '0.24.2'),
('bokeh', '1.0.4'),
('lz4', None),
('dask_ml', None),
('blosc', '1.8.1'))}},
'workers': {'tcp://10.16.2.6:43607': {'host': (('python', '3.7.2.final.0'),
('python-bits', 64),
('OS', 'Linux'),
('OS-release', '4.14.138+'),
('machine', 'x86_64'),
('processor', ''),
('byteorder', 'little'),
('LC_ALL', 'C.UTF-8'),
('LANG', 'C.UTF-8'),
('LOCALE', 'en_US.UTF-8')),
'packages': {'required': (('dask', '1.1.5'),
('distributed', '1.26.1'),
('msgpack', '0.6.1'),
('cloudpickle', '0.8.1'),
('tornado', '6.0.2'),
('toolz', '0.9.0')),
'optional': (('numpy', '1.16.2'),
('pandas', '0.24.2'),
('bokeh', '1.0.4'),
('lz4', None),
('dask_ml', None),
('blosc', '1.8.1'))}}},
'client': {'host': [('python', '3.7.3.final.0'),
('python-bits', 64),
('OS', 'Linux'),
('OS-release', '4.14.138+'),
('machine', 'x86_64'),
('processor', 'x86_64'),
('byteorder', 'little'),
('LC_ALL', 'en_US.UTF-8'),
('LANG', 'en_US.UTF-8'),
('LOCALE', 'en_US.UTF-8')],
'packages': {'required': [('dask', '1.1.5'),
('distributed', '1.26.1'),
('msgpack', '0.6.1'),
('cloudpickle', '0.8.1'),
('tornado', '6.0.2'),
('toolz', '0.9.0')],
'optional': [('numpy', '1.16.2'),
('pandas', '0.24.2'),
('bokeh', '1.0.4'),
('lz4', None),
('dask_ml', None),
('blosc', '1.8.1')]}}}
```