The distributed.comm.compression parameter is ignored, in RPC comms as well as when spilling to disk.
The problem is that it is read at import time in distributed.protocol.compression, which is loaded early on.

Try adding to the module, just before get_default_compression:

def broken_compress(x):
    assert False

def broken_decompress(x):
    assert False

compressions["broken"] = {"compress": broken_compress, "decompress": broken_decompress}

and then:

@gen_cluster(client=True, config={"distributed.comm.compression": "broken"})
async def test_broken_compression(c, s, a, b):
    x = c.submit(inc, 0)
    await x  # does not raise

Note how this test is particularly though as the worker runs in the same process, so there's no chance to reload the module. However, the situation does not improve when using Nannies.

As a result, the compression is hardcoded to:

1. lz4, if installed; otherwise
2. snappy, if installed; otherwise
3. no compression

Existing tests that change the compression from the default should be revised too. e.g. test_core.py::test_compression is setting the wrong config parameter so it's not testing anything useful.

This issue impedes unit tests that rely on some compression to be available (namely, test_scheduler.py::test_memory), since our 3.7/3.8 CI install neither lz4 nor snappy and @gen_cluster(client=True, config={"distributed.comm.compression": "zlib"}) won't work.