Recovery of JMS message listener fails with "invalid stream header"

[Faerylnahr]

Hi everybody,

I could need somebody who can point me into the right direction with this issue.

I get the following two lines of messages
WARN o.s.j.l.DefaultMessageListenerContainer - Setup of JMS message listener invoker failed for destination 'my-queue' - trying to recover. Cause: invalid stream header: 7B226D65
INFO o.s.j.l.DefaultMessageListenerContainer - Successfully refreshed JMS Connection

I've installed RabbitMQ 3.13.2 on Erlang 26.2.5 on Windows about a month ago. I have several producers sending MQTT messages to RabbitMQ and a binding that brings the incoming messages to 'my-queue'. The consumer is a Java service which consumes the messages via JMS and saves the content to the database. Previously I used ActiveMQ - that's why JMS is used here.

When I deploy the application everything runs fine. All messages are consumed. I cannot say when this situation started to appear (maybe a week ago) but at some point this setup suddenly showed a weird behaviour and I cannot directly reproduce it - from time to time the setup stops working with above messages. Logging shows me that some seconds befor this happens the DB actions suddenly last >2000ms (in normal cases it's between 10 - 40ms). After that the client isn't aible to recovery from this and the service get's stuck in the above mentioned loop (trying to recover... successfully refreshed... trying to recover...).

I already found this discussion: https://groups.google.com/g/rabbitmq-users/c/Zbcp9WL-vUM
My destination is configured like it's mentioned in the discussion:

 <bean id="my-queue" class="com.rabbitmq.jms.admin.RMQDestination">
    <property name="destinationName" value="#{myproperties.jmsQueueName}"/>
    <property name="amqpRoutingKey" value="" />
    <property name="amqpExchangeName" value="" />
    <property name="amqpQueueName" value="#{myproperties.jmsQueueName}"/>
    <property name="amqp" value="true" />
  </bean>

What else can I do to get it working? I'm really stuck and would appreciate any help 🙏

[acogoluegnes]

It's hard to tell what's wrong with this amount of information. Please provide a standalone project that reproduces the problem, maybe by faking a long processing time (more than 2 seconds, as you mentioned).

Note the default timeout for processing is 2 seconds, you can increase it with RMQConnectionFactory#onMessageTimeoutMs.

[Faerylnahr]

Hi Arnaud, that's what I finally did. Ich changed the onMessageTimoutMs to 20s. I could reproduce the problem by integrating a Timeout.sleep(random value between 0 and 10s). After 2 or 3 messages the consumer channel got closed.
Here you've posted some easy code that's basically what I did. OnMessage contains the random timeout and connectionFactory has the onMessageTimeoutMs property. That's it.
I'm sending simple messages every second and due to the faked timeout the channel gets closed and the above messages start to appear.

Maybe I'm to much of a newbe to RabbitMQ but I don't understand why a channel is closed when there is just a timeout for a single message.

[acogoluegnes]

The code you mentioned above does not contain the publishing part. That's why I'm asking for a standalone project that reproduces the issue.

[Faerylnahr]

Hi Arnaud, my problem is the following: in my current environment I have a producer that sends MQTT messages which are received in the amq.topic exchange. A binding leads these messages to my-queue (type: classic). I do not have a chance to change the code on producer side.

I even changed the connectionFactory and added "keepTextMessageType=true"

<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    <property name="targetConnectionFactory">
      <bean class="com.rabbitmq.jms.admin.RMQConnectionFactory">
        <property name="username" value="user"/>
        <property name="password" value="secret"/>
        <property name="virtualHost" value="/"/>
        <property name="host" value="some-host"/>
        <property name="port" value="5672"/>
        <property name="onMessageTimeoutMs" value="20000" />
        <property name="keepTextMessageType" value="true" />
      </bean>
    </property>
    <property name="sessionCacheSize" value="10"/>
  </bean>

But the situation remains the same. As soon as the listener onMessage method has to handle a case where processing time of onMessage > onMessageTimeoutMs it "crashes" with the above message ("invalid stream header") - redeploying the application resolves the problem temporarily until the next timout arises.

[acogoluegnes]

Last time before I close this discussion: if you want us to investigate this issue (for you), please provide a standalone, minimal-dependency (just RabbitMQ JMS client, no Spring) project with a step-by-step procedure to reproduce the issue.

[Faerylnahr]

Hi Arnaud,

I'm sorry, I don't want to annoy you, but I'm not sure if I get right what you want me to provide. I've tried to create a simple Java project, containing the RabbitMQ JMS client and a Consumer that helps to reproduce the issue. Is that what you meant?
rabbitmq-jms-client-demo.zip

As for the Producer: I am not sure in what way it might help to create a dummy producer in the standalone project as my actual producer in my productive environment uses a completly different way to produce messages. (As far as I know) I have no influence on this behaviour or any properties which might need to be set when the message is sent.

As for the Consumer: when I try to pause the message processing by using "Tread.sleep" I only get an InterruptException saying "sleep interrupted" - that's not what I see on my productive system :( I'am currently a little bit lost on how I can provide you more information on this issue :(

[acogoluegnes]

It does not matter if the publisher is not the same as the one in production, as long as it reproduces the same issue. The project you provided is a start, but should I update it and run it randomly until I get the same error as you?

You have to tell how to reproduce the issue. If I knew how to reproduce this issue, there would be a test case in the project test suite and a patch to fix it.

[Faerylnahr]

Ok, I'll try my best. I somehow managed to reproduce it. Although it's not exact the same message like in my productive environment.

1. I start the consumer --> is now listening to my-queue
2. I start the producer --> starts publishing messages 1/s
3. I look at the RabbitMQ web console and I see for each (consumer, producer) 1 connections and 1 channel
4. Console shows me "sleep interrupted" and a corresponding stacktrace, every time the onMessage processing lasts longer than the timeout is set with the connectionFactory
5. Looking at the stack trace I see, that the interrupt is caused by the DeliveryExecutor --> here I can find the message I see in my productive log files ("onMessage took too long and was interrupted")
6. Looking at the RabbitMQ web console I see the channel for the consumer disappear and it won't reappear
   Unfortunately I can't get it to the point where I get the message "Setup of JMS message listener invoker failed for destination 'my-queue' - trying to recover. Cause: invalid stream header: 7B226D65".

But: I changed my productive code that way that I added an exception listener (like I did in the sample code)
--> in the sample code the exception listener is not triggert (idkw)
--> in my productive code w/o exception listener I get the above message, channel is closed and not reopened
--> in my productive code w/ exception listener, I get the above message, channel is closed and immediately reopened --> now my RabbitMQ log is growing because of channels closed and created again ^^

Hope that helps :(

rabbitmq-jms-client-demo-v2.zip