Repository: jupyter/jupyter_client (python)

Issue:
Notebook execution started hanging in jupyter_client 8
Hello,

Starting last week, a test in the Dagster papermill / jupyter integration started hanging. The hang coincided with a number of jupyter-related releases, but we were able to isolate the problem to the `jupyter-client` 8 release - downgrading to the previous release made the problem go away.

To reproduce the problem:
- Check out the dagster github repo at tag 1.1.13, with a python 3.8 venv
- Install tox
- In the `python_modules/libraries/dagstermill` folder, run:
- `tox -vv -e py38 -- -k test_hello_world -x -s`

The test will hang while executing a notebook, but the hang will go away if you force the previous version of jupyter-client to be installed instead.

Code in Dagster that executes the notebook: https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagstermill/dagstermill/factory.py#L194-L213

Console output during the hang (this shows forever):
```
Executing:  50%|█████     | 1/2 [00:03<00:03,  3.18s/cell]
```

Stack trace of the notebook while execution is hanging:
```
Thread 0x10D504600 (idle): "MainThread"
    select (selectors.py:558)
    _run_once (asyncio/base_events.py:1823)
    run_forever (asyncio/base_events.py:570)
    run_until_complete (asyncio/base_events.py:603)
    just_run (nbclient/util.py:57)
    wrapped (nbclient/util.py:78)
    execute (papermill/clientwrap.py:46)
    execute_managed_notebook (dagstermill/engine.py:75)
    execute_notebook (papermill/engines.py:343)
    execute_notebook_with_engine (papermill/engines.py:49)
    execute_notebook (papermill/execute.py:107)
    _t_fn (dagstermill/asset_factory.py:86)
    iterate_with_context (dagster/_utils/__init__.py:457)
    _yield_compute_results (dagster/_core/execution/plan/compute.py:145)
    execute_core_compute (dagster/_core/execution/plan/compute.py:177)
    _step_output_error_checked_user_event_sequence (dagster/_core/execution/plan/execute_step.py:94)
    core_dagster_event_sequence_for_step (dagster/_core/execution/plan/execute_step.py:382)
    dagster_event_sequence_for_step (dagster/_core/execution/plan/execute_plan.py:265)
    inner_plan_execution_iterator (dagster/_core/execution/plan/execute_plan.py:114)
    __iter__ (dagster/_core/execution/api.py:1103)
    execute (dagster/_core/executor/multiprocess.py:93)
    _execute_command_in_child_process (dagster/_core/executor/child_process_executor.py:79)
    run (multiprocessing/process.py:108)
    _bootstrap (multiprocessing/process.py:315)
    _main (multiprocessing/spawn.py:129)
    spawn_main (multiprocessing/spawn.py:116)
    <module> (<string>:1)
```

Let me know if any other information would be helpful.

Pull request description:
Add papermill downstream check and fix kernel client replies
Fixes #922

Patch:
diff --git a/.github/workflows/downstream.yml b/.github/workflows/downstream.yml
index f901353..50c79d6 100644
--- a/.github/workflows/downstream.yml
+++ b/.github/workflows/downstream.yml
@@ -31,6 +31,16 @@ jobs:
           package_name: nbclient
           env_values: IPYKERNEL_CELL_NAME=\<IPY-INPUT\>
 
+  papermill:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v3
+      - uses: jupyterlab/maintainer-tools/.github/actions/base-setup@v1
+      - uses: jupyterlab/maintainer-tools/.github/actions/downstream-test@v1
+        with:
+          package_name: papermill
+
   nbconvert:
     runs-on: ubuntu-latest
     timeout-minutes: 15
@@ -115,3 +125,20 @@ jobs:
         run: |
           cd ${GITHUB_WORKSPACE}/../qtconsole
           xvfb-run --auto-servernum ${pythonLocation}/bin/python -m pytest -x -vv -s --full-trace --color=yes qtconsole
+
+  downstreams_check: # This job does nothing and is only used for the branch protection
+    if: always()
+    needs:
+      - ipykernel
+      - nbclient
+      - papermill
+      - nbconvert
+      - jupyter_server
+      - jupyter_kernel_test
+      - qtconsole
+    runs-on: ubuntu-latest
+    steps:
+      - name: Decide whether the needed jobs succeeded or failed
+        uses: re-actors/alls-green@release/v1
+        with:
+          jobs: ${{ toJSON(needs) }}
diff --git a/jupyter_client/asynchronous/client.py b/jupyter_client/asynchronous/client.py
index 53b5889..8f7b082 100644
--- a/jupyter_client/asynchronous/client.py
+++ b/jupyter_client/asynchronous/client.py
@@ -1,7 +1,6 @@
 """Implements an async kernel client"""
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
-import asyncio
 
 import zmq.asyncio
 from traitlets import Instance, Type
@@ -17,10 +16,8 @@ def wrapped(meth, channel):
         reply = kwargs.pop("reply", False)
         timeout = kwargs.pop("timeout", None)
         msg_id = meth(self, *args, **kwargs)
