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)
