Repository: dask/distributed

Issue:
Dask Dashboard "More" tab not accessible from "Workers" tab
**What happened**:
When on the "Workers" page of the Dask Dashboard, the "More" tab is not fully accessible. 
It drops down but I can't navigate to any items past "Groups". Seems like there's a conflict with the underlaying information in the read/write columns of the workers. See screencast below.

https://user-images.githubusercontent.com/68642378/127451332-87049b74-85a5-48bc-94f9-9c7601b07679.mov

**What you expected to happen**:
To be able to click on items in the "More" tab like "CPU", "Cluster Memory", etc. etc.

**Minimal Complete Verifiable Example**:


```python
import coiled
import dask.dataframe as  dd
from dask.distributed import LocalCluster

client = LocalCluster()
client
```

**Anything else we need to know?**:

**Environment**:

- Dask/Distributed version: 2021.07.1
- Bokeh version: 2.3.3
- Python version: 3.8
- Operating System: OS 11.3.1
- Install method (conda, pip, source): conda

[EDIT]
- Browser: Brave 1.27.108, Chrome 92.0.45 , Firefox 90.0.2
- clearing the cache does not resolve the issue for me 

Discussion:

Patch:
diff --git a/distributed/http/static/css/base.css b/distributed/http/static/css/base.css
index 0cc3583a..78607c7a 100644
--- a/distributed/http/static/css/base.css
+++ b/distributed/http/static/css/base.css
@@ -102,6 +102,10 @@ body {
   right: 4px;
 }
 
+.bk-root .bk-data-table {
+  z-index: 0;
+}
+
 .content {
   width: 100%;
   height: 100%;
diff --git a/distributed/scheduler.py b/distributed/scheduler.py
index 566f149f..3025695b 100644
--- a/distributed/scheduler.py
+++ b/distributed/scheduler.py
@@ -1969,6 +1969,7 @@ class SchedulerState:
             ("processing", "erred"): self.transition_processing_erred,
             ("no-worker", "released"): self.transition_no_worker_released,
             ("no-worker", "waiting"): self.transition_no_worker_waiting,
+            ("no-worker", "memory"): self.transition_no_worker_memory,
             ("released", "forgotten"): self.transition_released_forgotten,
             ("memory", "forgotten"): self.transition_memory_forgotten,
             ("erred", "released"): self.transition_erred_released,
@@ -2450,6 +2451,42 @@ class SchedulerState:
                 pdb.set_trace()
             raise
 
+    def transition_no_worker_memory(
+        self, key, nbytes=None, type=None, typename: str = None, worker=None
+    ):
+        try:
+            ws: WorkerState = self._workers_dv[worker]
+            ts: TaskState = self._tasks[key]
+            recommendations: dict = {}
+            client_msgs: dict = {}
+            worker_msgs: dict = {}
+
+            if self._validate:
+                assert not ts._processing_on
+                assert not ts._waiting_on
+                assert ts._state == "no-worker"
+
+            self._unrunnable.remove(ts)
+
+            if nbytes is not None:
+                ts.set_nbytes(nbytes)
+
+            self.check_idle_saturated(ws)
+
+            _add_to_memory(
+                self, ts, ws, recommendations, client_msgs, type=type, typename=typename
+            )
+            ts.state = "memory"
+
+            return recommendations, client_msgs, worker_msgs
+        except Exception as e:
+            logger.exception(e)
+            if LOG_PDB:
+                import pdb
+
+                pdb.set_trace()
+            raise
+
     @ccall
     @exceptval(check=False)
     def decide_worker(self, ts: TaskState) -> WorkerState:
diff --git a/distributed/worker.py b/distributed/worker.py
index 42a08ff2..4ac7e7ec 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -2738,6 +2738,13 @@ class Worker(ServerNode):
             assert name
 
             if name in self.plugins:
+                warnings.warn(
+                    "Attempting to add a worker plugin with the same name as an already registered "
+                    f"plugin ({name}). Currently this results in no change and the previously registered "
+                    "plugin is not overwritten. This behavior is deprecated and in a future release "
+                    f"the previously registered {name} worker plugin will be overwritten.",
+                    category=FutureWarning,
+                )
                 return {"status": "repeat"}
             else:
                 self.plugins[name] = plugin


Test patch:
diff --git a/distributed/diagnostics/tests/test_worker_plugin.py b/distributed/diagnostics/tests/test_worker_plugin.py
index f3aa8cc6..7ae01f09 100644
--- a/distributed/diagnostics/tests/test_worker_plugin.py
+++ b/distributed/diagnostics/tests/test_worker_plugin.py
@@ -198,7 +198,8 @@ async def test_registering_with_name_arg(c, s, w):
     assert list(responses.values()) == [{"status": "OK"}]
 
     async with Worker(s.address, loop=s.loop):
-        responses = await c.register_worker_plugin(FooWorkerPlugin(), name="foo")
+        with pytest.warns(FutureWarning, match="worker plugin will be overwritten"):
+            responses = await c.register_worker_plugin(FooWorkerPlugin(), name="foo")
         assert list(responses.values()) == [{"status": "repeat"}] * 2
 
 
diff --git a/distributed/tests/test_scheduler.py b/distributed/tests/test_scheduler.py
index 536fd3fd..b8b51a27 100644
--- a/distributed/tests/test_scheduler.py
+++ b/distributed/tests/test_scheduler.py
@@ -3183,3 +3183,24 @@ async def test_worker_heartbeat_after_cancel(c, s, *workers):
 
     while any(w.tasks for w in workers):
         await asyncio.gather(*[w.heartbeat() for w in workers])
+
+
+@gen_cluster(client=True, nthreads=[("", 1)])
+async def test_worker_reconnect_task_memory_with_resources(c, s, a):
+    async with Worker(s.address, resources={"A": 1}) as b:
+        b.periodic_callbacks["heartbeat"].stop()
+
+        futs = c.map(inc, range(10), resources={"A": 1})
+        res = c.submit(sum, futs)
+
+        while not b.executing_count and not b.data:
+            await asyncio.sleep(0.001)
+
+        await s.remove_worker(address=b.address, close=False)
+        while not res.done():
+            await b.heartbeat()
+
+        await res
+        assert ("no-worker", "memory") in {
+            (start, finish) for (_, start, finish, _, _) in s.transition_log
+        }