prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  BufferError: Local: Queue full
  I am receiving this error once every 2-4 days and I need to restart Sentry to fix it. This started after moving to the Docker version of Sentry.
  
  I never noticed this being an issue on 9.1.2 also with Clickhouse and Snuba running, but without Kafka.
  
  > https://observ.app/share/issue/4e4f208a500d48cc898770930706959a/
  
  I am not sure where to look / poke / monitor to see this queue that is being spoken of and how I can flush it / enlarge it if needed.
  
  `sentry queues list` showed all 0's so it's not looking like there is a massive backlog of events.
  
  Any help is appreciated!
  
  </issue>
  <code>
  [start of src/sentry/utils/pubsub.py]
  1 from __future__ import absolute_import
  2 
  3 import redis
  4 import logging
  5 
  6 from threading import Thread
  7 from six.moves.queue import Queue, Full
  8 
  9 
  10 class QueuedPublisherService(object):
  11     """
  12     A publisher that queues items locally and publishes them to a
  13     remote pubsub service on a background thread.
  14 
  15     Maintains a lossy internal queue for posting, will discard the
  16     value if the queue is full or not immediately available. Will also
  17     drop items if the publish operation to the remote service fails.
  18     """
  19 
  20     def __init__(self, publisher):
  21         self._started = False
  22         self.publisher = publisher
  23 
  24     def _start(self):
  25         if self._started:
  26             return True
  27 
  28         self.q = q = Queue(maxsize=100)
  29 
  30         def worker():
  31             while True:
  32                 (channel, key, value) = q.get()
  33                 try:
  34                     self.publisher.publish(channel, key=key, value=value)
  35                 except Exception as e:
  36                     logger = logging.getLogger("sentry.errors")
  37                     logger.debug("could not submit event to pubsub: %s" % e)
  38                 finally:
  39                     q.task_done()
  40 
  41         t = Thread(target=worker)
  42         t.setDaemon(True)
  43         t.start()
  44 
  45         self._started = True
  46         return True
  47 
  48     def publish(self, channel, value, key=None):
  49         if not self._start():
  50             return
  51 
  52         try:
  53             self.q.put((channel, key, value), block=False)
  54         except Full:
  55             return
  56 
  57 
  58 class RedisPublisher(object):
  59     def __init__(self, connection):
  60         self.rds = None if connection is None else redis.StrictRedis(**connection)
  61 
  62     def publish(self, channel, value, key=None):
  63         if self.rds is not None:
  64             self.rds.publish(channel, value)
  65 
  66 
  67 class KafkaPublisher(object):
  68     def __init__(self, connection, asynchronous=True):
  69         from confluent_kafka import Producer
  70 
  71         self.producer = Producer(connection or {})
  72         self.asynchronous = asynchronous
  73 
  74     def publish(self, channel, value, key=None):
  75         self.producer.produce(topic=channel, value=value, key=key)
  76         if not self.asynchronous:
  77             self.producer.flush()
  78
  [end of src/sentry/utils/pubsub.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/src/sentry/utils/pubsub.py b/src/sentry/utils/pubsub.py
  --- a/src/sentry/utils/pubsub.py
  +++ b/src/sentry/utils/pubsub.py
  @@ -73,5 +73,7 @@
   
       def publish(self, channel, value, key=None):
           self.producer.produce(topic=channel, value=value, key=key)
  -        if not self.asynchronous:
  +        if self.asynchronous:
  +            self.producer.poll(0)
  +        else:
               self.producer.flush()
  
verification_info: {"golden_diff": "diff --git a/src/sentry/utils/pubsub.py b/src/sentry/utils/pubsub.py\n--- a/src/sentry/utils/pubsub.py\n+++ b/src/sentry/utils/pubsub.py\n@@ -73,5 +73,7 @@\n \n     def publish(self, channel, value, key=None):\n         self.producer.produce(topic=channel, value=value, key=key)\n-        if not self.asynchronous:\n+        if self.asynchronous:\n+            self.producer.poll(0)\n+        else:\n             self.producer.flush()\n", "issue": "BufferError: Local: Queue full\nI am receiving this error once every 2-4 days and I need to restart Sentry to fix it. This started after moving to the Docker version of Sentry.\r\n\r\nI never noticed this being an issue on 9.1.2 also with Clickhouse and Snuba running, but without Kafka.\r\n\r\n> https://observ.app/share/issue/4e4f208a500d48cc898770930706959a/\r\n\r\nI am not sure where to look / poke / monitor to see this queue that is being spoken of and how I can flush it / enlarge it if needed.\r\n\r\n`sentry queues list` showed all 0's so it's not looking like there is a massive backlog of events.\r\n\r\nAny help is appreciated!\n", "before_files": [{"content": "from __future__ import absolute_import\n\nimport redis\nimport logging\n\nfrom threading import Thread\nfrom six.moves.queue import Queue, Full\n\n\nclass QueuedPublisherService(object):\n    \"\"\"\n    A publisher that queues items locally and publishes them to a\n    remote pubsub service on a background thread.\n\n    Maintains a lossy internal queue for posting, will discard the\n    value if the queue is full or not immediately available. Will also\n    drop items if the publish operation to the remote service fails.\n    \"\"\"\n\n    def __init__(self, publisher):\n        self._started = False\n        self.publisher = publisher\n\n    def _start(self):\n        if self._started:\n            return True\n\n        self.q = q = Queue(maxsize=100)\n\n        def worker():\n            while True:\n                (channel, key, value) = q.get()\n                try:\n                    self.publisher.publish(channel, key=key, value=value)\n                except Exception as e:\n                    logger = logging.getLogger(\"sentry.errors\")\n                    logger.debug(\"could not submit event to pubsub: %s\" % e)\n                finally:\n                    q.task_done()\n\n        t = Thread(target=worker)\n        t.setDaemon(True)\n        t.start()\n\n        self._started = True\n        return True\n\n    def publish(self, channel, value, key=None):\n        if not self._start():\n            return\n\n        try:\n            self.q.put((channel, key, value), block=False)\n        except Full:\n            return\n\n\nclass RedisPublisher(object):\n    def __init__(self, connection):\n        self.rds = None if connection is None else redis.StrictRedis(**connection)\n\n    def publish(self, channel, value, key=None):\n        if self.rds is not None:\n            self.rds.publish(channel, value)\n\n\nclass KafkaPublisher(object):\n    def __init__(self, connection, asynchronous=True):\n        from confluent_kafka import Producer\n\n        self.producer = Producer(connection or {})\n        self.asynchronous = asynchronous\n\n    def publish(self, channel, value, key=None):\n        self.producer.produce(topic=channel, value=value, key=key)\n        if not self.asynchronous:\n            self.producer.flush()\n", "path": "src/sentry/utils/pubsub.py"}]}