test: add multiprocess helper test

Author: NguyenHoangSon96

influxdb_client_3/__init__.py

@@ -740,8 +740,10 @@ def flush(self):
 
     def close(self):
         """Close the client and clean up resources."""
-        self._write_api.close()
-        self._query_api.close()
+        if self._write_api is not None:
+            self._write_api.close()
+        if self._query_api is not None:
+            self._query_api.close()
 
     def __enter__(self):
         return self

influxdb_client_3/write_client/client/util/multiprocessing_helper.py

@@ -154,7 +154,16 @@ def run(self):
                                    error_callback=self.kwargs.get('error_callback', _error_callback),
                                    retry_callback=self.kwargs.get('retry_callback', _retry_callback)
                                    )
+
+        # Still need to create the InfluxDBClient3 because the init logics of InfluxDBClient3 will create the WriteApi.
+        # it will make WriteApi class created properly.
         self.client = InfluxDBClient3(write_client_options=wco, **self.kwargs)
+
+        # Close and set _query_api to None because query_api is not needed in this process.
+        # We only need write_api.
+        self.client._query_api.close()
+        self.client._query_api = None
+
         self.write_api = self.client._write_api
         # Infinite loop - until poison pill
         while True:

tests/test_influxdb_client_3_integration.py

@@ -309,7 +309,7 @@ def test_batch_write_closed(self):
             list_results = reader.to_pylist()
             self.assertEqual(data_size, len(list_results))
 
-    # @pytest.mark.skipif(running_on_circleci, reason="Skipping this test on CircleCI")
+    @pytest.mark.skipif(running_on_circleci, reason="Skipping this test on CircleCI")
     def test_multiprocessing_helper(self):
         org = 'my-org'
         writer = MultiprocessingWriter(