This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 7663963b0e8a CAMEL-25610: camel-influxdb2 - Fail the exchange when
InfluxDB rejects the write (#27797)
7663963b0e8a is described below
commit 7663963b0e8a2c560a8c504d778a67d6295360ca
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 14 17:16:32 2026 +0530
CAMEL-25610: camel-influxdb2 - Fail the exchange when InfluxDB rejects the
write (#27797)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/influxdb2-component.adoc | 11 ++
.../src/main/docs/influxdb2-component.adoc | 11 ++
.../component/influxdb2/InfluxDb2Producer.java | 10 +-
.../InfluxDb2ProducerWriteFailureTest.java | 213 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++
5 files changed, 255 insertions(+), 3 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb2-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb2-component.adoc
index 9fb2383ee7d9..9fd774433882 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb2-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb2-component.adoc
@@ -85,3 +85,14 @@ YAML::
bucket: myBucket
----
====
+
+== Writes are synchronous
+
+The producer writes with the synchronous write API of the InfluxDB client
(`WriteApiBlocking`): the exchange
+continues when InfluxDB has answered the write request.
+A write that InfluxDB rejects (for example, a line it cannot parse, an unknown
bucket or a token that is not
+authorized), or a request that cannot be sent, fails the exchange with a
`CamelInfluxDb2Exception` whose cause is the
+`InfluxException` of the client, so the error handler of the route can
redeliver it or move it to a dead letter channel.
+
+Each exchange is one write request. To write many points with one request,
send a `Points`, `Measurements` or
+`Records` body, for example, from an xref:eips:aggregate-eip.adoc[Aggregate]
EIP.
diff --git a/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
b/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
index 9fb2383ee7d9..9fd774433882 100644
--- a/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
+++ b/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
@@ -85,3 +85,14 @@ YAML::
bucket: myBucket
----
====
+
+== Writes are synchronous
+
+The producer writes with the synchronous write API of the InfluxDB client
(`WriteApiBlocking`): the exchange
+continues when InfluxDB has answered the write request.
+A write that InfluxDB rejects (for example, a line it cannot parse, an unknown
bucket or a token that is not
+authorized), or a request that cannot be sent, fails the exchange with a
`CamelInfluxDb2Exception` whose cause is the
+`InfluxException` of the client, so the error handler of the route can
redeliver it or move it to a dead letter channel.
+
+Each exchange is one write request. To write many points with one request,
send a `Points`, `Measurements` or
+`Records` body, for example, from an xref:eips:aggregate-eip.adoc[Aggregate]
EIP.
diff --git
a/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Producer.java
b/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Producer.java
index 417414a552d0..ff525fef6a7d 100644
---
a/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Producer.java
+++
b/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Producer.java
@@ -17,7 +17,7 @@
package org.apache.camel.component.influxdb2;
import com.influxdb.client.InfluxDBClient;
-import com.influxdb.client.WriteApi;
+import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import org.apache.camel.Exchange;
@@ -39,13 +39,17 @@ public class InfluxDb2Producer extends DefaultProducer {
InfluxDBClient connection;
- WriteApi writeApi;
+ /**
+ * The synchronous write API of the client: a write returns when InfluxDB
has answered and throws when the answer is
+ * an error or the request could not be sent, so the outcome of the write
is the outcome of the exchange.
+ */
+ WriteApiBlocking writeApi;
public InfluxDb2Producer(InfluxDb2Endpoint endpoint) {
super(endpoint);
this.connection = endpoint.getInfluxDBClient();
this.endpoint = endpoint;
- this.writeApi = connection.makeWriteApi();
+ this.writeApi = connection.getWriteApiBlocking();
}
/**
diff --git
a/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2ProducerWriteFailureTest.java
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2ProducerWriteFailureTest.java
new file mode 100644
index 000000000000..4242cc7e889a
--- /dev/null
+++
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2ProducerWriteFailureTest.java
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.influxdb2;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.influxdb.client.InfluxDBClient;
+import com.influxdb.client.InfluxDBClientFactory;
+import com.influxdb.client.write.Point;
+import com.influxdb.exceptions.InfluxException;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.influxdb2.data.Record;
+import org.apache.camel.component.influxdb2.data.Records;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The producer writes with the real InfluxDB client against a local HTTP
server that plays the write API of InfluxDB
+ * 2.x: a write that the server rejects, or that never reaches it, must fail
the exchange, and an accepted write must be
+ * at the server when the exchange completes.
+ */
+class InfluxDb2ProducerWriteFailureTest extends CamelTestSupport {
+
+ private static final String ORG_ID = "00000000000000a1";
+
+ private final AtomicInteger writeStatus = new AtomicInteger(204);
+ private final List<String> writes = new CopyOnWriteArrayList<>();
+ private final List<String> writeQueries = new CopyOnWriteArrayList<>();
+
+ private HttpServer server;
+ private InfluxDBClient client;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ server = HttpServer.create(new
InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+ server.createContext("/api/v2/orgs",
+ exchange -> respond(exchange, 200, "{\"orgs\":[{\"id\":\"" +
ORG_ID + "\",\"name\":\"myOrg\"}]}"));
+ server.createContext("/api/v2/buckets",
+ exchange -> respond(exchange, 200,
+
"{\"buckets\":[{\"id\":\"00000000000000b1\",\"orgID\":\"" + ORG_ID
+ +
"\",\"name\":\"myBucket\",\"retentionRules\":[]}]}"));
+ server.createContext("/api/v2/write", exchange -> {
+ int status = writeStatus.get();
+ String lines = new
String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
+ if (status == 204) {
+
writeQueries.add(String.valueOf(exchange.getRequestURI().getQuery()));
+ writes.add(lines);
+ respond(exchange, 204, null);
+ } else {
+ respond(exchange, status,
"{\"code\":\"rejected\",\"message\":\"write rejected with " + status + "\"}");
+ }
+ });
+ server.start();
+
+ client = InfluxDBClientFactory.create("http://127.0.0.1:"; +
server.getAddress().getPort(), "token".toCharArray());
+
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("influxDbBean", client);
+ return context;
+ }
+
+ @Override
+ protected void cleanupResources() {
+ if (server != null) {
+ server.stop(0);
+ }
+ if (client != null) {
+ client.close();
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:write")
+ .to("influxdb2:influxDbBean?org=myOrg&bucket=myBucket")
+ .to("mock:after");
+ }
+ };
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { 400, 401, 404, 503 })
+ void pointRejectedByTheServerFailsTheExchange(int status) throws Exception
{
+ writeStatus.set(status);
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(0);
+
+ Exchange exchange = template.send("direct:write", e ->
e.getIn().setBody(point()));
+
+ CamelInfluxDb2Exception failure
+ = assertInstanceOf(CamelInfluxDb2Exception.class,
exchange.getException(),
+ "exception of an exchange whose write was answered
with " + status);
+ InfluxException cause = assertInstanceOf(InfluxException.class,
failure.getCause());
+ assertEquals(status, cause.status());
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void recordsRejectedByTheServerFailTheExchange() throws Exception {
+ writeStatus.set(400);
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(0);
+
+ Exchange exchange = template.send("direct:write",
+ e ->
e.getIn().setBody(Records.create("temperature,location=north value=60.0")
+ .addRecord("temperature,location=south value=61.0")));
+
+ assertInstanceOf(CamelInfluxDb2Exception.class,
exchange.getException(),
+ "exception of an exchange whose records were answered with
400");
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void writeToServerThatIsDownFailsTheExchange() throws Exception {
+ server.stop(0);
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(0);
+
+ Exchange exchange = template.send("direct:write", e ->
e.getIn().setBody(point()));
+
+ assertInstanceOf(CamelInfluxDb2Exception.class,
exchange.getException(),
+ "exception of an exchange whose write could not be sent");
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void acceptedPointIsAtTheServerWhenTheExchangeCompletes() throws Exception
{
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(1);
+
+ Exchange exchange = template.send("direct:write", e ->
e.getIn().setBody(point()));
+
+ assertNull(exchange.getException());
+ assertEquals(1, writes.size(), "write requests at the server when the
exchange has completed");
+ assertTrue(writes.get(0).startsWith("temperature,location=north
value=60.0"), writes.get(0));
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void
acceptedRecordIsWrittenWithTheOrganizationBucketAndPrecisionOfTheEndpoint()
throws Exception {
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(1);
+
+ Exchange exchange = template.send("direct:write",
+ e ->
e.getIn().setBody(Record.fromString("temperature,location=north value=60.0")));
+
+ assertNull(exchange.getException());
+ assertEquals(List.of("temperature,location=north value=60.0"), writes,
+ "write requests at the server when the exchange has
completed");
+ // the parameters of the write request, in any order; ms is the
default writePrecision of the endpoint
+ List<String> parameters = List.of(writeQueries.get(0).split("&"));
+ assertTrue(parameters.contains("org=myOrg"), writeQueries.get(0));
+ assertTrue(parameters.contains("bucket=myBucket"),
writeQueries.get(0));
+ assertTrue(parameters.contains("precision=ms"), writeQueries.get(0));
+ after.assertIsSatisfied();
+ }
+
+ private static Point point() {
+ return Point.measurement("temperature").addTag("location",
"north").addField("value", 60.0);
+ }
+
+ private static void respond(HttpExchange exchange, int status, String
json) throws IOException {
+ // read the request completely before answering
+ exchange.getRequestBody().readAllBytes();
+ if (json == null) {
+ exchange.sendResponseHeaders(status, -1);
+ } else {
+ byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(status, bytes.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(bytes);
+ }
+ }
+ exchange.close();
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 81ba4af58a03..ff5caf83b25d 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -245,6 +245,19 @@ nothing was stored and the exchange completed without an
error. The requests are
the answer like the other state operations: the states are stored, and a
request that Dapr rejects (or a sidecar that
cannot be reached) fails the exchange.
+=== camel-influxdb2 - a write is synchronous and a rejected write fails the
exchange
+
+The producer wrote with the batching write API of the InfluxDB client: the
point was put into a buffer of the client
+and sent later by a background thread, so the exchange completed before
anything was written, and a write that
+InfluxDB rejected (a line it cannot parse, an unknown bucket, a token that is
not authorized) or could not receive was
+only logged by the client while the route continued as if the data had been
stored. The producer now writes with the
+synchronous write API: the exchange continues when InfluxDB has answered, and
a failed write fails the exchange with a
+`CamelInfluxDb2Exception` (cause: the `InfluxException` of the client), which
the error handler can redeliver.
+
+Each exchange is now one write request, where 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. Routes
with a high message rate can send a
+`Points`, `Measurements` or `Records` body (for example from an aggregator) to
write many points with one request.
+
=== camel-jira - the producers wait for the answer of Jira
The `addComment`, `addIssueLink`, `addWorkLog`, `attach`, `deleteIssue`,
`transitionIssue` and `watchers` producers
