This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26153-hivemq-body-payload in repository https://gitbox.apache.org/repos/asf/camel.git
commit 46da21ba0c7e79d20241a9c5eedca70f00222203 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 17:44:12 2026 +0100 CAMEL-26153: camel-hivemq - fail the exchange for a body that does not 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 1bc188419d61..f50c6bd9e6b5 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 @@ -707,6 +707,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
