deprecate passing plugin classes or kwargs

Fixes #5698

distributed/client.py

@@ -4601,9 +4601,6 @@ def register_worker_plugin(self, plugin=None, name=None, nanny=None, **kwargs):
             If plugin has no name attribute a random name is used.
         nanny : bool, optional
             Whether to register the plugin with workers or nannies.
-        **kwargs : optional
-            If you pass a class as the plugin, instead of a class instance, then the
-            class will be instantiated with any extra keyword arguments.
 
         Examples
         --------
@@ -4638,7 +4635,12 @@ class will be instantiated with any extra keyword arguments.
         distributed.WorkerPlugin
         unregister_worker_plugin
         """
-        if isinstance(plugin, type):
+        if isinstance(plugin, type) or kwargs:
+            warnings.warn(
+                "passing plugin classes or kwargs is deprecated; "
+                "pass a constructed plugin instance instead",
+                category=FutureWarning,
+            )
             plugin = plugin(**kwargs)
 
         if name is None:

distributed/diagnostics/tests/test_worker_plugin.py

@@ -1,4 +1,5 @@
 import asyncio
+import functools
 
 import pytest
 
@@ -83,19 +84,49 @@ async def test_remove_with_client_raises(c, s):
         await c.unregister_worker_plugin("bar")
 
 
+@pytest.mark.parametrize(
+    "plugin",
+    [
+        functools.partial(MyPlugin),
+        MyPlugin,
+    ],
+)
 @gen_cluster(client=True, nthreads=[])
-async def test_create_with_client_and_plugin_from_class(c, s):
-    await c.register_worker_plugin(MyPlugin, data=456)
+async def test_create_with_client_and_plugin_from_kwargs(c, s, plugin):
+    with pytest.warns(
+        FutureWarning,
+        match=r"passing plugin classes or kwargs is deprecated; "
+        "pass a constructed plugin instance instead",
+    ):
+        await c.register_worker_plugin(plugin, data=456)
 
     worker = await Worker(s.address, loop=s.loop)
     assert worker._my_plugin_status == "setup"
     assert worker._my_plugin_data == 456
 
     # Give the plugin a new name so that it registers
-    await c.register_worker_plugin(MyPlugin, name="new", data=789)
+    with pytest.warns(
+        FutureWarning,
+        match=r"passing plugin classes or kwargs is deprecated; "
+        "pass a constructed plugin instance instead",
+    ):
+        await c.register_worker_plugin(plugin, data=789, name="new")
     assert worker._my_plugin_data == 789
 
 
+@gen_cluster(nthreads=[("127.0.0.1", 1)], client=True)
+async def test_plugin_class_warns(c, s, w):
+    class EmptyPlugin:
+        pass
+
+    with pytest.warns(
+        FutureWarning,
+        match=r"passing plugin classes or kwargs is deprecated; "
+        "pass a constructed plugin instance instead",
+    ):
+        await c.register_worker_plugin(EmptyPlugin)
+
+
 @gen_cluster(client=True, worker_kwargs={"plugins": [MyPlugin(5)]})
 async def test_create_on_construction(c, s, a, b):
     assert len(a.plugins) == len(b.plugins) == 1
@@ -264,7 +295,7 @@ def transition(self, *args, **kwargs):
         def teardown(self, worker):
             del self.worker.foo
 
-    await c.register_worker_plugin(MyCustomPlugin)
+    await c.register_worker_plugin(MyCustomPlugin())
 
     assert w.foo == 0
 
@@ -287,7 +318,7 @@ def transition(self, *args, **kwargs):
         def teardown(self, worker):
             del self.worker.bar
 
-    await c.register_worker_plugin(MyCustomPlugin)
+    await c.register_worker_plugin(MyCustomPlugin())
 
     assert not hasattr(w, "foo")
     assert w.bar == 0