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 b539152f3524 CAMEL-26231: camel-influxdb - fail the exchange on a
query error, do not change the Map body, and accept the operation in any case
and check it at startup (#29073)
b539152f3524 is described below
commit b539152f3524ef8d456835ed0752a3774c6a4a68
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 12:12:00 2026 +0100
CAMEL-26231: camel-influxdb - fail the exchange on a query error, do not
change the Map body, and accept the operation in any case and check it at
startup (#29073)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/components/influxdb.json | 4 +-
.../camel/catalog/docs/influxdb-component.adoc | 12 ++
.../apache/camel/component/influxdb/influxdb.json | 4 +-
.../src/main/docs/influxdb-component.adoc | 12 ++
.../camel/component/influxdb/InfluxDbEndpoint.java | 20 ++-
.../camel/component/influxdb/InfluxDbProducer.java | 31 ++++-
.../converters/CamelInfluxDbConverters.java | 12 +-
.../influxdb/InfluxDbProducerOperationTest.java | 102 +++++++++++++++
.../influxdb/InfluxDbProducerQueryErrorTest.java | 138 +++++++++++++++++++++
.../converters/CamelInfluxDbConverterTest.java | 27 ++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++
.../dsl/InfluxDbEndpointBuilderFactory.java | 3 +-
12 files changed, 359 insertions(+), 17 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb.json
index 33c950f81281..90c4aef1b0e9 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb.json
@@ -35,7 +35,7 @@
},
"bodyTypes": {
"producerIn": { "carriers": [ "pojo", "map" ], "byOption": [ { "option":
"batch", "value": "true", "carriers": [ "pojo" ], "javaType":
"org.influxdb.dto.BatchPoints" }, { "option": "operation", "value": "query",
"carriers": [ "any" ] }, { "option": "operation", "value": "ping", "carriers":
[ "any" ] } ], "description": "The org.influxdb.dto.Point to write, or a Map of
the field names and their values with the measurement name under the
camelInfluxDB.MeasurementName key. Text, bytes o [...]
- "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "query", "carriers": [ "pojo" ], "javaType":
"org.influxdb.dto.QueryResult" }, { "option": "operation", "value": "ping",
"carriers": [ "pojo" ], "javaType": "org.influxdb.dto.Pong" } ], "description":
"An insert leaves the body unchanged. A query gives the QueryResult of
InfluxDB, which carries an error the server reports (such as an unknown
database) instead of failing the exchange; ping gi [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "query", "carriers": [ "pojo" ], "javaType":
"org.influxdb.dto.QueryResult" }, { "option": "operation", "value": "ping",
"carriers": [ "pojo" ], "javaType": "org.influxdb.dto.Pong" } ], "description":
"An insert leaves the body unchanged. A query gives the QueryResult of
InfluxDB; an error the server reports in it (such as an unknown database) fails
the exchange instead. Ping gives the Pong [...]
},
"properties": {
"connectionBean": { "index": 0, "kind": "path", "displayName": "Connection
Bean", "group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Connection to the influx
database, of class InfluxDB.class" },
@@ -43,7 +43,7 @@
"batch": { "index": 2, "kind": "parameter", "displayName": "Batch",
"group": "producer", "label": "", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "description": "Define if this operation is a
batch operation or not" },
"checkDatabaseExistence": { "index": 3, "kind": "parameter",
"displayName": "Check Database Existence", "group": "producer", "label": "",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": false,
"description": "Define if we want to check the database existence while
starting the endpoint" },
"databaseName": { "index": 4, "kind": "parameter", "displayName":
"Database Name", "group": "producer", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "description": "The name of the database where the time
series will be stored" },
- "operation": { "index": 5, "kind": "parameter", "displayName":
"Operation", "group": "producer", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "insert", "description": "Define if
this operation is an insert or a query" },
+ "operation": { "index": 5, "kind": "parameter", "displayName":
"Operation", "group": "producer", "label": "", "required": false, "type":
"enum", "javaType": "java.lang.String", "enum": [ "insert", "query", "ping" ],
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"insert", "description": "Define if this operation is an insert, a query or a
ping. The value is not case-sensitive." },
"query": { "index": 6, "kind": "parameter", "displayName": "Query",
"group": "producer", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "description": "Define the query in case of operation query" },
"retentionPolicy": { "index": 7, "kind": "parameter", "displayName":
"Retention Policy", "group": "producer", "label": "", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "default", "description":
"The string that defines the retention policy to the data created by the
endpoint" },
"lazyStartProducer": { "index": 8, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produc [...]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb-component.adoc
index 7b170d9b0601..bd9e954fa582 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/influxdb-component.adoc
@@ -17,6 +17,7 @@ a time series database.
The native body type for this component is `Point` (the native influxdb class).
However, it can also accept `Map<String, Object>` as message body, and it will
get converted to `Point.class`, please note that the map must contain an
element with `camelInfluxDB.MeasurementName` as key.
+The other entries of the map are the fields of the point; the map itself is
not changed, so it can be immutable.
Additionally, you may register your own Converters to your data type to
`Point`, or use the
(un)marshalling tools provided by Camel.
@@ -51,6 +52,17 @@ include::partial$component-endpoint-options.adoc[]
include::partial$component-endpoint-headers.adoc[]
// component options: END
+== Operations
+
+The `operation` option selects what the producer does with each message:
`insert` (the default) writes the body,
+`query` runs the query of the `camelInfluxDB.query` header or of the `query`
option and gives its `QueryResult`, and
+`ping` gives the `Pong` of the server. The value is not case-sensitive, and an
endpoint with another operation fails
+to start.
+
+InfluxDB answers a query it cannot run, such as a query on a database that
does not exist, with the error in the
+result instead of an error status, so the client returns that result. An error
in the result of the query, or of one
+of its statements, fails the exchange with a `CamelInfluxDbException` whose
message holds the error of the server.
+
== Example
Below is an example route that stores a point into the db (taking the db name
from the URI)
diff --git
a/components/camel-influxdb/src/generated/resources/META-INF/org/apache/camel/component/influxdb/influxdb.json
b/components/camel-influxdb/src/generated/resources/META-INF/org/apache/camel/component/influxdb/influxdb.json
index 33c950f81281..90c4aef1b0e9 100644
---
a/components/camel-influxdb/src/generated/resources/META-INF/org/apache/camel/component/influxdb/influxdb.json
+++
b/components/camel-influxdb/src/generated/resources/META-INF/org/apache/camel/component/influxdb/influxdb.json
@@ -35,7 +35,7 @@
},
"bodyTypes": {
"producerIn": { "carriers": [ "pojo", "map" ], "byOption": [ { "option":
"batch", "value": "true", "carriers": [ "pojo" ], "javaType":
"org.influxdb.dto.BatchPoints" }, { "option": "operation", "value": "query",
"carriers": [ "any" ] }, { "option": "operation", "value": "ping", "carriers":
[ "any" ] } ], "description": "The org.influxdb.dto.Point to write, or a Map of
the field names and their values with the measurement name under the
camelInfluxDB.MeasurementName key. Text, bytes o [...]
- "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "query", "carriers": [ "pojo" ], "javaType":
"org.influxdb.dto.QueryResult" }, { "option": "operation", "value": "ping",
"carriers": [ "pojo" ], "javaType": "org.influxdb.dto.Pong" } ], "description":
"An insert leaves the body unchanged. A query gives the QueryResult of
InfluxDB, which carries an error the server reports (such as an unknown
database) instead of failing the exchange; ping gi [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "query", "carriers": [ "pojo" ], "javaType":
"org.influxdb.dto.QueryResult" }, { "option": "operation", "value": "ping",
"carriers": [ "pojo" ], "javaType": "org.influxdb.dto.Pong" } ], "description":
"An insert leaves the body unchanged. A query gives the QueryResult of
InfluxDB; an error the server reports in it (such as an unknown database) fails
the exchange instead. Ping gives the Pong [...]
},
"properties": {
"connectionBean": { "index": 0, "kind": "path", "displayName": "Connection
Bean", "group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "Connection to the influx
database, of class InfluxDB.class" },
@@ -43,7 +43,7 @@
"batch": { "index": 2, "kind": "parameter", "displayName": "Batch",
"group": "producer", "label": "", "required": false, "type": "boolean",
"javaType": "boolean", "deprecated": false, "autowired": false, "secret":
false, "defaultValue": false, "description": "Define if this operation is a
batch operation or not" },
"checkDatabaseExistence": { "index": 3, "kind": "parameter",
"displayName": "Check Database Existence", "group": "producer", "label": "",
"required": false, "type": "boolean", "javaType": "boolean", "deprecated":
false, "autowired": false, "secret": false, "defaultValue": false,
"description": "Define if we want to check the database existence while
starting the endpoint" },
"databaseName": { "index": 4, "kind": "parameter", "displayName":
"Database Name", "group": "producer", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "description": "The name of the database where the time
series will be stored" },
- "operation": { "index": 5, "kind": "parameter", "displayName":
"Operation", "group": "producer", "label": "", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": "insert", "description": "Define if
this operation is an insert or a query" },
+ "operation": { "index": 5, "kind": "parameter", "displayName":
"Operation", "group": "producer", "label": "", "required": false, "type":
"enum", "javaType": "java.lang.String", "enum": [ "insert", "query", "ping" ],
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"insert", "description": "Define if this operation is an insert, a query or a
ping. The value is not case-sensitive." },
"query": { "index": 6, "kind": "parameter", "displayName": "Query",
"group": "producer", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "description": "Define the query in case of operation query" },
"retentionPolicy": { "index": 7, "kind": "parameter", "displayName":
"Retention Policy", "group": "producer", "label": "", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "default", "description":
"The string that defines the retention policy to the data created by the
endpoint" },
"lazyStartProducer": { "index": 8, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produc [...]
diff --git a/components/camel-influxdb/src/main/docs/influxdb-component.adoc
b/components/camel-influxdb/src/main/docs/influxdb-component.adoc
index 7b170d9b0601..bd9e954fa582 100644
--- a/components/camel-influxdb/src/main/docs/influxdb-component.adoc
+++ b/components/camel-influxdb/src/main/docs/influxdb-component.adoc
@@ -17,6 +17,7 @@ a time series database.
The native body type for this component is `Point` (the native influxdb class).
However, it can also accept `Map<String, Object>` as message body, and it will
get converted to `Point.class`, please note that the map must contain an
element with `camelInfluxDB.MeasurementName` as key.
+The other entries of the map are the fields of the point; the map itself is
not changed, so it can be immutable.
Additionally, you may register your own Converters to your data type to
`Point`, or use the
(un)marshalling tools provided by Camel.
@@ -51,6 +52,17 @@ include::partial$component-endpoint-options.adoc[]
include::partial$component-endpoint-headers.adoc[]
// component options: END
+== Operations
+
+The `operation` option selects what the producer does with each message:
`insert` (the default) writes the body,
+`query` runs the query of the `camelInfluxDB.query` header or of the `query`
option and gives its `QueryResult`, and
+`ping` gives the `Pong` of the server. The value is not case-sensitive, and an
endpoint with another operation fails
+to start.
+
+InfluxDB answers a query it cannot run, such as a query on a database that
does not exist, with the error in the
+result instead of an error status, so the client returns that result. An error
in the result of the query, or of one
+of its statements, fails the exchange with a `CamelInfluxDbException` whose
message holds the error of the server.
+
== Example
Below is an example route that stores a point into the db (taking the db name
from the URI)
diff --git
a/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbEndpoint.java
b/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbEndpoint.java
index c1b34c4edb55..5d5a59dd30fb 100644
---
a/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbEndpoint.java
+++
b/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbEndpoint.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.influxdb;
+import java.util.Locale;
+import java.util.Set;
+
import org.apache.camel.Category;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
@@ -59,14 +62,16 @@ import org.slf4j.LoggerFactory;
@BodyType.When(option =
"operation", value = "ping", carriers = Carrier.POJO,
javaType =
"org.influxdb.dto.Pong") },
description = "An insert leaves the body
unchanged. A query gives the QueryResult of"
- + " InfluxDB, which carries
an error the server reports (such as an"
- + " unknown database) instead
of failing the exchange; ping gives the Pong"
- + " with the version of the
server"))
+ + " InfluxDB; an error the
server reports in it (such as an unknown"
+ + " database) fails the
exchange instead. Ping gives the Pong with the"
+ + " version of the server"))
public class InfluxDbEndpoint extends DefaultEndpoint {
private static final Logger LOG =
LoggerFactory.getLogger(InfluxDbEndpoint.class);
private static final String CREATE_DATABASE = "CREATE DATABASE ";
private static final String SHOW_DATABASES = "SHOW DATABASES";
+ private static final Set<String> OPERATIONS
+ = Set.of(InfluxDbOperations.INSERT, InfluxDbOperations.QUERY,
InfluxDbOperations.PING);
private InfluxDB influxDB;
@@ -79,7 +84,7 @@ public class InfluxDbEndpoint extends DefaultEndpoint {
private String retentionPolicy = "default";
@UriParam(defaultValue = "false")
private boolean batch;
- @UriParam(defaultValue = InfluxDbOperations.INSERT)
+ @UriParam(defaultValue = InfluxDbOperations.INSERT, enums =
"insert,query,ping")
private String operation = InfluxDbOperations.INSERT;
@UriParam
private String query;
@@ -106,6 +111,11 @@ public class InfluxDbEndpoint extends DefaultEndpoint {
protected void doInit() throws Exception {
super.doInit();
+ // an operation that is not supported fails the start of the endpoint,
instead of its first message
+ if (operation == null ||
!OPERATIONS.contains(operation.toLowerCase(Locale.ROOT))) {
+ throw new IllegalArgumentException(
+ "The operation " + operation + " is not supported: use one
of insert, query or ping");
+ }
if (checkDatabaseExistence) {
ensureDatabaseExists();
}
@@ -172,7 +182,7 @@ public class InfluxDbEndpoint extends DefaultEndpoint {
}
/**
- * Define if this operation is an insert or a query
+ * Define if this operation is an insert, a query or a ping. The value is
not case-sensitive.
*/
public void setOperation(String operation) {
this.operation = operation;
diff --git
a/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbProducer.java
b/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbProducer.java
index de98bff06eeb..ba11d7912178 100644
---
a/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbProducer.java
+++
b/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/InfluxDbProducer.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.influxdb;
+import java.util.Locale;
+
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.support.DefaultProducer;
@@ -56,7 +58,8 @@ public class InfluxDbProducer extends DefaultProducer {
public void process(Exchange exchange) throws Exception {
String dataBaseName = calculateDatabaseName(exchange);
String retentionPolicy = calculateRetentionPolicy(exchange);
- switch (endpoint.getOperation()) {
+ // the operation is not case-sensitive, and the endpoint fails to
start with one that is not supported
+ switch (endpoint.getOperation().toLowerCase(Locale.ROOT)) {
case InfluxDbOperations.INSERT:
doInsert(exchange, dataBaseName, retentionPolicy);
break;
@@ -99,10 +102,36 @@ public class InfluxDbProducer extends DefaultProducer {
String query = calculateQuery(exchange);
Query influxdbQuery = new Query(query, dataBaseName);
QueryResult resultSet = connection.query(influxdbQuery);
+ String error = errorOf(resultSet);
+ if (error != null) {
+ throw new CamelInfluxDbException("The query failed in InfluxDB: "
+ error);
+ }
MessageHelper.copyHeaders(exchange.getIn(), exchange.getOut(), true);
exchange.getMessage().setBody(resultSet);
}
+ /**
+ * The error that InfluxDB reports in the result of a query, or null.
InfluxDB answers a query that it cannot run,
+ * for example on a database that does not exist, with the error in the
result of the query or of one of its
+ * statements instead of an error status, so the client returns that
result instead of throwing.
+ */
+ private static String errorOf(QueryResult resultSet) {
+ if (resultSet == null) {
+ return null;
+ }
+ if (resultSet.hasError()) {
+ return resultSet.getError();
+ }
+ if (resultSet.getResults() != null) {
+ for (QueryResult.Result result : resultSet.getResults()) {
+ if (result != null && result.hasError()) {
+ return result.getError();
+ }
+ }
+ }
+ return null;
+ }
+
private void doPing(Exchange exchange) {
Pong result = connection.ping();
MessageHelper.copyHeaders(exchange.getIn(), exchange.getOut(), true);
diff --git
a/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverters.java
b/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverters.java
index e3d2aafaffb3..673e1af15820 100644
---
a/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverters.java
+++
b/components/camel-influxdb/src/main/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverters.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.influxdb.converters;
+import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.camel.Converter;
@@ -37,12 +38,11 @@ public final class CamelInfluxDbConverters {
throw new CamelInfluxDbException(format);
}
- String measurenmentNameString = measurementName.toString();
- Point.Builder pointBuilder = Point.measurement(measurenmentNameString);
- map.remove(InfluxDbConstants.MEASUREMENT_NAME);
- pointBuilder.fields(map);
- map.put(InfluxDbConstants.MEASUREMENT_NAME, measurementName);
- return pointBuilder.build();
+ // the fields are a copy of the map without the measurement name, so
the map of the caller is not changed (it can
+ // be immutable or shared with other code)
+ Map<String, Object> fields = new LinkedHashMap<>(map);
+ fields.remove(InfluxDbConstants.MEASUREMENT_NAME);
+ return
Point.measurement(measurementName.toString()).fields(fields).build();
}
}
diff --git
a/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/InfluxDbProducerOperationTest.java
b/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/InfluxDbProducerOperationTest.java
new file mode 100644
index 000000000000..c14ed68d9db9
--- /dev/null
+++
b/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/InfluxDbProducerOperationTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.influxdb;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.util.ObjectHelper;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Pong;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The operation is one of insert, query or ping, in any case, and an endpoint
with another operation fails to start
+ * instead of failing its first message.
+ */
+class InfluxDbProducerOperationTest extends AbstractInfluxDbTest {
+
+ @Test
+ void queryInUpperCaseRunsTheQuery() throws Exception {
+ QueryResult queryResult = new QueryResult();
+ queryResult.setResults(List.of(new QueryResult.Result()));
+
when(mockedDbConnection.query(any(Query.class))).thenReturn(queryResult);
+
addRouteTo("influxdb:influxDbBean?databaseName={{influxdb.testDb}}&operation=QUERY");
+
+ Exchange exchange = template.send("direct:start",
+ e -> e.getIn().setHeader(InfluxDbConstants.INFLUXDB_QUERY,
"select * from cpu"));
+
+ assertNull(exchange.getException());
+ assertSame(queryResult, exchange.getMessage().getBody());
+ }
+
+ @Test
+ void pingInMixedCasePings() throws Exception {
+ Pong pong = new Pong();
+ pong.setVersion("1.8.10");
+ when(mockedDbConnection.ping()).thenReturn(pong);
+
addRouteTo("influxdb:influxDbBean?databaseName={{influxdb.testDb}}&operation=Ping");
+
+ Exchange exchange = template.send("direct:start", e ->
e.getIn().setBody("test"));
+
+ assertNull(exchange.getException());
+ assertSame(pong, exchange.getMessage().getBody());
+ }
+
+ @Test
+ void insertInUpperCaseWritesThePoint() throws Exception {
+
addRouteTo("influxdb:influxDbBean?databaseName={{influxdb.testDb}}&operation=INSERT");
+
+ Point point = Point.measurement("cpu").addField("idle", 90L).build();
+
+ Exchange exchange = template.send("direct:start", e ->
e.getIn().setBody(point));
+
+ assertNull(exchange.getException());
+ verify(mockedDbConnection).write("myTestTimeSeries", "default", point);
+ }
+
+ @Test
+ void unsupportedOperationFailsTheStartOfTheRoute() {
+ Exception failure = assertThrows(Exception.class,
+ () ->
addRouteTo("influxdb:influxDbBean?databaseName={{influxdb.testDb}}&operation=delete"));
+
+ IllegalArgumentException cause =
ObjectHelper.getException(IllegalArgumentException.class, failure);
+ assertNotNull(cause, "IllegalArgumentException in the causes of " +
failure);
+ assertEquals("The operation delete is not supported: use one of
insert, query or ping", cause.getMessage());
+ }
+
+ private void addRouteTo(String uri) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to(uri);
+ }
+ });
+ }
+}
diff --git
a/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/InfluxDbProducerQueryErrorTest.java
b/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/InfluxDbProducerQueryErrorTest.java
new file mode 100644
index 000000000000..c54f64ef157e
--- /dev/null
+++
b/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/InfluxDbProducerQueryErrorTest.java
@@ -0,0 +1,138 @@
+/*
+ * 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.influxdb;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+import org.junit.jupiter.api.Test;
+
+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.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+/**
+ * InfluxDB answers a query it cannot run, such as one on a database that does
not exist, with the error in the result
+ * instead of an error status, and the client returns that result: such an
error must fail the exchange.
+ */
+class InfluxDbProducerQueryErrorTest extends AbstractInfluxDbTest {
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:query")
+
.to("influxdb:influxDbBean?databaseName={{influxdb.testDb}}&operation=query")
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void errorOfTheStatementFailsTheExchange() throws Exception {
+ // the answer of InfluxDB to a query on a database that does not exist
+ answer(queryResult(null, result("database not found:
myTestTimeSeries")));
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(0);
+
+ Exchange exchange = query();
+
+ CamelInfluxDbException failure =
assertInstanceOf(CamelInfluxDbException.class, exchange.getException(),
+ "exception of an exchange whose query failed");
+ assertEquals("The query failed in InfluxDB: database not found:
myTestTimeSeries", failure.getMessage());
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void errorOfTheQueryFailsTheExchange() throws Exception {
+ answer(queryResult("query interrupted"));
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(0);
+
+ Exchange exchange = query();
+
+ CamelInfluxDbException failure =
assertInstanceOf(CamelInfluxDbException.class, exchange.getException(),
+ "exception of an exchange whose query failed");
+ assertEquals("The query failed in InfluxDB: query interrupted",
failure.getMessage());
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void errorOfTheSecondStatementFailsTheExchange() throws Exception {
+ answer(queryResult(null, result(null), result("measurement not
found")));
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(0);
+
+ Exchange exchange = query();
+
+ CamelInfluxDbException failure =
assertInstanceOf(CamelInfluxDbException.class, exchange.getException(),
+ "exception of an exchange whose second statement failed");
+ assertEquals("The query failed in InfluxDB: measurement not found",
failure.getMessage());
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void resultWithoutErrorIsTheBody() throws Exception {
+ QueryResult answer = queryResult(null, result(null));
+ answer(answer);
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(1);
+
+ Exchange exchange = query();
+
+ assertNull(exchange.getException());
+ assertSame(answer, exchange.getMessage().getBody());
+ result.assertIsSatisfied();
+ }
+
+ private Exchange query() {
+ return template.send("direct:query",
+ e -> e.getIn().setHeader(InfluxDbConstants.INFLUXDB_QUERY,
"select * from cpu"));
+ }
+
+ private void answer(QueryResult queryResult) {
+
when(mockedDbConnection.query(any(Query.class))).thenReturn(queryResult);
+ }
+
+ private static QueryResult queryResult(String error, QueryResult.Result...
results) {
+ QueryResult queryResult = new QueryResult();
+ queryResult.setError(error);
+ queryResult.setResults(List.of(results));
+ return queryResult;
+ }
+
+ private static QueryResult.Result result(String error) {
+ QueryResult.Result result = new QueryResult.Result();
+ result.setError(error);
+ if (error == null) {
+ QueryResult.Series series = new QueryResult.Series();
+ series.setName("cpu");
+ series.setColumns(List.of("time", "idle"));
+ series.setValues(List.of(List.of("2026-11-03T00:00:00Z", 90)));
+ result.setSeries(List.of(series));
+ }
+ return result;
+ }
+}
diff --git
a/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverterTest.java
b/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverterTest.java
index ac405932bff7..4b0c8890e7fa 100644
---
a/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverterTest.java
+++
b/components/camel-influxdb/src/test/java/org/apache/camel/component/influxdb/converters/CamelInfluxDbConverterTest.java
@@ -17,6 +17,8 @@
package org.apache.camel.component.influxdb.converters;
import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import org.apache.camel.component.influxdb.InfluxDbConstants;
@@ -25,6 +27,7 @@ import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -87,6 +90,30 @@ public class CamelInfluxDbConverterTest {
}
+ @Test
+ void convertsAnImmutableMap() {
+ Map<String, Object> pointInMapFormat =
Map.of(InfluxDbConstants.MEASUREMENT_NAME, "testCPU", "busy", 99);
+
+ Point p = CamelInfluxDbConverters.fromMapToPoint(pointInMapFormat);
+
+ assertEquals("testCPU busy=99i", p.lineProtocol());
+ }
+
+ @Test
+ void doesNotChangeTheMapOfTheCaller() {
+ Map<String, Object> pointInMapFormat = new LinkedHashMap<>();
+ pointInMapFormat.put(InfluxDbConstants.MEASUREMENT_NAME, "testCPU");
+ pointInMapFormat.put("busy", 99);
+ Map<String, Object> before = new LinkedHashMap<>(pointInMapFormat);
+
+ Point p = CamelInfluxDbConverters.fromMapToPoint(pointInMapFormat);
+
+ assertEquals("testCPU busy=99i", p.lineProtocol());
+ assertEquals(before, pointInMapFormat);
+ // the entries are in the same order: the measurement name was not
removed and put back at the end
+ assertEquals(List.copyOf(before.keySet()),
List.copyOf(pointInMapFormat.keySet()));
+ }
+
@Test
public void canAddByte() {
Map<String, Object> pointInMapFormat = new HashMap<>();
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 488977773fba..d13479088c32 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
@@ -4758,6 +4758,17 @@ message whose data coding is 8-bit is now split by its
length in bytes.
`SmppSmCommand.getShortMessage(Message)` now declares `throws SmppException`.
+=== camel-influxdb - a query error fails the exchange, and the operation is
checked at startup
+
+InfluxDB answers a query it cannot run, such as a query on a database that
does not exist, with the error inside the
+result. The producer used to set that `QueryResult` as the body, and the route
continued as if the query had
+succeeded. An error in the result of the query, or of one of its statements,
now fails the exchange with a
+`CamelInfluxDbException` whose message holds the error of the server.
+
+The `operation` option now lists its values (`insert`, `query` and `ping`) and
is not case-sensitive, so
+`operation=QUERY` runs a query. An endpoint with another operation now fails
to start, where it used to start and fail
+every message with an `IllegalArgumentException`.
+
=== camel-paho, camel-paho-mqtt5 - a body that does not convert to bytes
The `paho` and `paho-mqtt5` producers now fail with an
`InvalidPayloadException` that names the type of the body for a
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/InfluxDbEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/InfluxDbEndpointBuilderFactory.java
index 8e4f55f4f802..cdf8420e6a42 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/InfluxDbEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/InfluxDbEndpointBuilderFactory.java
@@ -151,7 +151,8 @@ public interface InfluxDbEndpointBuilderFactory {
return this;
}
/**
- * Define if this operation is an insert or a query.
+ * Define if this operation is an insert, a query or a ping. The value
+ * is not case-sensitive.
*
* The option is a: <code>java.lang.String</code> type.
*
