KeyError: pickle-protocol when running df.npartitions

Running the following:
```
import dask.dataframe as dd
cols = ['Year', 'Month', 'DayOfWeek', 'Distance',
'DepDelay', 'CRSDepTime', 'UniqueCarrier', 'Origin', 'Dest']
df = dd.read_csv('/tmp/2000.csv', usecols=cols)
df.head()
df.npartitions
```

Get the following error:

```
2020-09-28 09:12:59,186 ERROR tornado.application - Exception in callback functools.partial(. at 0x7f3573436dd0>, exception=ValueError('Your Dask versions may not be in sync. Please ensure that you have the same version of dask and distributed on your client, scheduler, and worker machines')>)
Traceback (most recent call last):
File "/opt/sarvi/sig/devMiniconda483/anaconda/envs/notebookEnv/lib/python3.7/site-packages/distributed/comm/core.py", line 141, in handshake_configuration
print(remote["pickle-protocol"])
KeyError: 'pickle-protocol'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/opt/sarvi/sig/devMiniconda483/anaconda/envs/notebookEnv/lib/python3.7/site-packages/tornado/ioloop.py", line 743, in _run_callback
ret = callback()
File "/opt/sarvi/sig/devMiniconda483/anaconda/envs/notebookEnv/lib/python3.7/site-packages/tornado/tcpserver.py", line 327, in
gen.convert_yielded(future), lambda f: f.result()
File "/opt/sarvi/sig/devMiniconda483/anaconda/envs/notebookEnv/lib/python3.7/site-packages/distributed/comm/tcp.py", line 457, in _handle_stream
await self.on_connection(comm)
File "/opt/sarvi/sig/devMiniconda483/anaconda/envs/notebookEnv/lib/python3.7/site-packages/distributed/comm/core.py", line 236, in on_connection
comm.local_info, comm.remote_info
File "/opt/sarvi/sig/devMiniconda483/anaconda/envs/notebookEnv/lib/python3.7/site-packages/distributed/comm/core.py", line 149, in handshake_configuration
"Your Dask versions may not be in sync. "
ValueError: Your Dask versions may not be in sync. Please ensure that you have the same version of dask and distributed on your client, scheduler, and worker machines
```

**Anything else we need to know?**:
- I am using a single-host environment - client, scheduler and worker are all on the same host using the same conda environment
- I am using a custom implementation of a dask-jobqueue cluster. Tried to reproduce with LocalCluster, but could not
- The cluster works fine using Dask/Distributed 2.22.0 and lower, but starts encountering this error from 2.23.0.

**Environment**:

- Dask version: 2.23.0 and up
- Python version: 3.7.6
- Operating System: Redhat 7.8
- Install method (conda, pip, source): conda