-        fut: asyncio.Future = asyncio.Future()
-        fut.set_result(msg_id)
         if not reply:
-            return fut
+            return msg_id
         return self._recv_reply(msg_id, timeout=timeout, channel=channel)
 
     return _


Test patch:
diff --git a/tests/test_client.py b/tests/test_client.py
index 8dc3574..65da426 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -147,37 +147,37 @@ class TestAsyncKernelClient:
         assert called
 
     async def test_history(self, kc):
-        msg_id = await kc.history(session=0)
+        msg_id = kc.history(session=0)
         assert isinstance(msg_id, str)
         reply = await kc.history(session=0, reply=True, timeout=TIMEOUT)
         self._check_reply("history", reply)
 
     async def test_inspect(self, kc):
-        msg_id = await kc.inspect("who cares")
+        msg_id = kc.inspect("who cares")
         assert isinstance(msg_id, str)
         reply = await kc.inspect("code", reply=True, timeout=TIMEOUT)
         self._check_reply("inspect", reply)
 
     async def test_complete(self, kc):
-        msg_id = await kc.complete("who cares")
+        msg_id = kc.complete("who cares")
         assert isinstance(msg_id, str)
         reply = await kc.complete("code", reply=True, timeout=TIMEOUT)
         self._check_reply("complete", reply)
 
     async def test_is_complete(self, kc):
-        msg_id = await kc.is_complete("who cares")
+        msg_id = kc.is_complete("who cares")
         assert isinstance(msg_id, str)
         reply = await kc.is_complete("code", reply=True, timeout=TIMEOUT)
         self._check_reply("is_complete", reply)
 
     async def test_kernel_info(self, kc):
-        msg_id = await kc.kernel_info()
+        msg_id = kc.kernel_info()
         assert isinstance(msg_id, str)
         reply = await kc.kernel_info(reply=True, timeout=TIMEOUT)
         self._check_reply("kernel_info", reply)
 
     async def test_comm_info(self, kc):
-        msg_id = await kc.comm_info()
+        msg_id = kc.comm_info()
         assert isinstance(msg_id, str)
         reply = await kc.comm_info(reply=True, timeout=TIMEOUT)
         self._check_reply("comm_info", reply)
@@ -187,7 +187,7 @@ class TestAsyncKernelClient:
         self._check_reply("shutdown", reply)
 
     async def test_shutdown_id(self, kc):
-        msg_id = await kc.shutdown()
+        msg_id = kc.shutdown()
         assert isinstance(msg_id, str)
 
 
diff --git a/tests/test_kernelmanager.py b/tests/test_kernelmanager.py
index cfce5b5..1f52ab2 100644
--- a/tests/test_kernelmanager.py
+++ b/tests/test_kernelmanager.py
@@ -199,7 +199,7 @@ class TestKernelManager:
         km, kc = await jp_start_kernel("signaltest")
 
         async def execute(cmd):
-            request_id = await kc.execute(cmd)
+            request_id = kc.execute(cmd)
             while True:
                 reply = await kc.get_shell_msg(TIMEOUT)
                 if reply["parent_header"]["msg_id"] == request_id:
@@ -241,7 +241,7 @@ class TestKernelManager:
 
     async def _env_test_body(self, kc):
         async def execute(cmd):
-            request_id = await kc.execute(cmd)
+            request_id = kc.execute(cmd)
             while True:
                 reply = await kc.get_shell_msg(TIMEOUT)
                 if reply["parent_header"]["msg_id"] == request_id:
@@ -452,7 +452,7 @@ class TestAsyncKernelManager:
         km, kc = await jp_start_kernel("signaltest")
 
         async def execute(cmd):
-            request_id = await kc.execute(cmd)
+            request_id = kc.execute(cmd)
             while True:
                 reply = await kc.get_shell_msg(TIMEOUT)
                 if reply["parent_header"]["msg_id"] == request_id:
@@ -472,7 +472,7 @@ class TestAsyncKernelManager:
         assert reply["user_expressions"]["poll"] == [None] * N
 
         # start a job on the kernel to be interrupted
-        request_id = await kc.execute("sleep")
+        request_id = kc.execute("sleep")
         await asyncio.sleep(1)  # ensure sleep message has been handled before we interrupt
         await km.interrupt_kernel()
         while True:
diff --git a/tests/test_multikernelmanager.py b/tests/test_multikernelmanager.py
index 882529a..fc82d54 100644
--- a/tests/test_multikernelmanager.py
+++ b/tests/test_multikernelmanager.py
@@ -637,7 +637,7 @@ class TestAsyncKernelManager(AsyncTestCase):
 
         stream.on_recv(record_activity)
         while True:
-            await client.kernel_info()
+            await client.kernel_info(reply=True)
             if called:
                 break
             await asyncio.sleep(0.1)