fix: prevent process deadlock if the batch write thread dies

This is one part of a fix for influxdata#558

It:

* Adds a WriteOption `max_close_wait` (default 500,000ms)
* Adjust `__del__` so that we'll only wait `max_close_wait`ms for queued writes to complete
* Adds a warning if the threshold is hit

Author: btasker

influxdb_client/client/write_api.py

@@ -51,6 +51,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
                  max_retry_delay=125_000,
                  max_retry_time=180_000,
                  exponential_base=2,
+                 max_close_wait=300_000,
                  write_scheduler=ThreadPoolScheduler(max_workers=1)) -> None:
         """
         Create write api configuration.
@@ -66,6 +67,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
         :param max_retry_delay: the maximum delay between each retry attempt in milliseconds
         :param max_retry_time: total timeout for all retry attempts in milliseconds, if 0 retry is disabled
         :param exponential_base: base for the exponential retry delay
+        :parama max_close_wait: the maximum time to wait for writes to be flushed if close() is called
         :param write_scheduler:
         """
         self.write_type = write_type
@@ -78,6 +80,7 @@ def __init__(self, write_type: WriteType = WriteType.batching,
         self.max_retry_time = max_retry_time
         self.exponential_base = exponential_base
         self.write_scheduler = write_scheduler
+        self.max_close_wait = max_close_wait
 
     def to_retry_strategy(self, **kwargs):
         """
@@ -410,9 +413,31 @@ def __del__(self):
             self._subject.dispose()
             self._subject = None
 
-            # Wait for finish writing
+            """
+            We impose a maximum wait time to ensure that we do not cause a deadlock if the
+            background thread has exited abnormally
+
+            Each iteration waits 100ms, but sleep expects the unit to be seconds so convert
+            the maximum wait time to seconds.
+
+            We keep a counter of how long we've waited
+            """
+            max_wait_time = self._write_options.max_close_wait / 1000
+            waited = 0
+            sleep_period = 0.1
+
+            # Wait for writing to finish
             while not self._disposable.is_disposed:
-                sleep(0.1)
+                sleep(sleep_period)
+                waited += sleep_period
+
+                # Have we reached the upper limit?
+                if waited >= max_wait_time:
+                    logger.warning(
+                            "Reached max_close_wait (%s seconds) waiting for batches to finish writing. Force closing",
+                            max_wait_time
+                        )
+                    break
 
         if self._disposable:
             self._disposable = None