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 0d1d5ab0f989 CAMEL-26152: camel-sjms - fail the message when an InOut
reply cannot be sent, and support jmsMessageType=Stream (#28925)
0d1d5ab0f989 is described below
commit 0d1d5ab0f989eb858c3531d195925aaa0e23fca1
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 21:54:43 2026 +0100
CAMEL-26152: camel-sjms - fail the message when an InOut reply cannot be
sent, and support jmsMessageType=Stream (#28925)
… be sent, and support jmsMessageType=Stream
When the consumer could not send an InOut reply, sendReply set a
CamelExchangeException on the exchange, but onMessage only rethrew a
RuntimeCamelException, so the request was committed or acknowledged
with no reply and no log. The error is now rethrown as for a failed
route: rolled back when transacted (or recovered with
CLIENT_ACKNOWLEDGE) and logged, as camel-jms does.
JmsBinding had no case for jmsMessageType=Stream and returned null,
which failed with a NullPointerException. It now writes the body to a
StreamMessage as camel-jms does, and a type that cannot be sent fails
with a JMSException naming the type. camel-sjms2 inherits both fixes.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../org/apache/camel/catalog/components/sjms.json | 2 +-
.../org/apache/camel/catalog/components/sjms2.json | 2 +-
.../apache/camel/catalog/docs/sjms-component.adoc | 5 ++
.../apache/camel/catalog/docs/sjms2-component.adoc | 5 ++
.../org/apache/camel/component/sjms/sjms.json | 2 +-
.../camel-sjms/src/main/docs/sjms-component.adoc | 5 ++
.../apache/camel/component/sjms/SjmsEndpoint.java | 5 +-
.../sjms/consumer/EndpointMessageListener.java | 6 ++
.../camel/component/sjms/jms/JmsBinding.java | 34 ++++++++++-
.../consumer/InOutConsumerReplyFailureTest.java | 71 ++++++++++++++++++++++
.../sjms/producer/JmsMessageTypeTest.java | 37 +++++++++++
.../org/apache/camel/component/sjms2/sjms2.json | 2 +-
.../camel-sjms2/src/main/docs/sjms2-component.adoc | 5 ++
.../camel/component/sjms2/Sjms2Endpoint.java | 5 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++++
15 files changed, 186 insertions(+), 11 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json
index cb8e0b845246..3189ea01e657 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms.json
@@ -49,7 +49,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ],
"byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [
"pojo" ], "javaType": "jakarta.jms.Message" }, { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map",
"value", "list", "pojo", "null" ] } ], "description": "By the JMS message
received: a String for a TextMessage, a byte array for a BytesMessage, a Map
for a MapMessage, the jakarta.jms.StreamMessage its [...]
- "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, is dropped and the
caller times out" },
+ "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, fails the incoming
message (rolled back when transacted [...]
"producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ {
"option": "objectMessageEnabled", "value": "true", "carriers": [ "text",
"bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ],
"description": "Text is sent as a TextMessage; bytes, a stream or a file as a
BytesMessage; a Map of simple values as a MapMessage; null as an empty Message.
A number, a List or another Serial [...]
"producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, {
"option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ],
"description": "Unchanged for a message that expects no reply (InOnly). A
message that expects one (InOut) gets the body of the reply, as the consumer
gives it" }
},
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json
index 74c702d69998..85fe927f5c0f 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/sjms2.json
@@ -49,7 +49,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ],
"byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [
"pojo" ], "javaType": "jakarta.jms.Message" }, { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map",
"value", "list", "pojo", "null" ] } ], "description": "By the JMS message
received: a String for a TextMessage, a byte array for a BytesMessage, a Map
for a MapMessage, the jakarta.jms.StreamMessage its [...]
- "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, is dropped and the
caller times out" },
+ "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, fails the incoming
message (rolled back when transacted [...]
"producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ {
"option": "objectMessageEnabled", "value": "true", "carriers": [ "text",
"bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ],
"description": "Text is sent as a TextMessage; bytes, a stream or a file as a
BytesMessage; a Map of simple values as a MapMessage; null as an empty Message.
A number, a List or another Serial [...]
"producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, {
"option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ],
"description": "Unchanged for a message that expects no reply (InOnly). A
message that expects one (InOut) gets the body of the reply, as the consumer
gives it" }
},
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc
index c779d4ded2d8..c00b95fc0bcb 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms-component.adoc
@@ -307,6 +307,7 @@ The body is mapped to and from a JMS message as follows
(unless the option `jmsM
| Any other object | `Message` without a body (a WARN is logged) | `null`
|===
+With `jmsMessageType=Stream` the body is converted to an `InputStream`, and
its bytes are written to a `StreamMessage`.
A received `StreamMessage` is not mapped: the body is the
`jakarta.jms.StreamMessage` itself.
With `mapJmsMessage=false` the body is always the received
`jakarta.jms.Message`.
@@ -327,6 +328,10 @@ The _InOut_ Consumer uses this strategy as well ensuring
that all
response messages to the included `JMSReplyTo` destination also have the
`JMSCorrelationId` copied from the request as well.
+When the _InOut_ consumer cannot send the reply (for example, as its body
cannot be mapped to a JMS message), the
+incoming message fails as if the route had failed: it is rolled back when
`transacted=true` (or recovered with
+`CLIENT_ACKNOWLEDGE`), and the error is logged. No reply is sent, so the
caller waits until its `requestTimeout`.
+
=== Batch consuming
The consumer can be configured to receive messages in batches instead of one
at a time, by setting
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc
index e24128751a3f..9f7370ade30e 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sjms2-component.adoc
@@ -311,6 +311,7 @@ The body is mapped to and from a JMS message as follows
(unless the option `jmsM
| Any other object | `Message` without a body (a WARN is logged) | `null`
|===
+With `jmsMessageType=Stream` the body is converted to an `InputStream`, and
its bytes are written to a `StreamMessage`.
A received `StreamMessage` is not mapped: the body is the
`jakarta.jms.StreamMessage` itself.
With `mapJmsMessage=false` the body is always the received
`jakarta.jms.Message`.
@@ -331,6 +332,10 @@ The _InOut_ Consumer uses this strategy as well ensuring
that all
response messages to the included `JMSReplyTo` destination also have the
`JMSCorrelationId` copied from the request as well.
+When the _InOut_ consumer cannot send the reply (for example, as its body
cannot be mapped to a JMS message), the
+incoming message fails as if the route had failed: it is rolled back when
`transacted=true` (or recovered with
+`CLIENT_ACKNOWLEDGE`), and the error is logged. No reply is sent, so the
caller waits until its `requestTimeout`.
+
=== Batch consuming
The consumer can be configured to receive messages in batches instead of one
at a time, by setting
diff --git
a/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json
b/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json
index cb8e0b845246..3189ea01e657 100644
---
a/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json
+++
b/components/camel-sjms/src/generated/resources/META-INF/org/apache/camel/component/sjms/sjms.json
@@ -49,7 +49,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ],
"byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [
"pojo" ], "javaType": "jakarta.jms.Message" }, { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map",
"value", "list", "pojo", "null" ] } ], "description": "By the JMS message
received: a String for a TextMessage, a byte array for a BytesMessage, a Map
for a MapMessage, the jakarta.jms.StreamMessage its [...]
- "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, is dropped and the
caller times out" },
+ "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, fails the incoming
message (rolled back when transacted [...]
"producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ {
"option": "objectMessageEnabled", "value": "true", "carriers": [ "text",
"bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ],
"description": "Text is sent as a TextMessage; bytes, a stream or a file as a
BytesMessage; a Map of simple values as a MapMessage; null as an empty Message.
A number, a List or another Serial [...]
"producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, {
"option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ],
"description": "Unchanged for a message that expects no reply (InOnly). A
message that expects one (InOut) gets the body of the reply, as the consumer
gives it" }
},
diff --git a/components/camel-sjms/src/main/docs/sjms-component.adoc
b/components/camel-sjms/src/main/docs/sjms-component.adoc
index c779d4ded2d8..c00b95fc0bcb 100644
--- a/components/camel-sjms/src/main/docs/sjms-component.adoc
+++ b/components/camel-sjms/src/main/docs/sjms-component.adoc
@@ -307,6 +307,7 @@ The body is mapped to and from a JMS message as follows
(unless the option `jmsM
| Any other object | `Message` without a body (a WARN is logged) | `null`
|===
+With `jmsMessageType=Stream` the body is converted to an `InputStream`, and
its bytes are written to a `StreamMessage`.
A received `StreamMessage` is not mapped: the body is the
`jakarta.jms.StreamMessage` itself.
With `mapJmsMessage=false` the body is always the received
`jakarta.jms.Message`.
@@ -327,6 +328,10 @@ The _InOut_ Consumer uses this strategy as well ensuring
that all
response messages to the included `JMSReplyTo` destination also have the
`JMSCorrelationId` copied from the request as well.
+When the _InOut_ consumer cannot send the reply (for example, as its body
cannot be mapped to a JMS message), the
+incoming message fails as if the route had failed: it is rolled back when
`transacted=true` (or recovered with
+`CLIENT_ACKNOWLEDGE`), and the error is logged. No reply is sent, so the
caller waits until its `requestTimeout`.
+
=== Batch consuming
The consumer can be configured to receive messages in batches instead of one
at a time, by setting
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
index 32c238cedb20..62fdbc00b175 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
@@ -94,8 +94,9 @@ import org.slf4j.LoggerFactory;
Carrier.FILE, Carrier.MAP, Carrier.VALUE, Carrier.LIST,
Carrier.POJO, Carrier.NULL }),
description = "The reply, sent as the
producer sends a message. A reply that cannot be"
- + " sent, such as a number
or a List without objectMessageEnabled, is"
- + " dropped and the caller
times out"),
+ + " sent, such as a number
or a List without objectMessageEnabled,"
+ + " fails the incoming
message (rolled back when transacted)"
+ + " and the caller times
out"),
producerIn = @BodyType(value = {
Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
Carrier.MAP,
Carrier.NULL },
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
index 012e353972ff..fcc1da6368fb 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
@@ -450,6 +450,12 @@ public class EndpointMessageListener implements
SessionMessageListener {
sendReply(session, (String) replyDestination, message,
exchange, body, cause);
}
LOG.trace("onMessage.sendReply END");
+ // sendReply sets an exception on the exchange if the reply
could not be sent, so rethrow it
+ // to rollback the JMS message (if transacted or client
acknowledge) and log the error
+ Exception replyException = exchange.getException();
+ if (replyException != null && replyException != cause) {
+ rce = wrapRuntimeCamelException(replyException);
+ }
}
// if an exception occurred
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java
index ed1678b723c0..485a7c75e187 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/jms/JmsBinding.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.sjms.jms;
import java.io.File;
+import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputFilter;
import java.io.Reader;
@@ -56,6 +57,8 @@ import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.DefaultExchangeHolder;
import org.apache.camel.support.DeserializationFilterHelper;
import org.apache.camel.support.ExchangeHelper;
+import org.apache.camel.util.FileUtil;
+import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -523,7 +526,8 @@ public class JmsBinding {
*
* Create the {@link Message}
*
- * @return jmsMessage or null if the mapping was not successfully
+ * @return jmsMessage
+ * @throws JMSException if the message cannot be created, or the type is
not supported
*/
protected Message createJmsMessageForType(
Exchange exchange, Object body, Session session, CamelContext
context,
@@ -554,7 +558,7 @@ public class JmsBinding {
}
return message;
}
- case Object:
+ case Object: {
if (!isObjectMessageEnabled()) {
throw objectMessageDisabled("creating ObjectMessage");
}
@@ -572,10 +576,34 @@ public class JmsBinding {
}
}
return message;
+ }
+ case Stream: {
+ StreamMessage message = session.createStreamMessage();
+ if (body != null) {
+ InputStream is = null;
+ try {
+ is =
context.getTypeConverter().mandatoryConvertTo(InputStream.class, exchange,
body);
+ byte[] buffer = new byte[FileUtil.BUFFER_SIZE];
+ int len;
+ while ((len = is.read(buffer)) >= 0) {
+ message.writeBytes(buffer, 0, len);
+ }
+ } catch (NoTypeConversionAvailableException | IOException
e) {
+ // cannot convert to inputstream then thrown an
exception to avoid sending a null message
+ JMSException cause = new
MessageFormatException(e.getMessage());
+ cause.initCause(e);
+ throw cause;
+ } finally {
+ IOHelper.close(is);
+ }
+ }
+ return message;
+ }
default:
break;
}
- return null;
+ throw new JMSException(
+ "Cannot create JMS message of type " + type + ". Supported
types are: Bytes, Map, Object, Stream, Text.");
}
/**
diff --git
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOutConsumerReplyFailureTest.java
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOutConsumerReplyFailureTest.java
new file mode 100644
index 000000000000..fb1e697eaaba
--- /dev/null
+++
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/consumer/InOutConsumerReplyFailureTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.sjms.consumer;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ExchangeTimedOutException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.sjms.support.JmsTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * An InOut reply that cannot be turned into a JMS message must fail the JMS
message (rollback when transacted), and not
+ * be dropped silently while the request is committed.
+ */
+public class InOutConsumerReplyFailureTest extends JmsTestSupport {
+
+ private final AtomicInteger counter = new AtomicInteger();
+
+ @Override
+ protected boolean useJmx() {
+ return false;
+ }
+
+ @Test
+ public void testReplyFailureRollsBack() {
+ CamelExecutionException e = assertThrows(CamelExecutionException.class,
+ () -> template.requestBody(
+
"sjms:queue:in.out.queue.InOutConsumerReplyFailureTest?replyTo=in.out.queue.InOutConsumerReplyFailureTest.reply&requestTimeout=2000",
+ "Hello World"));
+ assertInstanceOf(ExchangeTimedOutException.class, e.getCause());
+
+ // the reply could not be sent, so the request is rolled back and
redelivered
+ await().atMost(10, TimeUnit.SECONDS).until(() -> counter.get() >= 2);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+
from("sjms:queue:in.out.queue.InOutConsumerReplyFailureTest?transacted=true")
+ .process(exchange -> {
+ counter.incrementAndGet();
+ // an Integer is Serializable, and ObjectMessage
is disabled by default
+ exchange.getMessage().setBody(123);
+ });
+ }
+ };
+ }
+
+}
diff --git
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/JmsMessageTypeTest.java
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/JmsMessageTypeTest.java
index 2df3bf604e74..3c08415e9e3a 100644
---
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/JmsMessageTypeTest.java
+++
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/JmsMessageTypeTest.java
@@ -17,21 +17,27 @@
package org.apache.camel.component.sjms.producer;
import jakarta.jms.BytesMessage;
+import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.MessageConsumer;
+import jakarta.jms.StreamMessage;
+import org.apache.camel.CamelExecutionException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.sjms.support.JmsTestSupport;
import org.apache.camel.util.ObjectHelper;
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.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class JmsMessageTypeTest extends JmsTestSupport {
private static final String TEST_DESTINATION_NAME =
"test.foo.JmsMessageTypeTest";
+ private static final String TEST_STREAM_DESTINATION_NAME =
"test.stream.JmsMessageTypeTest";
public JmsMessageTypeTest() {
}
@@ -63,12 +69,43 @@ public class JmsMessageTypeTest extends JmsTestSupport {
mc.close();
}
+ @Test
+ public void testJmsMessageTypeStream() throws Exception {
+ MessageConsumer mc = createQueueConsumer(TEST_STREAM_DESTINATION_NAME);
+ assertNotNull(mc);
+ final String expectedBody = "Hello World!";
+
+ template.sendBody("direct:stream", expectedBody);
+ Message message = mc.receive(5000);
+ assertNotNull(message);
+ assertInstanceOf(StreamMessage.class, message);
+
+ StreamMessage sm = (StreamMessage) message;
+ byte[] arr = new byte[64];
+ int len = sm.readBytes(arr);
+ assertEquals(12, len);
+ assertEquals(expectedBody, new String(arr, 0, len));
+
+ mc.close();
+ }
+
+ @Test
+ public void testJmsMessageTypeNotSupported() {
+ CamelExecutionException e = assertThrows(CamelExecutionException.class,
+ () ->
template.sendBody("sjms:queue:test.message.JmsMessageTypeTest?jmsMessageType=Message",
"Hello"));
+ JMSException cause = assertInstanceOf(JMSException.class,
e.getCause());
+ assertTrue(cause.getMessage().contains("Cannot create JMS message of
type Message"), cause.getMessage());
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
from("direct:start")
.to("sjms:queue:" + TEST_DESTINATION_NAME +
"?jmsMessageType=Bytes");
+
+ from("direct:stream")
+ .to("sjms:queue:" + TEST_STREAM_DESTINATION_NAME +
"?jmsMessageType=Stream");
}
};
}
diff --git
a/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json
b/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json
index 74c702d69998..85fe927f5c0f 100644
---
a/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json
+++
b/components/camel-sjms2/src/generated/resources/META-INF/org/apache/camel/component/sjms2/sjms2.json
@@ -49,7 +49,7 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "text", "bytes", "map", "pojo", "null" ],
"byOption": [ { "option": "mapJmsMessage", "value": "false", "carriers": [
"pojo" ], "javaType": "jakarta.jms.Message" }, { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes", "map",
"value", "list", "pojo", "null" ] } ], "description": "By the JMS message
received: a String for a TextMessage, a byte array for a BytesMessage, a Map
for a MapMessage, the jakarta.jms.StreamMessage its [...]
- "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, is dropped and the
caller times out" },
+ "consumerReply": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType" ], "byOption": [ { "option":
"objectMessageEnabled", "value": "true", "carriers": [ "text", "bytes",
"stream", "file", "map", "value", "list", "pojo", "null" ] } ], "description":
"The reply, sent as the producer sends a message. A reply that cannot be sent,
such as a number or a List without objectMessageEnabled, fails the incoming
message (rolled back when transacted [...]
"producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"null" ], "dependsOn": [ "jmsMessageType", "allowNullBody" ], "byOption": [ {
"option": "objectMessageEnabled", "value": "true", "carriers": [ "text",
"bytes", "stream", "file", "map", "value", "list", "pojo", "null" ] } ],
"description": "Text is sent as a TextMessage; bytes, a stream or a file as a
BytesMessage; a Map of simple values as a MapMessage; null as an empty Message.
A number, a List or another Serial [...]
"producerOut": { "carriers": [ "unknown" ], "byOption": [ { "option":
"exchangePattern", "value": "InOnly", "carriers": [ "unchanged" ] }, {
"option": "disableReplyTo", "value": "true", "carriers": [ "unchanged" ] } ],
"description": "Unchanged for a message that expects no reply (InOnly). A
message that expects one (InOut) gets the body of the reply, as the consumer
gives it" }
},
diff --git a/components/camel-sjms2/src/main/docs/sjms2-component.adoc
b/components/camel-sjms2/src/main/docs/sjms2-component.adoc
index e24128751a3f..9f7370ade30e 100644
--- a/components/camel-sjms2/src/main/docs/sjms2-component.adoc
+++ b/components/camel-sjms2/src/main/docs/sjms2-component.adoc
@@ -311,6 +311,7 @@ The body is mapped to and from a JMS message as follows
(unless the option `jmsM
| Any other object | `Message` without a body (a WARN is logged) | `null`
|===
+With `jmsMessageType=Stream` the body is converted to an `InputStream`, and
its bytes are written to a `StreamMessage`.
A received `StreamMessage` is not mapped: the body is the
`jakarta.jms.StreamMessage` itself.
With `mapJmsMessage=false` the body is always the received
`jakarta.jms.Message`.
@@ -331,6 +332,10 @@ The _InOut_ Consumer uses this strategy as well ensuring
that all
response messages to the included `JMSReplyTo` destination also have the
`JMSCorrelationId` copied from the request as well.
+When the _InOut_ consumer cannot send the reply (for example, as its body
cannot be mapped to a JMS message), the
+incoming message fails as if the route had failed: it is rolled back when
`transacted=true` (or recovered with
+`CLIENT_ACKNOWLEDGE`), and the error is logged. No reply is sent, so the
caller waits until its `requestTimeout`.
+
=== Batch consuming
The consumer can be configured to receive messages in batches instead of one
at a time, by setting
diff --git
a/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java
b/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java
index d2216108a235..4f2beb461ac4 100644
---
a/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java
+++
b/components/camel-sjms2/src/main/java/org/apache/camel/component/sjms2/Sjms2Endpoint.java
@@ -58,8 +58,9 @@ import org.apache.camel.spi.UriParam;
Carrier.FILE, Carrier.MAP, Carrier.VALUE, Carrier.LIST,
Carrier.POJO, Carrier.NULL }),
description = "The reply, sent as the
producer sends a message. A reply that cannot be"
- + " sent, such as a number
or a List without objectMessageEnabled, is"
- + " dropped and the caller
times out"),
+ + " sent, such as a number
or a List without objectMessageEnabled,"
+ + " fails the incoming
message (rolled back when transacted)"
+ + " and the caller times
out"),
producerIn = @BodyType(value = {
Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
Carrier.MAP,
Carrier.NULL },
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 22f5965ef016..2e58a5d51433 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
@@ -2742,6 +2742,17 @@ said, and a request without `JMSExpiration` never
expired. To send a request wit
`disableTimeToLive=true` and no `timeToLive`. Remove `preserveMessageQos=true`
to keep using only the QoS of the
endpoint for InOnly sends.
+=== camel-sjms - an InOut reply that cannot be sent fails the message, and
jmsMessageType=Stream
+
+When an InOut consumer could not send its reply (for example, a reply body
that cannot be mapped to a JMS message),
+the error was set on the exchange but ignored: the incoming message was
committed or acknowledged, no reply was sent
+and nothing was logged. The incoming message now fails as when the route
fails: it is rolled back when
+`transacted=true` (or recovered with `CLIENT_ACKNOWLEDGE`), so the broker
delivers it again, and the error is logged.
+
+With `jmsMessageType=Stream`, sending failed with a `NullPointerException`.
The body is now written to a
+`StreamMessage`. A `jmsMessageType` that cannot be sent (`Message`) fails with
a `JMSException` that names the type.
+`camel-sjms2` behaves the same.
+
=== camel-pulsar - the producer continues the sequence ids of the broker
The `pulsar` producer no longer passes the default `initialSequenceId=-1` to
the Pulsar client, so the client
