invalid stream header – when publishing message manually on rabbitmq console

I have an application with spring-jms, spring-boot and rabbitmq-jms.

When I try send a message and recieve by application runs successfully. But when I publish message manually on rabbitmq console I got error bellow on application listening:

com.rabbitmq.jms.util.RMQJMSException: invalid stream header: 61736466
    at com.rabbitmq.jms.client.RMQMessage.fromMessage(RMQMessage.java:1140) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessage.convertJmsMessage(RMQMessage.java:913) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessage.convertMessage(RMQMessage.java:907) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessageConsumer.receive(RMQMessageConsumer.java:356) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessageConsumer.receive(RMQMessageConsumer.java:269) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at org.springframework.jms.support.destination.JmsDestinationAccessor.receiveFromConsumer(JmsDestinationAccessor.java:132) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveMessage(AbstractPollingMessageListenerContainer.java:418) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:303) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:257) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1237) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1227) ~[spring-jms-5.3.13.jar:5.3.13]
    at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1120) ~[spring-jms-5.3.13.jar:5.3.13]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
Caused by: java.io.StreamCorruptedException: invalid stream header: 61736466
    at java.base/java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:940) ~[na:na]
    at java.base/java.io.ObjectInputStream.<init>(ObjectInputStream.java:379) ~[na:na]
    at com.rabbitmq.jms.util.WhiteListObjectInputStream.<init>(WhiteListObjectInputStream.java:90) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    at com.rabbitmq.jms.client.RMQMessage.fromMessage(RMQMessage.java:1114) ~[rabbitmq-jms-2.3.0.jar:2.3.0]
    ... 12 common frames omitted

This is my JMS configuration

@EnableJms
@Configuration
public class ConnectionRabbitConfig {

    @Autowired RabbitProperties rabbitProperties;

    @Bean
    public ConnectionFactory jmsConnectionFactory() {
        RMQConnectionFactory connectionFactory = new RMQConnectionFactory();
        connectionFactory.setUsername(rabbitProperties.getUser());
        connectionFactory.setPassword(rabbitProperties.getPass());
        connectionFactory.setVirtualHost(rabbitProperties.getVirtualhost());
        connectionFactory.setHost(rabbitProperties.getHost());
        connectionFactory.setPort(rabbitProperties.getPort());
        return connectionFactory;
    }


    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(@Autowired ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setAutoStartup(rabbitProperties.getListenerEnabled());
        return factory;
    }

    @Bean
    public JmsTemplate defaultJmsTemplate(@Autowired ConnectionFactory connectionFactory) {
        return new JmsTemplate(connectionFactory);
    }

}

Publishing message

@Autowired
private JmsTemplate jmsTemplate;

public void publish(String queueName, String message, Map<String, Object> headers) {
    jmsTemplate.convertAndSend(queueName, message );
}

Consuming message

@JmsListener(destination = "queue-1")
public void consume(@Payload Message message,  @Headers MessageHeaders headers) {
        // runs successfully if I listen message added by same application
        // but if I try to add manualy on rabbit console I got error.
}

libs

  implementation 'org.springframework:spring-jms:5.3.13'
  implementation 'com.rabbitmq.jms:rabbitmq-jms:2.3.0'

Why do I need add on properties or headers on rabbit console for listening on application successfully? Or Do I need change something on configuration code?

Answer

Just adding my solution here.

By default spring was adding my listeners into an exchange jms.durable.queues on queues with non-ampq

Because this when publishing by application was working but when adding TextMessage on console was throwing error. All messages published on console was a non-ampq message getting conflict on spring-jms-client lib for invalid headers.

Solution:

I added a destination resolver that connects to my queues on amqp without creating default exchange jms.durable.queues More details of solution here: https://stackoverflow.com/a/70269301/7505687

Advertisement

Answer

Solution: