clear queue when socket.gaierror occurs

enjoy-binbin · enjoy-binbin · commit bb8efb8930c5 · 2021-04-24T14:36:30.000+08:00
diff --git a/fluent/asyncsender.py b/fluent/asyncsender.py
@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 
+import socket
 import threading
 from queue import Queue, Full, Empty
 
@@ -81,11 +82,7 @@ def close(self, flush=True):
                 return
             self._closed = True
             if not flush:
-                while True:
-                    try:
-                        self._queue.get(block=False)
-                    except Empty:
-                        break
+                self._clear_queue()
             self._queue.put(_TOMBSTONE)
             self._send_thread.join()
 
@@ -101,6 +98,13 @@ def queue_blocking(self):
     def queue_circular(self):
         return self._queue_circular
 
+    def _clear_queue(self):
+        while True:
+            try:
+                self._queue.get(block=False)
+            except Empty:
+                break
+
     def _send(self, bytes_):
         with self.lock:
             if self._closed:
@@ -120,8 +124,21 @@ def _send(self, bytes_):
 
             return True
 
+    def _send_internal(self, bytes_):
+        send_internal_result = super(FluentSender, self)._send_internal(bytes_)
+        if send_internal_result is False:
+            # when send_result is False, super() caught socket.error
+            # and assigned the error to self.last_error
+            if isinstance(self.last_error, socket.gaierror):
+                # clear the queue to avoid blocking and print the log
+                self._clear_queue()
+                self._queue.put(_TOMBSTONE)
+                print("%s. Please check address: (%s, %s)" % (str(self.last_error), self.host, self.port))
+
+        return send_internal_result
+
     def _send_loop(self):
-        send_internal = super(FluentSender, self)._send_internal
+        send_internal = self._send_internal
 
         try:
             while True:
