This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26152-sjms-reply-errors in repository https://gitbox.apache.org/repos/asf/camel.git
commit e34ea0e7b213f1adb684db626b3ff90c27d71b4f Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 17:46:00 2026 +0100 CAMEL-26152: camel-sjms - fail the message when an InOut reply cannot 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 1bc188419d61..e15a3ccfec65 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 @@ -2429,6 +2429,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
