title: Polling byte / large messages from ActiveMQ Artemis using Apache Camel ConsumerTemplate
body: <p>I'm struggling with a problem in an application based on Apache Camel when connecting to ActiveMQ Artemis via JMS. At the end of one of the Camel routes, messages are stored in an Artemis JMS queue. A legacy component running in the same application picks them up from there periodically using a <code>ConsumerTemplate</code>.</p>
  
  <p>This works fine for Camel messages with plain text bodies, but causes errors when using byte array bodies: It seems Artemis treats any message with byte body as a <a href="https://activemq.apache.org/components/artemis/documentation/latest/large-messages.html" rel="nofollow noreferrer">"large  message", which are streamed instead of kept in memory</a>. Receiving via the <code>ConsumerTemplate</code> works, but as soon as the body or headers are accessed, an exception as follows is raised:</p>
  
  <pre><code>org.apache.camel.RuntimeCamelException: Failed to extract body due to: javax.jms.IllegalStateException: AMQ119023: The large message lost connection with its session, either because of a rollback or a closed session. Message: ActiveMQMessage[ID:90c4d1d5-3233-11ea-b0cc-44032c68a56f]:PERSISTENT/ClientLargeMessageImpl[messageID=2974, durable=true, address=mytest,userID=90c4d1d5-3233-11ea-b0cc-44032c68a56f,properties=TypedProperties[firedTime=Wed Jan 08 17:26:03 CET 2020,__AMQ_CID=90b4f34e-3233-11ea-b0cc-44032c68a56f,breadcrumbId=ID-NB045-evolit-co-at-1578500762151-0-1,_AMQ_ROUTING_TYPE=1,_AMQ_LARGE_SIZE=3]]
          at org.apache.camel.component.jms.JmsBinding.extractBodyFromJms(JmsBinding.java:172) ~[camel-jms-2.22.1.jar:2.22.1]
          at org.apache.camel.component.jms.JmsMessage.createBody(JmsMessage.java:221) ~[camel-jms-2.22.1.jar:2.22.1]
          at org.apache.camel.impl.MessageSupport.getBody(MessageSupport.java:54) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.example.cdi.JmsPoller.someMethod(JmsPoller.java:36) ~[classes/:?]
          at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_171]
          at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_171]
          at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_171]
          at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_171]
          at org.apache.camel.component.bean.MethodInfo.invoke(MethodInfo.java:481) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.component.bean.MethodInfo$1.doProceed(MethodInfo.java:300) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.component.bean.MethodInfo$1.proceed(MethodInfo.java:273) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.component.bean.AbstractBeanProcessor.process(AbstractBeanProcessor.java:188) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:53) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.component.bean.BeanProducer.process(BeanProducer.java:41) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:148) ~[camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:548) [camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201) [camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:201) [camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.component.timer.TimerConsumer.sendTimerExchange(TimerConsumer.java:197) [camel-core-2.22.1.jar:2.22.1]
          at org.apache.camel.component.timer.TimerConsumer$1.run(TimerConsumer.java:79) [camel-core-2.22.1.jar:2.22.1]
          at java.util.TimerThread.mainLoop(Timer.java:555) [?:1.8.0_171]
          at java.util.TimerThread.run(Timer.java:505) [?:1.8.0_171]
  Caused by: javax.jms.IllegalStateException: AMQ119023: The large message lost connection with its session, either because of a rollback or a closed session
          at org.apache.activemq.artemis.core.client.impl.LargeMessageControllerImpl.saveBuffer(LargeMessageControllerImpl.java:273) ~[artemis-core-client-2.6.2.jar:2.6.2]
          at org.apache.activemq.artemis.core.client.impl.ClientLargeMessageImpl.saveToOutputStream(ClientLargeMessageImpl.java:115) ~[artemis-core-client-2.6.2.jar:2.6.2]
          at org.apache.activemq.artemis.jms.client.ActiveMQMessage.saveToOutputStream(ActiveMQMessage.java:853) ~[artemis-jms-client-2.6.2.jar:2.6.2]
          at org.apache.activemq.artemis.jms.client.ActiveMQMessage.setObjectProperty(ActiveMQMessage.java:693) ~[artemis-jms-client-2.6.2.jar:2.6.2]
          at org.apache.camel.component.jms.JmsBinding.createByteArrayFromBytesMessage(JmsBinding.java:251) ~[camel-jms-2.22.1.jar:2.22.1]
          at org.apache.camel.component.jms.JmsBinding.extractBodyFromJms(JmsBinding.java:163) ~[camel-jms-2.22.1.jar:2.22.1]
          ... 21 more
  Caused by: org.apache.activemq.artemis.api.core.ActiveMQIllegalStateException: AMQ119023: The large message lost connection with its session, either because of a rollback or a closed session
          at org.apache.activemq.artemis.core.client.impl.LargeMessageControllerImpl.saveBuffer(LargeMessageControllerImpl.java:273) ~[artemis-core-client-2.6.2.jar:2.6.2]
          at org.apache.activemq.artemis.core.client.impl.ClientLargeMessageImpl.saveToOutputStream(ClientLargeMessageImpl.java:115) ~[artemis-core-client-2.6.2.jar:2.6.2]
          at org.apache.activemq.artemis.jms.client.ActiveMQMessage.saveToOutputStream(ActiveMQMessage.java:853) ~[artemis-jms-client-2.6.2.jar:2.6.2]
          at org.apache.activemq.artemis.jms.client.ActiveMQMessage.setObjectProperty(ActiveMQMessage.java:693) ~[artemis-jms-client-2.6.2.jar:2.6.2]
          at org.apache.camel.component.jms.JmsBinding.createByteArrayFromBytesMessage(JmsBinding.java:251) ~[camel-jms-2.22.1.jar:2.22.1]
          at org.apache.camel.component.jms.JmsBinding.extractBodyFromJms(JmsBinding.java:163) ~[camel-jms-2.22.1.jar:2.22.1]
          ... 21 more
  </code></pre>
  
  <p>The problem also occurs for messages that do not exceed the <code>minLargeMessageSize</code> of Artemis, in a test program even for 3 bytes.</p>
  
  <p>Coincidentally, the same problem occurred in a standalone application used for testing the application. There, I was able to solve the issue by keeping the JMS session and receiver open until the JMS message body and headers were completely read. With Camel, that's abstracted away in the Spring <code>JmsTemplate</code> that Camel is based on.</p>
  
  <hr>
  
  <p>I consulted the <a href="https://camel.apache.org/components/latest/jms-component.html" rel="nofollow noreferrer">user documentation of the Camel JMS component</a> to find configuration options that might help me. I've tried the following:</p>
  
  <ul>
  <li><code>eagerLoadingOfProperties=true</code> on consumer side: no effect, only seems to affect <code>MessageListenerContainer</code>. The documentation says:</li>
  </ul>
  
  <blockquote>
    <p>It uses [...] Spring’s JmsTemplate for sending and a MessageListenerContainer for consuming.</p>
  </blockquote>
  
  <p>However, while debugging it seemed that <code>MessageListenerContainer</code> is only used when consuming messages from an JMS endpoint in a Camel route. Using a <code>ConsumerTemplate</code> like in my case uses a <code>JmsTemplate</code> for consuming.</p>
  
  <ul>
  <li><code>messageConverter</code> and <code>mapJmsMessage</code> on consumer side: no effect, they are executed when the session has already been closed</li>
  <li><code>alwaysCopyMessage</code> on producer side: I thought maybe copying prevents use of streamed large messages, no effect</li>
  <li><code>streamMessageTypeEnabled=false</code> on producer side: no effect</li>
  <li><code>jmsMessageType=Bytes</code> on both producer and consumer side: no effect</li>
  <li><code>transferExchange=true</code> on both producer and consumer side: this does seem to solve my specific case, but it feels like a workaround. Documentation advises to use the option with caution.</li>
  </ul>
  
  <p>So right now, <code>transferExchange</code> seems to be my best bet, assuming it really solves my issue in all test cases. Nevertheless, I'd be glad to get better understanding on the issue or different solutions: </p>
  
  <ol>
  <li>Why does Artemis treat small byte array messages as large messages anyway?</li>
  <li>Does Camel ConsumerTemplate support streamed large messages at all?</li>
  </ol>
  
  <p>My versions are Camel 2.22.1 and Artemis 2.10.1.</p>
  
  <hr>
  
  <p>I've been able to reproduce my problem by modifying the Camel Example <code>camel-example-cdi</code> from the release package of Camel to have the minimal classes shown below.
  In addition I've added <code>camel-jms</code> and Artemis dependencies and started Artemis locally, both like described in the <code>camel-example-artemis-large-messages</code> example.</p>
  
  <pre class="lang-java prettyprint-override"><code>public class MyRoutes extends RouteBuilder {
  
      @Override
      public void configure() {
          setupJmsComponent();
  
          from("timer:writeTimer?period=6000")
                  .log("writing to JMS")
                  .setBody(() -&gt; new byte[]{0,1,2})
                  .to(JmsPoller.ENDPOINT);
  
          from("timer:pollTimer?period=3000")
              .to("bean:jmsPoller");
      }
  
      private void setupJmsComponent() {
          ActiveMQJMSConnectionFactory connectionFactory = new ActiveMQJMSConnectionFactory("tcp://localhost:61616");
          JmsComponent jmsComponent = new JmsComponent();
          jmsComponent.setConnectionFactory(connectionFactory);
          getContext().addComponent("jms", jmsComponent);
      }
  
  }
  </code></pre>
  
  <pre class="lang-java prettyprint-override"><code>@Singleton
  @Named("jmsPoller")
  public class JmsPoller {
      static final String ENDPOINT = "jms:queue:mytest";
  
      @Inject
      private ConsumerTemplate consumerTemplate;
  
      public void someMethod(String body) {
          Exchange exchange = consumerTemplate.receive(ENDPOINT, 1000L);
          System.out.println("Received " + (exchange == null ? null : exchange.getIn().getBody()));
      }
  
  }
  </code></pre>