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 f5475dfc57c9 CAMEL-26153: camel-hivemq - fail the exchange for a body
that does not convert to byte[] instead of publishing an empty payload (#28924)
f5475dfc57c9 is described below
commit f5475dfc57c9809f5d19a2bc862a5c3178e03ad6
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:48:01 2026 +0100
CAMEL-26153: camel-hivemq - fail the exchange for a body that does not
convert to byte[] instead of publishing an empty payload (#28924)
…t convert to byte[] instead of publishing an empty payload
The producer read the body with the non-mandatory getBody(byte[].class) and
replaced null with an empty array, so a Map, POJO or number body was
published
as an empty payload without an error. It now uses
getMandatoryBody(byte[].class)
when there is a body, and keeps the empty payload only for no body. A
failure
while reading the headers or the body now fails the exchange through the
async
callback. Body types, component doc and upgrade guide updated.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/components/hivemq.json | 2 +-
.../camel/catalog/docs/hivemq-component.adoc | 4 +
.../org/apache/camel/component/hivemq/hivemq.json | 2 +-
.../src/main/docs/hivemq-component.adoc | 4 +
.../camel/component/hivemq/HiveMQEndpoint.java | 3 +-
.../camel/component/hivemq/HiveMQProducer.java | 27 ++-
.../component/hivemq/HiveMQProducerBodyTest.java | 187 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 +
8 files changed, 225 insertions(+), 11 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hivemq.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hivemq.json
index c7984eddd189..db9154c22ec2 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hivemq.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/hivemq.json
@@ -47,7 +47,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"description": "The payload of the MQTT message, an empty byte array for an
empty payload. The topic is in the CamelHiveMQTopic header" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as the payload; no body is sent as an empty payload. A Map, a POJO or a
number does not convert and is sent as an empty payload too, without an error:
marshal it first" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as the payload; no body is sent as an empty payload. A Map, a POJO or a
number does not convert and fails the exchange: marshal it first" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hivemq-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hivemq-component.adoc
index 6438de7e6a8f..76f5624ad4aa 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hivemq-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/hivemq-component.adoc
@@ -76,6 +76,10 @@ String payload = "message";
producerTemplate.sendBody("hivemq:topic", payload);
----
+The producer sends what converts to a byte array: text, bytes, a stream or a
file. A message without
+a body is sent as an empty payload. A body that does not convert to a byte
array, such as a `Map`, a POJO
+or a number, fails the exchange with an `InvalidPayloadException`: marshal it
first, for example to JSON.
+
== Examples
For example, the following snippet reads messages from an MQTT broker
diff --git
a/components/camel-hivemq/src/generated/resources/META-INF/org/apache/camel/component/hivemq/hivemq.json
b/components/camel-hivemq/src/generated/resources/META-INF/org/apache/camel/component/hivemq/hivemq.json
index c7984eddd189..db9154c22ec2 100644
---
a/components/camel-hivemq/src/generated/resources/META-INF/org/apache/camel/component/hivemq/hivemq.json
+++
b/components/camel-hivemq/src/generated/resources/META-INF/org/apache/camel/component/hivemq/hivemq.json
@@ -47,7 +47,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]",
"description": "The payload of the MQTT message, an empty byte array for an
empty payload. The topic is in the CamelHiveMQTopic header" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as the payload; no body is sent as an empty payload. A Map, a POJO or a
number does not convert and is sent as an empty payload too, without an error:
marshal it first" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "null" ],
"description": "What converts to a byte array: text, bytes, a stream or a file,
sent as the payload; no body is sent as an empty payload. A Map, a POJO or a
number does not convert and fails the exchange: marshal it first" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git a/components/camel-hivemq/src/main/docs/hivemq-component.adoc
b/components/camel-hivemq/src/main/docs/hivemq-component.adoc
index 6438de7e6a8f..76f5624ad4aa 100644
--- a/components/camel-hivemq/src/main/docs/hivemq-component.adoc
+++ b/components/camel-hivemq/src/main/docs/hivemq-component.adoc
@@ -76,6 +76,10 @@ String payload = "message";
producerTemplate.sendBody("hivemq:topic", payload);
----
+The producer sends what converts to a byte array: text, bytes, a stream or a
file. A message without
+a body is sent as an empty payload. A body that does not convert to a byte
array, such as a `Map`, a POJO
+or a number, fails the exchange with an `InvalidPayloadException`: marshal it
first, for example to JSON.
+
== Examples
For example, the following snippet reads messages from an MQTT broker
diff --git
a/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQEndpoint.java
b/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQEndpoint.java
index d93321f93ca7..cb1c7250b7bf 100644
---
a/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQEndpoint.java
+++
b/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQEndpoint.java
@@ -43,8 +43,7 @@ import org.apache.camel.support.DefaultEndpoint;
producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE, Carrier.NULL },
description = "What converts to a byte
array: text, bytes, a stream or a file, sent as"
+ " the payload; no body is
sent as an empty payload. A Map, a POJO or a"
- + " number does not convert
and is sent as an empty payload too, without"
- + " an error: marshal it
first"),
+ + " number does not convert
and fails the exchange: marshal it first"),
producerOut = @BodyType(Carrier.UNCHANGED))
public class HiveMQEndpoint extends DefaultEndpoint implements
EndpointServiceLocation {
diff --git
a/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQProducer.java
b/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQProducer.java
index f233bb6e7398..2d1476cb568f 100644
---
a/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQProducer.java
+++
b/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQProducer.java
@@ -61,14 +61,27 @@ public class HiveMQProducer extends DefaultAsyncProducer {
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
- String targetTopic =
exchange.getIn().getHeader(HiveMQConstants.OVERRIDE_TOPIC, endpoint.getTopic(),
String.class);
- MqttQos qos = exchange.getIn().getHeader(HiveMQConstants.MQTT_QOS,
endpoint.getConfiguration().getQos(), MqttQos.class);
- boolean retained =
exchange.getIn().getHeader(HiveMQConstants.MQTT_RETAINED,
endpoint.getConfiguration().isRetained(),
- Boolean.class);
+ String targetTopic;
+ MqttQos qos;
+ boolean retained;
+ byte[] payload;
+ try {
+ targetTopic =
exchange.getIn().getHeader(HiveMQConstants.OVERRIDE_TOPIC, endpoint.getTopic(),
String.class);
+ qos = exchange.getIn().getHeader(HiveMQConstants.MQTT_QOS,
endpoint.getConfiguration().getQos(), MqttQos.class);
+ retained =
exchange.getIn().getHeader(HiveMQConstants.MQTT_RETAINED,
endpoint.getConfiguration().isRetained(),
+ Boolean.class);
- byte[] payload = exchange.getIn().getBody(byte[].class);
- if (payload == null) {
- payload = new byte[0];
+ // no body is an empty payload, but a body that does not convert
to byte[] must fail instead of
+ // being published as an empty payload
+ if (exchange.getIn().getBody() == null) {
+ payload = new byte[0];
+ } else {
+ payload = exchange.getIn().getMandatoryBody(byte[].class);
+ }
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
}
client.publish(targetTopic, payload, qos, retained)
diff --git
a/components/camel-hivemq/src/test/java/org/apache/camel/component/hivemq/HiveMQProducerBodyTest.java
b/components/camel-hivemq/src/test/java/org/apache/camel/component/hivemq/HiveMQProducerBodyTest.java
new file mode 100644
index 000000000000..9b7521b8524c
--- /dev/null
+++
b/components/camel-hivemq/src/test/java/org/apache/camel/component/hivemq/HiveMQProducerBodyTest.java
@@ -0,0 +1,187 @@
+/*
+ * 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.hivemq;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.TypeConversionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The producer publishes what converts to a byte array, an empty payload for
no body, and fails the exchange for a body
+ * that does not convert, instead of publishing it as an empty payload.
+ */
+class HiveMQProducerBodyTest {
+
+ private final List<byte[]> published = new CopyOnWriteArrayList<>();
+ private CamelContext context;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ context = createContext();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to("hivemq:test/topic");
+ }
+ });
+ context.start();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ context.close();
+ }
+
+ @Test
+ void textBodyIsPublished() {
+ Exchange out = send("Hello");
+
+ assertThat(out.getException()).isNull();
+ assertThat(published).singleElement()
+ .isEqualTo("Hello".getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void noBodyIsPublishedAsEmptyPayload() {
+ Exchange out = send(null);
+
+ assertThat(out.getException()).isNull();
+ assertThat(published).singleElement()
+ .satisfies(payload -> assertThat(payload).isEmpty());
+ }
+
+ @Test
+ void mapBodyFailsTheExchange() {
+ Exchange out = send(Map.of("id", 1));
+
+
assertThat(out.getException()).isInstanceOf(InvalidPayloadException.class);
+ assertThat(published).isEmpty();
+ }
+
+ @Test
+ void numberBodyFailsTheExchange() {
+ Exchange out = send(42);
+
+
assertThat(out.getException()).isInstanceOf(InvalidPayloadException.class);
+ assertThat(published).isEmpty();
+ }
+
+ @Test
+ void pojoBodyFailsTheExchange() {
+ Exchange out = send(new Order("A-1"));
+
+
assertThat(out.getException()).isInstanceOf(InvalidPayloadException.class);
+ assertThat(published).isEmpty();
+ }
+
+ @Test
+ void invalidQosHeaderFailsTheExchange() {
+ Exchange out =
context.createProducerTemplate().request("direct:start", e -> {
+ e.getIn().setBody("Hello");
+ e.getIn().setHeader(HiveMQConstants.MQTT_QOS, "NO_SUCH_QOS");
+ });
+
+
assertThat(out.getException()).isInstanceOf(TypeConversionException.class);
+ assertThat(published).isEmpty();
+ }
+
+ private Exchange send(Object body) {
+ return context.createProducerTemplate().request("direct:start", e ->
e.getIn().setBody(body));
+ }
+
+ private CamelContext createContext() {
+ CamelContext camelContext = new DefaultCamelContext();
+ camelContext.addComponent("hivemq", new HiveMQComponent() {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters)
+ throws Exception {
+ HiveMQEndpoint endpoint = new HiveMQEndpoint(uri, this,
getConfiguration().copy(), remaining) {
+ @Override
+ HiveMQClientAdapter createClient() {
+ return new RecordingClient();
+ }
+ };
+ setProperties(endpoint, parameters);
+ return endpoint;
+ }
+ });
+ return camelContext;
+ }
+
+ private record Order(String id) {
+ }
+
+ private final class RecordingClient implements HiveMQClientAdapter {
+ @Override
+ public CompletableFuture<?> connect(boolean cleanStart) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void stop() {
+ }
+
+ @Override
+ public boolean isConnected() {
+ return true;
+ }
+
+ @Override
+ public boolean isConnectedOrReconnecting() {
+ return true;
+ }
+
+ @Override
+ public CompletableFuture<?> subscribe(String topicFilter, MqttQos qos,
Consumer<HiveMQMessage> callback) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<?> unsubscribe(String topicFilter) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<?> publish(String topic, byte[] payload,
MqttQos qos, boolean retained) {
+ published.add(payload);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public <T> Optional<T> getClient(Class<T> clazz) {
+ return Optional.empty();
+ }
+ }
+}
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 e173917f8172..793cc1657bd6 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
@@ -739,6 +739,13 @@ configuration (the `camel.component.google-mail.*`
options), kept until the comp
client the component had already created, for example the one of a
`google-mail:messages/modify` endpoint, so
credentials set only on an endpoint URI no longer reach the transformer: set
them on the component as well.
+=== camel-hivemq - a body that does not convert to a byte array fails the
exchange
+
+The `hivemq` producer published a message body that has no conversion to
`byte[]`, such as a `Map`, a POJO or a
+number, as an empty payload, without an error, so the data was lost. It now
fails the exchange with an
+`InvalidPayloadException`. Marshal such a body first (for example to JSON) or
convert it to a `String`. A message
+without a body is still published as an empty payload.
+
=== camel-jcache - a JCacheManager closes only what no other manager uses
`JCacheManager.close()`, which a `jcache` endpoint calls when it stops, closed
its cache, its `CacheManager` and the
