Apache Camel JMS Component: bridgeErrorHandler does not work as documented

Question

Question

Hi, I try to bridge the error handling from the JMS consumer to the error handler of my camel route which is a DLC by using the flag bridgeErrorHandler=true but the JMSException which occurs when I consume a message is never handled by the DLC ErrorHandler but instead by the DefaultSpringErrorHandler and thus is only logged and ignored.

In my DLC I have configured that additional steps should happen when an exception occurs which are now sadly ignored. Maybe you can tell me what I did wrong?

Here are the details:

Project Setup

I am using:

• spring-boot:2.1.0.RELEASE
• apache camel:2.23.3
• camel-jms-starter:2.23.3

The route:

@Component
public class BridgeErrorHandlerTestQuellRoute extends RouteBuilder{

    private static final String ROUTE_ALIAS = "BridgeErrorHandlerTestQuellRoute";

    public static final String ENDPOINT_QUELLE = "{{BridgeErrorHandlerTestQuelle}}";

    @Override
    public void configure() {
        errorHandler(deadLetterChannel(DLCRoute.ENDPOINT_DLC));

        // @formatter:off
        from(ENDPOINT_QUELLE).routeId(ROUTE_ALIAS)
            .removeHeaders("*", "JMS*") // Entfernt alle nicht notwendigen JMS-Header
            .log(INFO, log, "Start Route: ${routeId}")
            .to("mock:mymock")
            .log(INFO, log, "Ende Route: ${routeId}")
        // @formatter:on
    }
}

the configuration:

The endpoint itself is defined in the application.properties (I had to change the topic name though):

topic.props=subscriptionShared=true&\
            transacted=true&\
            recoveryInterval=-1&\
            connectionFactory=#connectionFactory&\
            bridgeErrorHandler=true

BridgeErrorHandlerTestQuelle=jms:topic:IBMMQSERVER/AT/ALL/MYTOPIC?durableSubscriptionName=BridgeErrorHandlerTest.bridgeerrorhandlertest&${topic.props}

the connectionfactory bean

The connectionfactory is for IBM MQ:

    @Bean
    public ConnectionFactory connectionFactory() {
        try {
            MQConnectionFactory factory = new MQConnectionFactory();
            factory.setHostName(properties.getHost());
            factory.setPort(properties.getPort());
            factory.setQueueManager(properties.getQueueManager());
            factory.setChannel(properties.getChannel());
            factory.setStringProperty(WMQConstants.USERID, properties.getUser());
            factory.setStringProperty(WMQConstants.PASSWORD, properties.getPassword());
            factory.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
            return factory;
        } catch (JMSException e) {
            throw new RuntimeException(e);
        }
    }

the exception

The exception occurs at: com.ibm.msg.client.wmq.common.internal.WMQUtils#computeTextFromBytes(byte[], int, int, com.ibm.mq.jmqi.system.JmqiCodepage) and is a DetailedJMSException

WARN  EndpointMessageListener        : Execution of JMS message listener failed. Caused by: [org.apache.camel.RuntimeCamelException - Failed to extract body due to: com.ibm.msg.client.jms.DetailedJMSException: JMSCMQ1049: Mit dem Zeichensatz '1208(UTF-8) Unmappable Action: REPORT Unmappable Replacement: 63' kann die Zeichenfolge '[B@22b4f934' gar nicht oder nur teilweise konvertiert werden.
Es wurde versucht, Zeichenfolgedaten mit einem Zeichensatz zu versenden bzw. zu empfangen, der den Inhalt der Zeichenfolge nicht umsetzen kann.
Codieren Sie eine Nachricht nur mit einem Zeichensatz, von dem bekannt ist, dass er für die zu übertragenden Zeichenfolgedaten geeignet ist.. Message: 
[removed due to sensitive details]
org.apache.camel.RuntimeCamelException: Failed to extract body due to: com.ibm.msg.client.jms.DetailedJMSException: JMSCMQ1049: Mit dem Zeichensatz '1208(UTF-8) Unmappable Action: REPORT Unmappable Replacement: 63' kann die Zeichenfolge '[B@22b4f934' gar nicht oder nur teilweise konvertiert werden.
Es wurde versucht, Zeichenfolgedaten mit einem Zeichensatz zu versenden bzw. zu empfangen, der den Inhalt der Zeichenfolge nicht umsetzen kann.
Codieren Sie eine Nachricht nur mit einem Zeichensatz, von dem bekannt ist, dass er für die zu übertragenden Zeichenfolgedaten geeignet ist.. Message: 
  [removed due to sensitive details]
    at org.apache.camel.component.jms.JmsBinding.extractBodyFromJms(JmsBinding.java:176)
    at org.apache.camel.component.jms.JmsMessage.createBody(JmsMessage.java:227)
    at org.apache.camel.impl.MessageSupport.getBody(MessageSupport.java:54)
    at org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:132)
    at org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:53)
    at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:548)
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201)
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:138)
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:101)
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201)
    at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:97)
    at org.apache.camel.component.jms.EndpointMessageListener.onMessage(EndpointMessageListener.java:113)
    at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:736)
    at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:696)
    at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:674)
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:318)
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:245)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1189)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1179)
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1076)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: com.ibm.msg.client.jms.DetailedJMSException: JMSCMQ1049: Mit dem Zeichensatz '1208(UTF-8) Unmappable Action: REPORT Unmappable Replacement: 63' kann die Zeichenfolge '[B@22b4f934' gar nicht oder nur teilweise konvertiert werden.
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
    at com.ibm.msg.client.commonservices.j2se.NLSServices.createException(NLSServices.java:319)
    at com.ibm.msg.client.commonservices.nls.NLSServices.createException(NLSServices.java:226)
    at com.ibm.msg.client.wmq.common.internal.WMQUtils.computeTextFromBytes(WMQUtils.java:382)
    at com.ibm.msg.client.wmq.common.internal.WMQUtils.computeTextFromByteBuffer(WMQUtils.java:421)
    at com.ibm.msg.client.wmq.common.internal.messages.WMQTextMessage.getText(WMQTextMessage.java:240)
    at com.ibm.msg.client.jms.internal.JmsTextMessageImpl.getText(JmsTextMessageImpl.java:205)
    at com.ibm.jms.JMSTextMessage.getText(JMSTextMessage.java:124)
    at org.apache.camel.component.jms.JmsBinding.extractBodyFromJms(JmsBinding.java:161)
    ... 22 common frames omitted
Caused by: java.nio.charset.MalformedInputException: Input length = 1
    at java.nio.charset.CoderResult.throwException(CoderResult.java:281)
    at java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:816)
    at com.ibm.mq.jmqi.system.JmqiCodepage.bytesToString(JmqiCodepage.java:745)
    at com.ibm.msg.client.wmq.common.internal.WMQUtils.computeTextFromBytes(WMQUtils.java:375)
    ... 27 common frames omitted

Answer 1

Okay so we are improving this for Camel 3, where you can turn on eagerLoadingOfProperties (component level, or per endpoint) and then such JMS payload errors will be detected and the message body will be set with an exception message; which you can configure, and the caused exception on the Exchange. This allows Camel's routing error handler to detect this as an exception and route to the dead letter channel.

See JIRA ticket: https://issues.apache.org/jira/browse/CAMEL-14083

Answer 2

After

• a lot of debugging and
• trying different "constellations" of the route configuration (global onException and route specific onException-definitions)

I figured out that with the camel-jms-starter:2.23.x and camel-jms-starter:2.24.x the problem cannot be solved in a programmatic way as I want to keep the transacted=true flag. I think that it's just a bug in the camel-jms-starter that the exception which occurs is not handled by my configured Route-ErrorHandler but I can't change that.

So the current solution for me is to establish a Poison-Message mechanism in the linked MQ-Broker. This way I get out of the endless-loop and I can route the message to another queue.

Edit:
Even though I don't really like it there is a solution which can be build into the route. If I add a custom made processor to the route which just catches the exception during body conversion my configured errorHandler will be triggered and because it's the DLC Error Handler the message is successfully consumed and I don't have an endless loop. Note: Using convertBodyTo leads to the same unexpected behaviour as mentioned in the question because it uses the same API. The processor works because until then the body of the exchange was yet not touched.

Here is the code:
The changed route:

    @Override
    public void configure() {
        errorHandler(deadLetterChannel(DLCRoute.ENDPOINT_DLC));

        // @formatter:off
        from(ENDPOINT_QUELLE).routeId(ROUTE_ALIAS)
            .removeHeaders("*", "JMS*")
            .process(myCustomConverter)
            .log(INFO, log, "Start Route: ${routeId}")
            .to("mock:mymock")
            .log(INFO, log, "Ende Route: ${routeId}")
        // @formatter:on
    } 

And here is the myCustomConverter bean (with a dummy implementation):

@Component
public class MyCustomConverter implements Processor {

    private static final Logger LOG = LoggerFactory.getLogger(MyCustomConverter.class);
    @Override
    public void process(Exchange exchange) throws Exception {
        try {
            exchange.getIn().getBody(String.class);
        } catch( Exception ex) {
            LOG.error("Logging for the error");
            exchange.getIn().setBody(""); // TODO: Do something which makes sense
        }
    }
}

The reason why I don't like this workaround is that in a future version the framework might handle the same situation well and I would have to make changes in the code to remove this workaround. So for me the poisson message configuration is the way to go.

Answer 3

Moving out body conversion from jms consumer endpoint to route flow helped me as well.

Setting mapJmsMessage=false in jms consumer and adding

.setBody().method(SimpleMessageConverter.class, "fromMessage")

into route flow works fine.

I mean org.springframework.jms.support.converter.SimpleMessageConverter