[jira] [Created] (CAMEL-25610) camel-influxdb2 - the producer completes the exchange before anything is written; a write that InfluxDB rejects or cannot receive is only logged by the client and the route continues as if the data was stored

["shashank (Jira)" <ji...@apache.org>]

shashank created CAMEL-25610:
--------------------------------

             Summary: camel-influxdb2 - the producer completes the exchange before anything is written; a write that InfluxDB rejects or cannot receive is only logged by the client and the route continues as if the data was stored
                 Key: CAMEL-25610
                 URL: https://issues.apache.org/jira/browse/CAMEL-25610
             Project: Camel
          Issue Type: Bug
          Components: camel-influxdb
            Reporter: shashank
            Assignee: shashank


{{InfluxDb2Producer}} creates its write API with {{connection.makeWriteApi()}} (line 48 at main a7169d7c6ab6). That is the batching {{WriteApi}} of influxdb-client-java: {{writePoint}}, {{writeMeasurement}}, {{writeRecord}}, {{writePoints}}, {{writeMeasurements}} and {{writeRecords}} (lines 102, 115, 128, 142, 156, 169) only put the data into a buffer of the client and return. A background thread sends the buffer later (defaults of {{WriteOptions}} in 8.0.0: after 1000 ms or 1000 points) and reports a failed request as a {{WriteErrorEvent}}, which is logged with {{java.util.logging}} ("An error occurred during writing of data"). The producer does not listen to these events.

So, with default options:
# the exchange completes before InfluxDB has received anything;
# the {{catch (Exception ex) { exchange.setException(new CamelInfluxDb2Exception(ex)); }}} around each write (lines 103, 116, 129, 143, 157, 170) is never reached for a failed write;
# a write that InfluxDB answers with an error (a 4xx answer, for example 400 for a line it cannot parse, 401 / 403 for the token, 404 for an unknown bucket or organization, 413, 5xx after the retries of the client) or that cannot be sent at all (server down) is lost, and the route continues after the producer as if the data had been stored. There is no redelivery, no dead letter channel, no failed exchange; only a line in the log of the client.
# points that are still in the buffer when the JVM ends without {{InfluxDBClient.close()}} are lost as well.

{{from("kafka:measurements").to("influxdb2:client?org=o&bucket=b")}} therefore commits the offsets of records that were never written.

h3. Reproduction

New {{InfluxDb2ProducerWriteFailureTest}} (8 tests). It uses the real {{InfluxDBClient}} against a local HTTP server ({{com.sun.net.httpserver.HttpServer}}) that plays the organization, bucket and write resources of the InfluxDB 2.x API; route {{direct:write -> influxdb2:influxDbBean?org=myOrg&bucket=myBucket -> mock:after}}. With the producer of main all eight fail:
* {{pointRejectedByTheServerFailsTheExchange}} for 400, 401, 404 and 503: {{exception of an exchange whose write was answered with 400 ==> Unexpected null value, expected: <org.apache.camel.component.influxdb2.CamelInfluxDb2Exception> but was: <null>}};
* {{recordsRejectedByTheServerFailTheExchange}} (a {{Records}} body, answer 400): the same;
* {{writeToServerThatIsDownFailsTheExchange}}: the same;
* {{acceptedPointIsAtTheServerWhenTheExchangeCompletes}}: {{write requests at the server when the exchange has completed ==> expected: <1> but was: <0>}};
* {{acceptedRecordIsWrittenWithTheOrganizationBucketAndPrecisionOfTheEndpoint}} (a {{Record}} body; with the fix it also checks {{org=myOrg}}, {{bucket=myBucket}} and {{precision=ms}} in the query of the write request): {{write requests at the server when the exchange has completed ==> expected: <[temperature,location=north value=60.0]> but was: <[]>}}.

h3. Proposed fix

Use the synchronous write API, {{connection.getWriteApiBlocking()}}. {{WriteApiBlocking}} has the same six methods with the same parameters; they return after the answer of InfluxDB and throw {{InfluxException}} for an error answer or a request that could not be sent. The existing catch blocks then set it on the exchange as {{CamelInfluxDb2Exception}} with the {{InfluxException}} as cause.

Behaviour change, with a 4.24 upgrade-guide entry and a section in the component documentation:
* a failed write fails the exchange;
* each exchange is one write request (the client used to collect the points of up to one second into one request), and the client no longer retries a write in the background; a route with a high message rate can send a {{Points}}, {{Measurements}} or {{Records}} body to write many points with one request.

An option that keeps the batching write API (with a listener for the error events) could be added separately if somebody needs it; with it the outcome of a write cannot be the outcome of the exchange.

camel-influxdb2 tests with the fix: 21 run, 0 failures (13 existing, 8 new). No InfluxDB server was used.

How it was found: a sweep of the producers for failures that are not reported to the exchange, and a small TLA+ model of the write (exchange, buffer of the client, flush, answer of the server). For the code of main TLC reports that a successful exchange has nothing stored when it completes (2 states), and that a point of a successful exchange is dropped when the server rejects it (3 states) or when the process stops with a filled buffer; with the synchronous write the three properties hold. The model is only a sketch; the test above is the evidence.

Affected: main (a7169d7c6ab6, tested). camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x have the same lines (48, 102, 115, 128, 142, 156, 169): read, not run.

Duplicate check (2026-10-13): JIRA text "influxdb2 write error", "influxdb2 WriteApi": none; GitHub pull requests with "influxdb2" in the title: only 8824 (the component itself).

_Filed with Claude Code on behalf of allthingssecurity._




--
This message was sent by Atlassian Jira
(v8.20.10#820010)