[Regression] Allow Client.map to accept futures as args and kwargs

distributed/client.py

@@ -892,7 +892,7 @@ def _layer(self) -> dict[Key, GraphNode]:
 
         if not self.kwargs:
             dsk = {
-                key: Task(key, self.func, *args)
+                key: Task(key, self.func, *parse_input(args))  # type: ignore[misc]
                 for key, args in zip(self.keys, zip(*self.iterables))
             }
 
@@ -907,12 +907,17 @@ def _layer(self) -> dict[Key, GraphNode]:
                 else:
                     kwargs2[k] = parse_input(v)
 
-                dsk.update(
-                    {
-                        key: Task(key, self.func, *args, **kwargs2)
-                        for key, args in zip(self.keys, zip(*self.iterables))
-                    }
-                )
+            dsk.update(
+                {
+                    key: Task(
+                        key,
+                        self.func,
+                        *parse_input(args),  # type: ignore[misc]
+                        **kwargs2,
+                    )
+                    for key, args in zip(self.keys, zip(*self.iterables))
+                }
+            )
         return dsk
 
 

distributed/tests/test_client.py

@@ -8397,3 +8397,26 @@ async def test_count_serialization(c, s, a, use_lambda):
     expected_global = expected_local + 1
     assert x.local_count <= expected_local
     assert x.global_count <= expected_global
+
+
+@pytest.mark.parametrize("use_kwarg", [False, "simple", "future"])
+@gen_cluster(client=True)
+async def test_map_accepts_nested_futures(c, s, a, b, use_kwarg):
+    def reducer(futs, *, offset=0, **kwargs):
+        return sum(futs) + offset
+
+    f1 = c.submit(lambda: 10)
+    f2 = c.submit(lambda: 20)
+    offset = None
+    if use_kwarg == "simple":
+        future = c.map(reducer, [[f1, f2]], foo=True)[0]
+    elif use_kwarg == "future":
+        offset = c.submit(lambda: 1)
+
+        future = c.map(reducer, [[f1, f2]], offset=offset)[0]
+
+    else:
+        future = c.map(reducer, [[f1, f2]])[0]
+
+    result = await future.result()
+    assert result == 30 if not offset else 31