fix bug with retry for causing taskqueue to spin down (#745)

The early return caused `monitor_task.cancel()` not to be called which
would snowball into the taskqueue shutting down early and the task never
getting retried.

sdk/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "beta9"
-version = "0.1.126"
+version = "0.1.127"
 description = ""
 authors = ["beam.cloud <[email protected]>"]
 packages = [

sdk/src/beta9/runner/taskqueue.py

@@ -337,13 +337,16 @@ def process_tasks(self, channel: Channel) -> None:
                                 )
                             )
                             if not complete_task_response.ok:
+                                print("hello")
                                 raise RunnerException("Unable to end task")
 
                             if task_status == TaskStatus.Retry:
-                                print(
-                                    f"Retrying task <{task.id}> after {caught_exception} exception"
+                                message = (
+                                    complete_task_response.message
+                                    or f"Retrying task <{task.id}> after {caught_exception} exception"
                                 )
-                                return
+                                print(message)
+                                continue
 
                             print(f"Task completed <{task.id}>, took {duration}s")
                             send_callback(
@@ -352,7 +355,7 @@ def process_tasks(self, channel: Channel) -> None:
                                 payload=result or {},
                                 task_status=task_status,
                                 override_callback_url=kwargs.get("callback_url"),
-                            )  # Send callback to callback_url, if defined
+                            )
 
                         except BaseException:
                             print(traceback.format_exc())