[
https://issues.apache.org/jira/browse/CAMEL-25610?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25610:
--------------------------------
Fix Version/s: 4.24.0
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> {{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)
