Better error when launching execution on misconfigured dask

[natekupp]

With a test job in test-dask.py:

from dagster import (
    ModeDefinition,
    pipeline,
    solid,
    RepositoryDefinition,
)
from dagster.core.definitions.executor import default_executors
from dagster_dask import dask_executor


@solid
def simple(_):
    return 1


@pipeline(mode_defs=[ModeDefinition(executor_defs=default_executors + [dask_executor])])
def dask_engine_pipeline():
    return simple()


def define_repo():
    return RepositoryDefinition(name='foobar', pipeline_defs=[dask_engine_pipeline])

and repository.yaml

repository:
  file: test-dask.py
  fn: define_repo

Then launching a dask scheduler and worker:

dask-scheduler
dask-worker localhost:8786

Running this pipeline in dagit with config:

storage:
  filesystem:
execution:
  dask:

will result in:

Traceback (most recent call last):
  File “/Users/nate/.pyenv/versions/3.7.4/envs/dagster37/lib/python3.7/site-packages/sqlalchemy/pool/base.py”, line 671, in _finalize_fairy
    fairy._reset(pool)
  File “/Users/nate/.pyenv/versions/3.7.4/envs/dagster37/lib/python3.7/site-packages/sqlalchemy/pool/base.py”, line 858, in _reset
    pool._dialect.do_rollback(self)
  File “/Users/nate/.pyenv/versions/3.7.4/envs/dagster37/lib/python3.7/site-packages/sqlalchemy/engine/default.py”, line 502, in do_rollback
    dbapi_connection.rollback()
sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 123145393209344 and this is thread id 4457528768.

When not specifying an address and running, it works fine, and launching a local dask scheduler/worker and specifying localhost:8786 address also works. But, not specifying an address w/ a running local cluster produces that error, and we should provide the user with a clearer error message