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 53219dce7deb CAMEL-26230: camel-influxdb2 - fail a Point without
fields, call InfluxDB at startup only to create the organization or bucket, and
do not change the Map body (#29072)
53219dce7deb is described below
commit 53219dce7deb1a06cc84386154578ce7ba5a07f1
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 12:12:11 2026 +0100
CAMEL-26230: camel-influxdb2 - fail a Point without fields, call InfluxDB
at startup only to create the organization or bucket, and do not change the Map
body (#29072)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/components/influxdb2.json | 2 +-
.../camel/catalog/docs/influxdb2-component.adoc | 17 ++
.../camel/component/influxdb2/influxdb2.json | 2 +-
.../src/main/docs/influxdb2-component.adoc | 17 ++
.../component/influxdb2/InfluxDb2Endpoint.java | 59 +++--
.../component/influxdb2/InfluxDb2Producer.java | 35 ++-
.../converters/CamelInfluxDb2Converters.java | 13 +-
.../influxdb2/InfluxDb2EndpointStartupTest.java | 247 +++++++++++++++++++++
.../component/influxdb2/InfluxDb2ProducerTest.java | 4 +-
.../InfluxDb2ProducerWriteFailureTest.java | 94 ++++++++
.../converters/CamelInfluxDbConverterTest.java | 27 +++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 ++
12 files changed, 496 insertions(+), 35 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb2.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb2.json
index b0b337fad97c..dcce3b9ef9df 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb2.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/influxdb2.json
@@ -34,7 +34,7 @@
"CamelInfluxDB2WritePrecision": { "index": 2, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "com.influxdb.client.domain.WritePrecision", "enum": [ "ms", "s",
"us", "ns" ], "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "InfluxDb Write precision.", "constantName":
"org.apache.camel.component.influxdb2.InfluxDb2Constants#WRITE_PRECISION" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo", "map" ], "byOption": [ { "option":
"operation", "value": "PING", "carriers": [ "any" ] } ], "description": "A
com.influxdb.client.write.Point, or the Measurement, Record, Points,
Measurements or Records wrappers of the component (package
org.apache.camel.component.influxdb2.data), or a Map of the field names and
their values with the measurement name under the CamelInfluxDB2MeasurementName
key. Text, a List or an annotated class that is not wrapp [...]
+ "producerIn": { "carriers": [ "pojo", "map" ], "byOption": [ { "option":
"operation", "value": "PING", "carriers": [ "any" ] } ], "description": "A
com.influxdb.client.write.Point, or the Measurement, Record, Points,
Measurements or Records wrappers of the component (package
org.apache.camel.component.influxdb2.data), or a Map of the field names and
their values with the measurement name under the CamelInfluxDB2MeasurementName
key. Text, a List or an annotated class that is not wrapp [...]
"producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "PING", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" } ], "description": "An insert leaves the body unchanged;
ping gives a Boolean, true when the server answered" }
},
"properties": {
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 9991a0c1b2f1..eb9be9f59978 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
@@ -16,6 +16,7 @@ This component allows you to interact with
https://influxdata.com/time-series-pl
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 `CamelInfluxDB2MeasurementName` 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.
@@ -96,3 +97,19 @@ authorized), or a request that cannot be sent, fails the
exchange with a `CamelI
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.
+
+InfluxDB needs at least one field in a point, and the client skips a point
without a field that has a value (no field
+at all, or only fields whose value is `null`, NaN or infinite) instead of
writing it. Such a `Point`, also one converted
+from a `Map` or one of the points of a `Points` body, fails the exchange with
a `CamelInfluxDb2Exception`, and nothing
+is written.
+
+== Organization and bucket at startup
+
+When `autoCreateOrg` is `true` (the default), the endpoint looks the
organization up when it starts and creates it when
+it does not exist. When `autoCreateBucket` is `true` (the default), it does
the same for the bucket, in the organization
+of the endpoint. With `autoCreateOrg=false`, a missing bucket is created only
when that organization exists, and
+otherwise the endpoint fails to start.
+
+A write names the organization and the bucket, so the endpoint does not need
them to be looked up: with
+`autoCreateOrg=false` and `autoCreateBucket=false` it does not call InfluxDB
when it starts, and the route starts while
+InfluxDB is down. An organization or a bucket that does not exist then fails
the write, and so the exchange.
diff --git
a/components/camel-influxdb2/src/generated/resources/META-INF/org/apache/camel/component/influxdb2/influxdb2.json
b/components/camel-influxdb2/src/generated/resources/META-INF/org/apache/camel/component/influxdb2/influxdb2.json
index b0b337fad97c..dcce3b9ef9df 100644
---
a/components/camel-influxdb2/src/generated/resources/META-INF/org/apache/camel/component/influxdb2/influxdb2.json
+++
b/components/camel-influxdb2/src/generated/resources/META-INF/org/apache/camel/component/influxdb2/influxdb2.json
@@ -34,7 +34,7 @@
"CamelInfluxDB2WritePrecision": { "index": 2, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "com.influxdb.client.domain.WritePrecision", "enum": [ "ms", "s",
"us", "ns" ], "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "InfluxDb Write precision.", "constantName":
"org.apache.camel.component.influxdb2.InfluxDb2Constants#WRITE_PRECISION" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo", "map" ], "byOption": [ { "option":
"operation", "value": "PING", "carriers": [ "any" ] } ], "description": "A
com.influxdb.client.write.Point, or the Measurement, Record, Points,
Measurements or Records wrappers of the component (package
org.apache.camel.component.influxdb2.data), or a Map of the field names and
their values with the measurement name under the CamelInfluxDB2MeasurementName
key. Text, a List or an annotated class that is not wrapp [...]
+ "producerIn": { "carriers": [ "pojo", "map" ], "byOption": [ { "option":
"operation", "value": "PING", "carriers": [ "any" ] } ], "description": "A
com.influxdb.client.write.Point, or the Measurement, Record, Points,
Measurements or Records wrappers of the component (package
org.apache.camel.component.influxdb2.data), or a Map of the field names and
their values with the measurement name under the CamelInfluxDB2MeasurementName
key. Text, a List or an annotated class that is not wrapp [...]
"producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "PING", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" } ], "description": "An insert leaves the body unchanged;
ping gives a Boolean, true when the server answered" }
},
"properties": {
diff --git a/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
b/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
index 9991a0c1b2f1..eb9be9f59978 100644
--- a/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
+++ b/components/camel-influxdb2/src/main/docs/influxdb2-component.adoc
@@ -16,6 +16,7 @@ This component allows you to interact with
https://influxdata.com/time-series-pl
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 `CamelInfluxDB2MeasurementName` 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.
@@ -96,3 +97,19 @@ authorized), or a request that cannot be sent, fails the
exchange with a `CamelI
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.
+
+InfluxDB needs at least one field in a point, and the client skips a point
without a field that has a value (no field
+at all, or only fields whose value is `null`, NaN or infinite) instead of
writing it. Such a `Point`, also one converted
+from a `Map` or one of the points of a `Points` body, fails the exchange with
a `CamelInfluxDb2Exception`, and nothing
+is written.
+
+== Organization and bucket at startup
+
+When `autoCreateOrg` is `true` (the default), the endpoint looks the
organization up when it starts and creates it when
+it does not exist. When `autoCreateBucket` is `true` (the default), it does
the same for the bucket, in the organization
+of the endpoint. With `autoCreateOrg=false`, a missing bucket is created only
when that organization exists, and
+otherwise the endpoint fails to start.
+
+A write names the organization and the bucket, so the endpoint does not need
them to be looked up: with
+`autoCreateOrg=false` and `autoCreateBucket=false` it does not call InfluxDB
when it starts, and the route starts while
+InfluxDB is down. An organization or a bucket that does not exist then fails
the write, and so the exchange.
diff --git
a/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Endpoint.java
b/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Endpoint.java
index 24b205798d05..2763de435a2f 100644
---
a/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Endpoint.java
+++
b/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/InfluxDb2Endpoint.java
@@ -56,7 +56,9 @@ import org.slf4j.LoggerFactory;
+ " and their values with the
measurement name under the"
+ "
CamelInfluxDB2MeasurementName key. Text, a List or an annotated class"
+ " that is not wrapped fail:
wrap a line protocol String in a Record and"
- + " an annotated object in a
Measurement. Ping does not read the body"),
+ + " an annotated object in a
Measurement. A Point without a field that has"
+ + " a value (also from a Map
or in Points) fails the exchange, as the client"
+ + " would skip it. Ping does
not read the body"),
producerOut = @BodyType(value = Carrier.UNCHANGED,
byOption = @BodyType.When(option =
"operation", value = "PING", carriers = Carrier.VALUE,
javaType =
"java.lang.Boolean"),
@@ -191,37 +193,44 @@ public class InfluxDb2Endpoint extends DefaultEndpoint {
@Override
protected void doInit() throws Exception {
super.doInit();
- ensureOrgExists();
- ensureBucketExists();
+ // InfluxDB is called at startup only to create the organization or
the bucket when it is missing. A write names
+ // the organization and the bucket, so an endpoint that creates
neither does not need the server to start.
+ if (autoCreateOrg) {
+ ensureOrgExists();
+ }
+ if (autoCreateBucket) {
+ ensureBucketExists();
+ }
}
private void ensureOrgExists() {
- Organization organization = null;
+ Organization organization = findOrganization();
+ if (organization == null) {
+ LOG.debug("Organization {} doesn't exist. Creating it...", org);
+ organization =
getInfluxDBClient().getOrganizationsApi().createOrganization(org);
+ }
+ if (organization != null) {
+ setOrgID(organization.getId());
+ }
+ }
+
+ private Organization findOrganization() {
OrganizationsQuery organizationsQuery = new OrganizationsQuery();
organizationsQuery.setOrg(org);
- boolean exists = false;
try {
List<Organization> organizations =
getInfluxDBClient().getOrganizationsApi().findOrganizations(organizationsQuery);
- if (organizations.stream().anyMatch(o -> o.getName().equals(org)))
{
- exists = true;
- organization = organizations.stream().filter(o ->
o.getName().equals(org)).findFirst().get();
+ for (Organization organization : organizations) {
+ if (organization.getName().equals(org)) {
+ return organization;
+ }
}
} catch (NotFoundException ex) {
- exists = false;
- }
- if (!exists && autoCreateOrg) {
- LOG.debug("Organization {} doesn't exist. Creating it...", org);
- organization =
getInfluxDBClient().getOrganizationsApi().createOrganization(org);
- }
- if (organization != null) {
- setOrgID(organization.getId());
+ // the organization does not exist
}
+ return null;
}
private void ensureBucketExists() {
-
- ensureOrgExists();
-
boolean exists = false;
BucketsQuery bucketsQuery = new BucketsQuery();
bucketsQuery.setOrg(org);
@@ -234,7 +243,17 @@ public class InfluxDb2Endpoint extends DefaultEndpoint {
} catch (NotFoundException ex) {
exists = false;
}
- if (!exists && autoCreateBucket) {
+ if (!exists) {
+ if (!autoCreateOrg) {
+ // the organization was not looked up at startup: the bucket
is created in it only when it exists
+ Organization organization = findOrganization();
+ if (organization == null) {
+ throw new CamelInfluxDb2Exception(
+ "Cannot create the bucket " + bucket + " because
the organization " + org
+ + " does not exist and
autoCreateOrg is false");
+ }
+ setOrgID(organization.getId());
+ }
LOG.debug("Bucket {} doesn't exist. Creating it...", bucket);
getInfluxDBClient().getBucketsApi().createBucket(bucket,
getOrgID());
}
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 ff525fef6a7d..146856209ff7 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
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.influxdb2;
+import java.util.List;
+
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.WritePrecision;
@@ -100,10 +102,11 @@ public class InfluxDb2Producer extends DefaultProducer {
throws InvalidPayloadException {
Point point = exchange.getIn().getMandatoryBody(Point.class);
try {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Writing point {}", point.toLineProtocol());
- }
+ String line = lineProtocolOf(point, "The point");
+ LOG.debug("Writing point {}", line);
writeApi.writePoint(bucketName, orgName, point);
+ } catch (CamelInfluxDb2Exception ex) {
+ exchange.setException(ex);
} catch (Exception ex) {
exchange.setException(new CamelInfluxDb2Exception(ex));
}
@@ -143,7 +146,16 @@ public class InfluxDb2Producer extends DefaultProducer {
if (LOG.isDebugEnabled()) {
LOG.debug("Writing points {}", points);
}
- writeApi.writePoints(bucketName, orgName,
points.getInfluxPoints());
+ List<Point> influxPoints = points.getInfluxPoints();
+ for (int i = 0; influxPoints != null && i < influxPoints.size();
i++) {
+ // a null element has nothing to write, and the client leaves
it out as before
+ if (influxPoints.get(i) != null) {
+ lineProtocolOf(influxPoints.get(i), "The point at index "
+ i + " of the Points body");
+ }
+ }
+ writeApi.writePoints(bucketName, orgName, influxPoints);
+ } catch (CamelInfluxDb2Exception ex) {
+ exchange.setException(ex);
} catch (Exception ex) {
exchange.setException(new CamelInfluxDb2Exception(ex));
}
@@ -176,6 +188,21 @@ public class InfluxDb2Producer extends DefaultProducer {
}
}
+ /**
+ * The line protocol of the point. InfluxDB needs at least one field, and
the client skips a point that has no field
+ * with a value (no field, or only fields whose value is null, NaN or
infinite) without an error, so such a point
+ * fails the exchange instead of completing it with nothing written.
+ */
+ private static String lineProtocolOf(Point point, String name) {
+ String line = point.toLineProtocol();
+ if (line.isEmpty()) {
+ throw new CamelInfluxDb2Exception(
+ name + " has no field with a value: InfluxDB needs at
least one field, and the client would skip"
+ + " the point without writing
it");
+ }
+ return line;
+ }
+
private void doPing(Exchange exchange) {
Boolean result = connection.ping();
MessageHelper.copyHeaders(exchange.getIn(), exchange.getOut(), true);
diff --git
a/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDb2Converters.java
b/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDb2Converters.java
index 203ab5d10a30..d3bc1ca3a26e 100644
---
a/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDb2Converters.java
+++
b/components/camel-influxdb2/src/main/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDb2Converters.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.influxdb2.converters;
+import java.util.LinkedHashMap;
import java.util.Map;
import com.influxdb.client.write.Point;
@@ -40,12 +41,10 @@ public final class CamelInfluxDb2Converters {
throw new CamelInfluxDb2Exception(format);
}
- String measurenmentNameString = measurementName.toString();
- Point point = Point.measurement(measurenmentNameString);
- map.remove(InfluxDb2Constants.MEASUREMENT);
- point.addFields(map);
- map.put(InfluxDb2Constants.MEASUREMENT, measurementName);
-
- return point;
+ // 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(InfluxDb2Constants.MEASUREMENT);
+ return Point.measurement(measurementName.toString()).addFields(fields);
}
}
diff --git
a/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2EndpointStartupTest.java
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2EndpointStartupTest.java
new file mode 100644
index 000000000000..f85e059de207
--- /dev/null
+++
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2EndpointStartupTest.java
@@ -0,0 +1,247 @@
+/*
+ * 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.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.influxdb.client.InfluxDBClient;
+import com.influxdb.client.InfluxDBClientFactory;
+import com.influxdb.client.write.Point;
+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.test.junit6.CamelTestSupport;
+import org.apache.camel.util.ObjectHelper;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The endpoint calls InfluxDB when it starts only to create the organization
or the bucket when it is missing. A local
+ * HTTP server plays the organizations, buckets and write APIs of InfluxDB 2.x
and records the requests it receives.
+ */
+class InfluxDb2EndpointStartupTest extends CamelTestSupport {
+
+ private static final String ORG_ID = "00000000000000a1";
+ private static final String BUCKET_JSON
+ = "{\"id\":\"00000000000000b1\",\"orgID\":\"" + ORG_ID +
"\",\"name\":\"myBucket\",\"retentionRules\":[]}";
+
+ private final AtomicBoolean orgExists = new AtomicBoolean(true);
+ private final AtomicBoolean bucketExists = new AtomicBoolean(true);
+ private final List<String> requests = new CopyOnWriteArrayList<>();
+ private final List<String> createdBuckets = new CopyOnWriteArrayList<>();
+ private final List<String> writeQueries = new CopyOnWriteArrayList<>();
+
+ private HttpServer server;
+ private InfluxDBClient client;
+ private InfluxDBClient clientOfAServerThatIsDown;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ server = HttpServer.create(new
InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+ server.createContext("/api/v2/orgs", exchange -> {
+ String method = record(exchange);
+ if ("POST".equals(method)) {
+ orgExists.set(true);
+ respond(exchange, 201, "{\"id\":\"" + ORG_ID +
"\",\"name\":\"myOrg\"}");
+ } else if (orgExists.get()) {
+ respond(exchange, 200, "{\"orgs\":[{\"id\":\"" + ORG_ID +
"\",\"name\":\"myOrg\"}]}");
+ } else {
+ respond(exchange, 404, "{\"code\":\"not
found\",\"message\":\"organization name not found\"}");
+ }
+ });
+ server.createContext("/api/v2/buckets", exchange -> {
+ String body = new String(exchange.getRequestBody().readAllBytes(),
StandardCharsets.UTF_8);
+ String method = record(exchange);
+ if ("POST".equals(method)) {
+ createdBuckets.add(body);
+ bucketExists.set(true);
+ respond(exchange, 201, BUCKET_JSON);
+ } else if (!orgExists.get()) {
+ respond(exchange, 404, "{\"code\":\"not
found\",\"message\":\"organization name not found\"}");
+ } else {
+ respond(exchange, 200, "{\"buckets\":[" + (bucketExists.get()
? BUCKET_JSON : "") + "]}");
+ }
+ });
+ server.createContext("/api/v2/write", exchange -> {
+ record(exchange);
+
writeQueries.add(String.valueOf(exchange.getRequestURI().getQuery()));
+ respond(exchange, 204, null);
+ });
+ server.start();
+ client = InfluxDBClientFactory.create("http://127.0.0.1:"; +
server.getAddress().getPort(), "token".toCharArray());
+
+ // nothing listens on the port of a socket that has been closed
+ int portOfAServerThatIsDown;
+ try (ServerSocket socket = new ServerSocket(0, 0,
InetAddress.getLoopbackAddress())) {
+ portOfAServerThatIsDown = socket.getLocalPort();
+ }
+ clientOfAServerThatIsDown
+ = InfluxDBClientFactory.create("http://127.0.0.1:"; +
portOfAServerThatIsDown, "token".toCharArray());
+
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("influxDbBean", client);
+ context.getRegistry().bind("influxDbThatIsDown",
clientOfAServerThatIsDown);
+ return context;
+ }
+
+ @Override
+ protected void cleanupResources() {
+ if (server != null) {
+ server.stop(0);
+ }
+ if (client != null) {
+ client.close();
+ }
+ if (clientOfAServerThatIsDown != null) {
+ clientOfAServerThatIsDown.close();
+ }
+ }
+
+ @Test
+ void endpointThatCreatesNeitherDoesNotCallInfluxDbAtStartup() throws
Exception {
+
addRouteTo("influxdb2:influxDbBean?org=myOrg&bucket=myBucket&autoCreateOrg=false&autoCreateBucket=false");
+
+ assertEquals(List.of(), requests, "requests when the route has
started");
+
+ Exchange exchange = template.send("direct:write", e ->
e.getIn().setBody(point()));
+
+ assertNull(exchange.getException());
+ assertEquals(List.of("POST /api/v2/write"), requests, "requests when
the exchange has completed");
+ // the write names the organization and the bucket
+ 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));
+ }
+
+ @Test
+ void routeStartsWhileInfluxDbIsDownWhenTheEndpointCreatesNeither() {
+ assertDoesNotThrow(() -> addRouteTo(
+
"influxdb2:influxDbThatIsDown?org=myOrg&bucket=myBucket&autoCreateOrg=false&autoCreateBucket=false"));
+
+ 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");
+ }
+
+ @Test
+ void defaultsLookTheOrganizationAndTheBucketUpOnce() throws Exception {
+ addRouteTo("influxdb2:influxDbBean?org=myOrg&bucket=myBucket");
+
+ assertEquals(List.of("GET /api/v2/orgs", "GET /api/v2/buckets"),
requests,
+ "requests when the route has started");
+ }
+
+ @Test
+ void defaultsCreateTheMissingOrganizationAndBucket() throws Exception {
+ orgExists.set(false);
+ bucketExists.set(false);
+
+ addRouteTo("influxdb2:influxDbBean?org=myOrg&bucket=myBucket");
+
+ assertEquals(List.of("GET /api/v2/orgs", "POST /api/v2/orgs", "GET
/api/v2/buckets", "POST /api/v2/buckets"),
+ requests, "requests when the route has started");
+ assertTrue(createdBuckets.get(0).contains("\"orgID\":\"" + ORG_ID +
"\""), createdBuckets.get(0));
+ }
+
+ @Test
+ void autoCreateBucketAloneLooksOnlyTheBucketUp() throws Exception {
+
addRouteTo("influxdb2:influxDbBean?org=myOrg&bucket=myBucket&autoCreateOrg=false");
+
+ assertEquals(List.of("GET /api/v2/buckets"), requests, "requests when
the route has started");
+ }
+
+ @Test
+ void autoCreateBucketAloneCreatesTheMissingBucketInTheOrganization()
throws Exception {
+ bucketExists.set(false);
+
+
addRouteTo("influxdb2:influxDbBean?org=myOrg&bucket=myBucket&autoCreateOrg=false");
+
+ assertEquals(List.of("GET /api/v2/buckets", "GET /api/v2/orgs", "POST
/api/v2/buckets"), requests,
+ "requests when the route has started");
+ assertTrue(createdBuckets.get(0).contains("\"orgID\":\"" + ORG_ID +
"\""), createdBuckets.get(0));
+ }
+
+ @Test
+ void autoCreateBucketAloneFailsToStartWhenTheOrganizationDoesNotExist() {
+ orgExists.set(false);
+ bucketExists.set(false);
+
+ Exception failure = assertThrows(Exception.class,
+ () ->
addRouteTo("influxdb2:influxDbBean?org=myOrg&bucket=myBucket&autoCreateOrg=false"));
+
+ CamelInfluxDb2Exception cause =
ObjectHelper.getException(CamelInfluxDb2Exception.class, failure);
+ assertNotNull(cause, "CamelInfluxDb2Exception in the causes of " +
failure);
+ assertEquals("Cannot create the bucket myBucket because the
organization myOrg does not exist and autoCreateOrg is"
+ + " false",
+ cause.getMessage());
+ assertEquals(List.of("GET /api/v2/buckets", "GET /api/v2/orgs"),
requests,
+ "requests when the route has failed to start");
+ }
+
+ private void addRouteTo(String uri) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:write").to(uri);
+ }
+ });
+ }
+
+ private String record(HttpExchange exchange) {
+ String method = exchange.getRequestMethod();
+ requests.add(method + " " + exchange.getRequestURI().getPath());
+ return method;
+ }
+
+ 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/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2ProducerTest.java
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2ProducerTest.java
index dd1dbea6131c..624de1e6c8c5 100644
---
a/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2ProducerTest.java
+++
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/InfluxDb2ProducerTest.java
@@ -78,7 +78,7 @@ public class InfluxDb2ProducerTest extends
AbstractInfluxDbTest {
}
});
- Point point = Point.measurement("temperature");
+ Point point = Point.measurement("temperature").addField("value", 60.0);
TestSupport.sendBody(template, "direct:test", point);
@@ -174,7 +174,7 @@ public class InfluxDb2ProducerTest extends
AbstractInfluxDbTest {
});
ArrayList<Point> points = new ArrayList<>() {
{
- add(Point.measurement("temperature"));
+ add(Point.measurement("temperature").addField("value", 60.0));
}
};
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
index 4242cc7e889a..9cb815809836 100644
---
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
@@ -21,9 +21,13 @@ import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
@@ -34,12 +38,15 @@ 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.Points;
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.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -191,6 +198,93 @@ class InfluxDb2ProducerWriteFailureTest extends
CamelTestSupport {
after.assertIsSatisfied();
}
+ static Stream<Arguments> pointsWithoutAFieldToWrite() {
+ return Stream.of(
+ Arguments.of("no field",
Point.measurement("temperature").addTag("location", "north")),
+ Arguments.of("a null field",
+ Point.measurement("temperature").addTag("location",
"north").addField("value", (Number) null)),
+ Arguments.of("a NaN field",
+ Point.measurement("temperature").addTag("location",
"north").addField("value", Double.NaN)),
+ Arguments.of("an infinite field",
+ Point.measurement("temperature").addTag("location",
"north")
+ .addField("value", Double.POSITIVE_INFINITY)));
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("pointsWithoutAFieldToWrite")
+ void pointWithoutAFieldToWriteFailsTheExchange(String description, Point
point) throws Exception {
+ 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 point has " + description);
+ assertEquals("The point has no field with a value: InfluxDB needs at
least one field, and the client would skip"
+ + " the point without writing it",
+ failure.getMessage());
+ assertEquals(List.of(), writes, "write requests at the server");
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void mapWithOnlyTheMeasurementNameFailsTheExchange() throws Exception {
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(0);
+ Map<String, Object> body = new HashMap<>();
+ body.put(InfluxDb2Constants.MEASUREMENT, "temperature");
+
+ Exchange exchange = template.send("direct:write", e ->
e.getIn().setBody(body));
+
+ assertInstanceOf(CamelInfluxDb2Exception.class,
exchange.getException(),
+ "exception of an exchange whose map has no field");
+ assertEquals(List.of(), writes, "write requests at the server");
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void pointsWithAPointWithoutFieldsFailTheExchangeAndNothingIsWritten()
throws Exception {
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(0);
+ Points points = Points.create(List.of(point(),
Point.measurement("temperature").addTag("location", "south")));
+
+ Exchange exchange = template.send("direct:write", e ->
e.getIn().setBody(points));
+
+ CamelInfluxDb2Exception failure =
assertInstanceOf(CamelInfluxDb2Exception.class, exchange.getException(),
+ "exception of an exchange whose second point has no field");
+ assertTrue(failure.getMessage().startsWith("The point at index 1 of
the Points body has no field with a value"),
+ failure.getMessage());
+ assertEquals(List.of(), writes, "write requests at the server");
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void nullPointOfPointsIsLeftOutAsBefore() throws Exception {
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(1);
+
+ Exchange exchange = template.send("direct:write", e ->
e.getIn().setBody(Points.create(Arrays.asList(point(), null))));
+
+ assertNull(exchange.getException());
+ assertEquals(List.of("temperature,location=north value=60.0"), writes,
+ "write requests at the server when the exchange has
completed");
+ after.assertIsSatisfied();
+ }
+
+ @Test
+ void immutableMapIsWritten() throws Exception {
+ MockEndpoint after = getMockEndpoint("mock:after");
+ after.expectedMessageCount(1);
+
+ Exchange exchange = template.send("direct:write",
+ e -> e.getIn().setBody(Map.of(InfluxDb2Constants.MEASUREMENT,
"temperature", "value", 60.0)));
+
+ assertNull(exchange.getException());
+ assertEquals(List.of("temperature value=60.0"), writes,
+ "write requests at the server when the exchange has
completed");
+ after.assertIsSatisfied();
+ }
+
private static Point point() {
return Point.measurement("temperature").addTag("location",
"north").addField("value", 60.0);
}
diff --git
a/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDbConverterTest.java
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDbConverterTest.java
index 9eb33d60aa0c..3a68f06bfff9 100644
---
a/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDbConverterTest.java
+++
b/components/camel-influxdb2/src/test/java/org/apache/camel/component/influxdb2/converters/CamelInfluxDbConverterTest.java
@@ -17,6 +17,8 @@
package org.apache.camel.component.influxdb2.converters;
import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import com.influxdb.client.write.Point;
@@ -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(InfluxDb2Constants.MEASUREMENT, "testCPU", "busy", 99);
+
+ Point p = CamelInfluxDb2Converters.fromMapToPoint(pointInMapFormat);
+
+ assertEquals("testCPU busy=99i", p.toLineProtocol());
+ }
+
+ @Test
+ void doesNotChangeTheMapOfTheCaller() {
+ Map<String, Object> pointInMapFormat = new LinkedHashMap<>();
+ pointInMapFormat.put(InfluxDb2Constants.MEASUREMENT, "testCPU");
+ pointInMapFormat.put("busy", 99);
+ Map<String, Object> before = new LinkedHashMap<>(pointInMapFormat);
+
+ Point p = CamelInfluxDb2Converters.fromMapToPoint(pointInMapFormat);
+
+ assertEquals("testCPU busy=99i", p.toLineProtocol());
+ 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 d13479088c32..eead8427aa02 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
@@ -4713,6 +4713,20 @@ by the time the next step read it. They now behave as
follows:
* `xmlSecurity`: the catalog now gives the defaults the data format uses:
`secureTagContents` is `true`, and
`keyCipherAlgorithm` has no default. The behaviour is unchanged.
+=== camel-influxdb2 - a point without fields fails the exchange, and InfluxDB
is called at startup only to create the organization or bucket
+
+A `Point` without a field that has a value (no field at all, or only fields
whose value is `null`, NaN or infinite) now
+fails the exchange with a `CamelInfluxDb2Exception`. This also applies to a
`Point` converted from a `Map` body and to
+each point of a `Points` body. The InfluxDB client skipped such a point, so
the exchange completed while nothing was
+written.
+
+The endpoint now looks the organization up when it starts only when
`autoCreateOrg` is `true` (the default), and the
+bucket only when `autoCreateBucket` is `true` (the default). With both options
set to `false` it no longer calls
+InfluxDB when it starts, so the route starts while InfluxDB is down, and an
organization or a bucket that does not exist
+fails the first write instead of being looked up. With `autoCreateOrg=false`
and `autoCreateBucket=true` the
+organization is looked up only when the bucket must be created, and the
endpoint fails to start with a message that
+names them when that organization does not exist.
+
=== camel-mail - the text of multipart mails, the charset of sent text, and
producers on imap or pop3
A multipart mail without a text part, such as one with only a PDF attachment,
now converts to a `String` or an
