From c1f12bfc255eb7a9f541d02b468271f4ce77c04a Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Fri, 9 Oct 2026 01:02:37 +0400 Subject: [PATCH] fix: write the last batch when WriteApi is closed in an atexit handler Since Python 3.9 concurrent.futures shuts its thread pools down before atexit handlers run, so the batching pipeline could no longer schedule the final write and dropped the batch with "cannot schedule new futures after interpreter shutdown". When the write scheduler refuses the batch because the interpreter is shutting down, write it on the current thread instead of dropping it. Fixes #644 --- CHANGELOG.md | 1 + influxdb_client/client/write_api.py | 9 +++ tests/test_WriteApiBatching.py | 88 +++++++++++++++++++++++++++++ 3 files changed, 98 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca6775c6..5362caf6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ 1. [#714](https://github.com/influxdata/influxdb-client-python/pull/714): Implement `WriteApi.flush()` to force pending batch writes without disposing the API (mirrors C#/Java batch writer). 1. [#715](https://github.com/influxdata/influxdb-client-python/pull/715): Accept numeric zero for `RangeThreshold.min`/`max` and greater/lesser `value` (identity checks vs `None`, not truthiness); cover create-check serialization. 1. [#720](https://github.com/influxdata/influxdb-client-python/pull/720): Prevent the `api_client` class from raising exceptions while the Python interpreter clears the client state during shutdown. +1. [#733](https://github.com/influxdata/influxdb-client-python/pull/733): Write the last batch when the batching `WriteApi` is closed in an `atexit` handler instead of dropping it with "cannot schedule new futures after interpreter shutdown". ## 1.50.0 [2026-01-23] diff --git a/influxdb_client/client/write_api.py b/influxdb_client/client/write_api.py index 67892224..21c93e4e 100644 --- a/influxdb_client/client/write_api.py +++ b/influxdb_client/client/write_api.py @@ -194,6 +194,11 @@ def _body_reduce(batch_items): return b'\n'.join(map(lambda batch_item: batch_item.data, batch_items)) +def _is_interpreter_shutdown(exception: Exception) -> bool: + # e.g. "cannot schedule new futures after interpreter shutdown" + return isinstance(exception, RuntimeError) and 'interpreter shutdown' in str(exception) + + class WriteApi(_BaseWriteApi): """ Implementation for '/api/v2/write' endpoint. @@ -548,6 +553,10 @@ def _to_response(self, data: _BatchItem, delay: timedelta): ops.subscribe_on(self._write_options.write_scheduler), # use delay if its specified ops.delay(duetime=delay, scheduler=self._write_options.write_scheduler), + # since Python 3.9 the thread pools stop accepting work before 'atexit' handlers run, + # so write the batch on the current thread instead of dropping it + ops.catch(handler=lambda exception, source: rx.just(data) if _is_interpreter_shutdown(exception) + else rx.throw(exception)), # invoke http call ops.map(lambda x: self._http(x)), # catch exception to fail batch response diff --git a/tests/test_WriteApiBatching.py b/tests/test_WriteApiBatching.py index ad453c2a..cb4c91a1 100644 --- a/tests/test_WriteApiBatching.py +++ b/tests/test_WriteApiBatching.py @@ -2,7 +2,12 @@ from __future__ import absolute_import +import http.server +import os +import subprocess import sys +import textwrap +import threading import time import unittest from collections import namedtuple @@ -804,5 +809,88 @@ def test_flush_synchronous_is_noop(self): self._write_client.flush() +class BatchingWriteInterpreterShutdownTest(unittest.TestCase): + """ + Closing the batching WriteApi in an ``atexit`` handler writes the last batch. + + Since Python 3.9 the thread pools used by the batching pipeline stop accepting work before ``atexit`` + handlers run, so the scenario runs in a separate interpreter against a local HTTP server. + """ + + def setUp(self) -> None: + self.requests = [] + requests = self.requests + + class WriteHandler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + requests.append(self.rfile.read(int(self.headers['Content-Length'])).decode('utf-8')) + self.send_response(204) + self.end_headers() + + def log_message(self, format, *args): + pass + + self.httpd = http.server.ThreadingHTTPServer(('127.0.0.1', 0), WriteHandler) + self.httpd_thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self.httpd_thread.start() + + def tearDown(self) -> None: + self.httpd.shutdown() + self.httpd.server_close() + + def _run_script(self, script): + return subprocess.run([sys.executable, '-c', textwrap.dedent(script), + 'http://127.0.0.1:%d' % self.httpd.server_address[1]], + cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=60) + + def test_close_in_atexit(self): + result = self._run_script(""" + import atexit + import sys + + from influxdb_client import InfluxDBClient + + client = InfluxDBClient(url=sys.argv[1], token="my-token", org="my-org") + write_api = client.write_api() + + @atexit.register + def close(): + print("closing") + write_api.close() + client.close() + + write_api.write("my-bucket", record="h2o_feet,location=coyote_creek water_level=1 1") + """) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertEqual(["h2o_feet,location=coyote_creek water_level=1 1"], self.requests, result.stderr) + self.assertIn("closing", result.stdout) + self.assertNotIn("cannot schedule new futures", result.stderr) + + def test_write_and_close_in_atexit(self): + result = self._run_script(""" + import atexit + import sys + + from influxdb_client import InfluxDBClient + + client = InfluxDBClient(url=sys.argv[1], token="my-token", org="my-org") + write_api = client.write_api() + + @atexit.register + def close(): + print("closing") + write_api.write("my-bucket", record="h2o_feet,location=coyote_creek water_level=2 2") + write_api.close() + client.close() + """) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertEqual(["h2o_feet,location=coyote_creek water_level=2 2"], self.requests, result.stderr) + self.assertIn("closing", result.stdout) + self.assertNotIn("cannot schedule new futures", result.stderr) + + if __name__ == '__main__': unittest.main()