Revert change to wait_all_completed

Author: dandavison

temporalio/worker/_activity.py

@@ -201,10 +201,12 @@ async def drain_poll_queue(self) -> None:
 
     # Only call this after run()/drain_poll_queue() have returned. This will not
     # raise an exception.
+    # TODO(dan): based on the comment above it looks like the intention may have been to use
+    # return_exceptions=True
     async def wait_all_completed(self) -> None:
         running_tasks = [v.task for v in self._running_activities.values() if v.task]
         if running_tasks:
-            await asyncio.gather(*running_tasks, return_exceptions=True)
+            await asyncio.gather(*running_tasks, return_exceptions=False)
 
     def _handle_cancel_activity_task(
         self, task_token: bytes, cancel: temporalio.bridge.proto.activity_task.Cancel