`test_local_cluster_redundant_kwarg[True]` flaky

[gjoseph92]

https://github.com/dask/distributed/runs/6733192686

I notice this FIXME comment about LocalCluster leaking:

distributed/distributed/deploy/tests/test_local.py

Lines 1089 to 1090 in 6d85a85

	# FIXME: LocalCluster leaks if LocalCluster.__aenter__ raises
	await cluster.close()

And then the test is failing with coroutine 'InProc.write' was never awaited. @graingert think that's related?

cls = <class '_pytest.runner.CallInfo'>
func = <function call_runtest_hook.<locals>.<lambda> at 0x7f8add061790>
when = 'teardown'
reraise = (<class '_pytest.outcomes.Exit'>, <class 'KeyboardInterrupt'>)

    @classmethod
    def from_call(
        cls,
        func: "Callable[[], TResult]",
        when: "Literal['collect', 'setup', 'call', 'teardown']",
        reraise: Optional[
            Union[Type[BaseException], Tuple[Type[BaseException], ...]]
        ] = None,
    ) -> "CallInfo[TResult]":
        """Call func, wrapping the result in a CallInfo.
    
        :param func:
            The function to call. Called without arguments.
        :param when:
            The phase in which the function is called.
        :param reraise:
            Exception or exceptions that shall propagate if raised by the
            function, instead of being wrapped in the CallInfo.
        """
        excinfo = None
        start = timing.time()
        precise_start = timing.perf_counter()
        try:
>           result: Optional[TResult] = func()

/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py:338: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/runner.py:259: in <lambda>
    lambda: ihook(item=item, **kwds), when=when, reraise=reraise
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/_hooks.py:265: in __call__
    return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/pluggy/_manager.py:80: in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/unraisableexception.py:93: in pytest_runtest_teardown
    yield from unraisable_exception_runtest_hook()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    def unraisable_exception_runtest_hook() -> Generator[None, None, None]:
        with catch_unraisable_exception() as cm:
            yield
            if cm.unraisable:
                if cm.unraisable.err_msg is not None:
                    err_msg = cm.unraisable.err_msg
                else:
                    err_msg = "Exception ignored in"
                msg = f"{err_msg}: {cm.unraisable.object!r}\n\n"
                msg += "".join(
                    traceback.format_exception(
                        cm.unraisable.exc_type,
                        cm.unraisable.exc_value,
                        cm.unraisable.exc_traceback,
                    )
                )
>               warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
E               pytest.PytestUnraisableExceptionWarning: Exception ignored in: <coroutine object InProc.write at 0x7f8adbc22f40>
E               
E               Traceback (most recent call last):
E                 File "/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/warnings.py", line 506, in _warn_unawaited_coroutine
E                   warn(msg, category=RuntimeWarning, stacklevel=2, source=coro)
E               RuntimeWarning: coroutine 'InProc.write' was never awaited

/usr/share/miniconda3/envs/dask-distributed/lib/python3.8/site-packages/_pytest/unraisableexception.py:78: PytestUnraisableExceptionWarning

[gjoseph92]

Yup, pytest --count=1000 -x "distributed/deploy/tests/test_local.py::test_local_cluster_redundant_kwarg" fails in 5 tries for me on main