Write API Batching mode does not flush when closing

[alexmreis]

Specifications

• Client Version: 1.31.0
• InfluxDB Version: InfluxDB Cloud (runtime.version() returns v0.184.2)
• Platform: Linux

Using the python SDK in batching mode:

1. Sending Point object data
2. The client is closed before the flush interval has passed
3. No data is sent to influx.

This is probably due to flush() not being implemented, nor invoked from __del__() or close(), despite the PyDoc documentation saying otherwise (!!!)

Code sample to reproduce problem

class InfluxWriter:
    def open(self, partition_id, epoch_id):
        print("========> Opening connection to Influx")
        
        token = os.getenv("INFLUXDB_TOKEN")
        org = os.getenv("INFLUXDB_ORG")
        url = os.getenv("INFLUXDB_URL")
        bucket = os.getenv("INFLUXDB_BUCKET")

        self.org = org
        self.bucket = bucket

        self.client = influxdb_client.InfluxDBClient(url=url, token=token, org=org)
        self.write_api = self.client.write_api(write_type = WriteType.batching)

        return True

    def process(self, row):
        logging.warning("Sending data to influx %s", row)
        point = Point("test_measurement_1")
        value = row

        point.time = value["window_end"]
        point.field("count", value["count"])

        self.write_api.write(bucket=self.bucket, org=self.org, record=point)

    def close(self, error):        
        logging.warning("Closing connection to influx")
        self.write_api.close()
        self.client.close()

writer = InfluxWriter()
writer.open()
writer.process({"window_end": datetime.now(), "count": 1})
writer.close()

Expected behavior

When closing the write API, all data pending in the current batch should be written out to influx in blocking fashion.

Actual behavior

No data is sent to influx due to the flush timer not being triggered.

Additional info

No response

[bednar]

Hi @alexmreis,

thanks for using our client.

I've tested your code by:

import codecs
import logging
from datetime import datetime

from influxdb_client import WritePrecision, InfluxDBClient, Point
from influxdb_client.client import influxdb_client
from influxdb_client.client.write_api import SYNCHRONOUS, WriteType


class InfluxWriter:
    def __init__(self):
        self.write_api = None
        self.client = None
        self.bucket = None
        self.org = None

    def open(self):
        print("========> Opening connection to Influx")

        token = "my-token"
        org = "my-org"
        url = "http://localhost:8086"
        bucket = "my-bucket"

        self.org = org
        self.bucket = bucket

        self.client = influxdb_client.InfluxDBClient(url=url, token=token, org=org, debug=False)
        self.write_api = self.client.write_api(write_type=WriteType.batching)

        return True

    def process(self, row):
        logging.warning("Sending data to influx %s", row)
        point = Point("test_measurement_1")
        value = row

        point.time = value["window_end"]
        point.field("count", value["count"])

        self.write_api.write(bucket=self.bucket, org=self.org, record=point)

    def close(self):
        logging.warning("Closing connection to influx")
        self.write_api.close()
        self.client.close()


writer = InfluxWriter()
writer.open()
writer.process({"window_end": datetime.now(), "count": 1})
writer.close()

And everything works correctly:

Try to enable debug mode by influxdb_client.InfluxDBClient(url=url, token=token, org=org, debug=True) to see ingesting data into InfluxDB.

Regards

[bednar]

This issue has been closed because it has not had recent activity. Please reopen if this issue is still important to you and you have additionally information.