prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  PubSub: Default thread-based subscriber Policy's on_response() implementation degenerates into series of synchronous calls because of debug statement
  https://github.com/GoogleCloudPlatform/google-cloud-python/blob/4e17158c0f14e0e91cbeb74c3e8483c5a231e9ac/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py#L147
  
  While testing out the new PubSub API (v 0.28.3), I noticed that the default thread-based `Policy`, which uses a `ThreadPoolExecutor(max_workers=10)` seemed to be executing the subscriber callbacks sequentially, instead of asynchronously.  Upon introspection of the source, it looks like the above referenced code:
  ```
  ...
  future = self._executor.submit(self._callback, message)
  logger.debug('Result: %s' % future.result())
  ```
  will block until `self._callback` completes, because of the call to `future.result()`.  Or am I losing my mind?  Thanks!
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py`
  Content:
  ```
  1 # Copyright 2017, Google Inc. All rights reserved.
  2 #
  3 # Licensed under the Apache License, Version 2.0 (the "License");
  4 # you may not use this file except in compliance with the License.
  5 # You may obtain a copy of the License at
  6 #
  7 #     http://www.apache.org/licenses/LICENSE-2.0
  8 #
  9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 
  15 from __future__ import absolute_import
  16 
  17 from concurrent import futures
  18 from queue import Queue
  19 import logging
  20 import threading
  21 
  22 import grpc
  23 
  24 from google.cloud.pubsub_v1 import types
  25 from google.cloud.pubsub_v1.subscriber import _helper_threads
  26 from google.cloud.pubsub_v1.subscriber.policy import base
  27 from google.cloud.pubsub_v1.subscriber.message import Message
  28 
  29 
  30 logger = logging.getLogger(__name__)
  31 
  32 
  33 class Policy(base.BasePolicy):
  34     """A consumer class based on :class:`threading.Thread`.
  35 
  36     This consumer handles the connection to the Pub/Sub service and all of
  37     the concurrency needs.
  38     """
  39     def __init__(self, client, subscription, flow_control=types.FlowControl(),
  40                  executor=None, queue=None):
  41         """Instantiate the policy.
  42 
  43         Args:
  44             client (~.pubsub_v1.subscriber.client): The subscriber client used
  45                 to create this instance.
  46             subscription (str): The name of the subscription. The canonical
  47                 format for this is
  48                 ``projects/{project}/subscriptions/{subscription}``.
  49             flow_control (~google.cloud.pubsub_v1.types.FlowControl): The flow
  50                 control settings.
  51             executor (~concurrent.futures.ThreadPoolExecutor): (Optional.) A
  52                 ThreadPoolExecutor instance, or anything duck-type compatible
  53                 with it.
  54             queue (~queue.Queue): (Optional.) A Queue instance, appropriate
  55                 for crossing the concurrency boundary implemented by
  56                 ``executor``.
  57         """
  58         # Default the callback to a no-op; it is provided by `.open`.
  59         self._callback = lambda message: None
  60 
  61         # Create a queue for keeping track of shared state.
  62         if queue is None:
  63             queue = Queue()
  64         self._request_queue = queue
  65 
  66         # Call the superclass constructor.
  67         super(Policy, self).__init__(
  68             client=client,
  69             flow_control=flow_control,
  70             subscription=subscription,
  71         )
  72 
  73         # Also maintain a request queue and an executor.
  74         logger.debug('Creating callback requests thread (not starting).')
  75         if executor is None:
  76             executor = futures.ThreadPoolExecutor(max_workers=10)
  77         self._executor = executor
  78         self._callback_requests = _helper_threads.QueueCallbackThread(
  79             self._request_queue,
  80             self.on_callback_request,
  81         )
  82 
  83     def close(self):
  84         """Close the existing connection."""
  85         # Close the main subscription connection.
  86         self._consumer.helper_threads.stop('callback requests worker')
  87         self._consumer.stop_consuming()
  88 
  89     def open(self, callback):
  90         """Open a streaming pull connection and begin receiving messages.
  91 
  92         For each message received, the ``callback`` function is fired with
  93         a :class:`~.pubsub_v1.subscriber.message.Message` as its only
  94         argument.
  95 
  96         Args:
  97             callback (Callable): The callback function.
  98         """
  99         # Start the thread to pass the requests.
  100         logger.debug('Starting callback requests worker.')
  101         self._callback = callback
  102         self._consumer.helper_threads.start(
  103             'callback requests worker',
  104             self._request_queue,
  105             self._callback_requests,
  106         )
  107 
  108         # Actually start consuming messages.
  109         self._consumer.start_consuming()
  110 
  111         # Spawn a helper thread that maintains all of the leases for
  112         # this policy.
  113         logger.debug('Spawning lease maintenance worker.')
  114         self._leaser = threading.Thread(target=self.maintain_leases)
  115         self._leaser.daemon = True
  116         self._leaser.start()
  117 
  118     def on_callback_request(self, callback_request):
  119         """Map the callback request to the appropriate GRPC request."""
  120         action, kwargs = callback_request[0], callback_request[1]
  121         getattr(self, action)(**kwargs)
  122 
  123     def on_exception(self, exception):
  124         """Bubble the exception.
  125 
  126         This will cause the stream to exit loudly.
  127         """
  128         # If this is DEADLINE_EXCEEDED, then we want to retry.
  129         # That entails just returning None.
  130         deadline_exceeded = grpc.StatusCode.DEADLINE_EXCEEDED
  131         if getattr(exception, 'code', lambda: None)() == deadline_exceeded:
  132             return
  133 
  134         # Raise any other exception.
  135         raise exception
  136 
  137     def on_response(self, response):
  138         """Process all received Pub/Sub messages.
  139 
  140         For each message, schedule a callback with the executor.
  141         """
  142         for msg in response.received_messages:
  143             logger.debug('New message received from Pub/Sub: %r', msg)
  144             logger.debug(self._callback)
  145             message = Message(msg.message, msg.ack_id, self._request_queue)
  146             future = self._executor.submit(self._callback, message)
  147             logger.debug('Result: %s' % future.result())
  148
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py b/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py
  --- a/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py
  +++ b/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py
  @@ -30,6 +30,11 @@
   logger = logging.getLogger(__name__)
   
   
  +def _callback_completed(future):
  +    """Simple callback that just logs a `Future`'s result."""
  +    logger.debug('Result: %s', future.result())
  +
  +
   class Policy(base.BasePolicy):
       """A consumer class based on :class:`threading.Thread`.
   
  @@ -144,4 +149,4 @@
               logger.debug(self._callback)
               message = Message(msg.message, msg.ack_id, self._request_queue)
               future = self._executor.submit(self._callback, message)
  -            logger.debug('Result: %s' % future.result())
  +            future.add_done_callback(_callback_completed)
  
verification_info: {"golden_diff": "diff --git a/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py b/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py\n--- a/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py\n+++ b/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py\n@@ -30,6 +30,11 @@\n logger = logging.getLogger(__name__)\n \n \n+def _callback_completed(future):\n+    \"\"\"Simple callback that just logs a `Future`'s result.\"\"\"\n+    logger.debug('Result: %s', future.result())\n+\n+\n class Policy(base.BasePolicy):\n     \"\"\"A consumer class based on :class:`threading.Thread`.\n \n@@ -144,4 +149,4 @@\n             logger.debug(self._callback)\n             message = Message(msg.message, msg.ack_id, self._request_queue)\n             future = self._executor.submit(self._callback, message)\n-            logger.debug('Result: %s' % future.result())\n+            future.add_done_callback(_callback_completed)\n", "issue": "PubSub: Default thread-based subscriber Policy's on_response() implementation degenerates into series of synchronous calls because of debug statement\nhttps://github.com/GoogleCloudPlatform/google-cloud-python/blob/4e17158c0f14e0e91cbeb74c3e8483c5a231e9ac/pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py#L147\r\n\r\nWhile testing out the new PubSub API (v 0.28.3), I noticed that the default thread-based `Policy`, which uses a `ThreadPoolExecutor(max_workers=10)` seemed to be executing the subscriber callbacks sequentially, instead of asynchronously.  Upon introspection of the source, it looks like the above referenced code:\r\n```\r\n...\r\nfuture = self._executor.submit(self._callback, message)\r\nlogger.debug('Result: %s' % future.result())\r\n```\r\nwill block until `self._callback` completes, because of the call to `future.result()`.  Or am I losing my mind?  Thanks!\n", "before_files": [{"content": "# Copyright 2017, Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\n\nfrom concurrent import futures\nfrom queue import Queue\nimport logging\nimport threading\n\nimport grpc\n\nfrom google.cloud.pubsub_v1 import types\nfrom google.cloud.pubsub_v1.subscriber import _helper_threads\nfrom google.cloud.pubsub_v1.subscriber.policy import base\nfrom google.cloud.pubsub_v1.subscriber.message import Message\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass Policy(base.BasePolicy):\n    \"\"\"A consumer class based on :class:`threading.Thread`.\n\n    This consumer handles the connection to the Pub/Sub service and all of\n    the concurrency needs.\n    \"\"\"\n    def __init__(self, client, subscription, flow_control=types.FlowControl(),\n                 executor=None, queue=None):\n        \"\"\"Instantiate the policy.\n\n        Args:\n            client (~.pubsub_v1.subscriber.client): The subscriber client used\n                to create this instance.\n            subscription (str): The name of the subscription. The canonical\n                format for this is\n                ``projects/{project}/subscriptions/{subscription}``.\n            flow_control (~google.cloud.pubsub_v1.types.FlowControl): The flow\n                control settings.\n            executor (~concurrent.futures.ThreadPoolExecutor): (Optional.) A\n                ThreadPoolExecutor instance, or anything duck-type compatible\n                with it.\n            queue (~queue.Queue): (Optional.) A Queue instance, appropriate\n                for crossing the concurrency boundary implemented by\n                ``executor``.\n        \"\"\"\n        # Default the callback to a no-op; it is provided by `.open`.\n        self._callback = lambda message: None\n\n        # Create a queue for keeping track of shared state.\n        if queue is None:\n            queue = Queue()\n        self._request_queue = queue\n\n        # Call the superclass constructor.\n        super(Policy, self).__init__(\n            client=client,\n            flow_control=flow_control,\n            subscription=subscription,\n        )\n\n        # Also maintain a request queue and an executor.\n        logger.debug('Creating callback requests thread (not starting).')\n        if executor is None:\n            executor = futures.ThreadPoolExecutor(max_workers=10)\n        self._executor = executor\n        self._callback_requests = _helper_threads.QueueCallbackThread(\n            self._request_queue,\n            self.on_callback_request,\n        )\n\n    def close(self):\n        \"\"\"Close the existing connection.\"\"\"\n        # Close the main subscription connection.\n        self._consumer.helper_threads.stop('callback requests worker')\n        self._consumer.stop_consuming()\n\n    def open(self, callback):\n        \"\"\"Open a streaming pull connection and begin receiving messages.\n\n        For each message received, the ``callback`` function is fired with\n        a :class:`~.pubsub_v1.subscriber.message.Message` as its only\n        argument.\n\n        Args:\n            callback (Callable): The callback function.\n        \"\"\"\n        # Start the thread to pass the requests.\n        logger.debug('Starting callback requests worker.')\n        self._callback = callback\n        self._consumer.helper_threads.start(\n            'callback requests worker',\n            self._request_queue,\n            self._callback_requests,\n        )\n\n        # Actually start consuming messages.\n        self._consumer.start_consuming()\n\n        # Spawn a helper thread that maintains all of the leases for\n        # this policy.\n        logger.debug('Spawning lease maintenance worker.')\n        self._leaser = threading.Thread(target=self.maintain_leases)\n        self._leaser.daemon = True\n        self._leaser.start()\n\n    def on_callback_request(self, callback_request):\n        \"\"\"Map the callback request to the appropriate GRPC request.\"\"\"\n        action, kwargs = callback_request[0], callback_request[1]\n        getattr(self, action)(**kwargs)\n\n    def on_exception(self, exception):\n        \"\"\"Bubble the exception.\n\n        This will cause the stream to exit loudly.\n        \"\"\"\n        # If this is DEADLINE_EXCEEDED, then we want to retry.\n        # That entails just returning None.\n        deadline_exceeded = grpc.StatusCode.DEADLINE_EXCEEDED\n        if getattr(exception, 'code', lambda: None)() == deadline_exceeded:\n            return\n\n        # Raise any other exception.\n        raise exception\n\n    def on_response(self, response):\n        \"\"\"Process all received Pub/Sub messages.\n\n        For each message, schedule a callback with the executor.\n        \"\"\"\n        for msg in response.received_messages:\n            logger.debug('New message received from Pub/Sub: %r', msg)\n            logger.debug(self._callback)\n            message = Message(msg.message, msg.ack_id, self._request_queue)\n            future = self._executor.submit(self._callback, message)\n            logger.debug('Result: %s' % future.result())\n", "path": "pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py"}], "after_files": [{"content": "# Copyright 2017, Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\n\nfrom concurrent import futures\nfrom queue import Queue\nimport logging\nimport threading\n\nimport grpc\n\nfrom google.cloud.pubsub_v1 import types\nfrom google.cloud.pubsub_v1.subscriber import _helper_threads\nfrom google.cloud.pubsub_v1.subscriber.policy import base\nfrom google.cloud.pubsub_v1.subscriber.message import Message\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef _callback_completed(future):\n    \"\"\"Simple callback that just logs a `Future`'s result.\"\"\"\n    logger.debug('Result: %s', future.result())\n\n\nclass Policy(base.BasePolicy):\n    \"\"\"A consumer class based on :class:`threading.Thread`.\n\n    This consumer handles the connection to the Pub/Sub service and all of\n    the concurrency needs.\n    \"\"\"\n    def __init__(self, client, subscription, flow_control=types.FlowControl(),\n                 executor=None, queue=None):\n        \"\"\"Instantiate the policy.\n\n        Args:\n            client (~.pubsub_v1.subscriber.client): The subscriber client used\n                to create this instance.\n            subscription (str): The name of the subscription. The canonical\n                format for this is\n                ``projects/{project}/subscriptions/{subscription}``.\n            flow_control (~google.cloud.pubsub_v1.types.FlowControl): The flow\n                control settings.\n            executor (~concurrent.futures.ThreadPoolExecutor): (Optional.) A\n                ThreadPoolExecutor instance, or anything duck-type compatible\n                with it.\n            queue (~queue.Queue): (Optional.) A Queue instance, appropriate\n                for crossing the concurrency boundary implemented by\n                ``executor``.\n        \"\"\"\n        # Default the callback to a no-op; it is provided by `.open`.\n        self._callback = lambda message: None\n\n        # Create a queue for keeping track of shared state.\n        if queue is None:\n            queue = Queue()\n        self._request_queue = queue\n\n        # Call the superclass constructor.\n        super(Policy, self).__init__(\n            client=client,\n            flow_control=flow_control,\n            subscription=subscription,\n        )\n\n        # Also maintain a request queue and an executor.\n        logger.debug('Creating callback requests thread (not starting).')\n        if executor is None:\n            executor = futures.ThreadPoolExecutor(max_workers=10)\n        self._executor = executor\n        self._callback_requests = _helper_threads.QueueCallbackThread(\n            self._request_queue,\n            self.on_callback_request,\n        )\n\n    def close(self):\n        \"\"\"Close the existing connection.\"\"\"\n        # Close the main subscription connection.\n        self._consumer.helper_threads.stop('callback requests worker')\n        self._consumer.stop_consuming()\n\n    def open(self, callback):\n        \"\"\"Open a streaming pull connection and begin receiving messages.\n\n        For each message received, the ``callback`` function is fired with\n        a :class:`~.pubsub_v1.subscriber.message.Message` as its only\n        argument.\n\n        Args:\n            callback (Callable): The callback function.\n        \"\"\"\n        # Start the thread to pass the requests.\n        logger.debug('Starting callback requests worker.')\n        self._callback = callback\n        self._consumer.helper_threads.start(\n            'callback requests worker',\n            self._request_queue,\n            self._callback_requests,\n        )\n\n        # Actually start consuming messages.\n        self._consumer.start_consuming()\n\n        # Spawn a helper thread that maintains all of the leases for\n        # this policy.\n        logger.debug('Spawning lease maintenance worker.')\n        self._leaser = threading.Thread(target=self.maintain_leases)\n        self._leaser.daemon = True\n        self._leaser.start()\n\n    def on_callback_request(self, callback_request):\n        \"\"\"Map the callback request to the appropriate GRPC request.\"\"\"\n        action, kwargs = callback_request[0], callback_request[1]\n        getattr(self, action)(**kwargs)\n\n    def on_exception(self, exception):\n        \"\"\"Bubble the exception.\n\n        This will cause the stream to exit loudly.\n        \"\"\"\n        # If this is DEADLINE_EXCEEDED, then we want to retry.\n        # That entails just returning None.\n        deadline_exceeded = grpc.StatusCode.DEADLINE_EXCEEDED\n        if getattr(exception, 'code', lambda: None)() == deadline_exceeded:\n            return\n\n        # Raise any other exception.\n        raise exception\n\n    def on_response(self, response):\n        \"\"\"Process all received Pub/Sub messages.\n\n        For each message, schedule a callback with the executor.\n        \"\"\"\n        for msg in response.received_messages:\n            logger.debug('New message received from Pub/Sub: %r', msg)\n            logger.debug(self._callback)\n            message = Message(msg.message, msg.ack_id, self._request_queue)\n            future = self._executor.submit(self._callback, message)\n            future.add_done_callback(_callback_completed)\n", "path": "pubsub/google/cloud/pubsub_v1/subscriber/policy/thread.py"}]}