Trying to test the queue persistence. Can't send messages when breakpoint on consumer

java jms activemq spring-jms
Sebastien Lorber picture Sebastien Lorber · Sep 4, 2012 · Viewed 10.5k times · Source

I'm trying to test the queue persistence of ActiveMQ.

I have an embedded ActiveMQ server with an unique consumer. This embedded server receive JMS messages from many other JVM applications.

It works fine, consumer application receive the notifications.

So i've tried to test the persistence of the messages. I've put a (remote) breakpoint on the MessageListener of the consumer so that i can enqueue many messages and make the ActiveMQ server crash. On server restart, i'd like all the enqueued messages to be able to be consumed, and not to be lost.

And then i tried that test. I got into that breakpoint on the first message send. But for all messages i try to send, i get the following stacktrack on the producer side:

Exception in thread "main" org.springframework.jms.UncategorizedJmsException: Uncategorized exception occured during JMS processing; nested exception is javax.jms.JMSException: Wire format negotiation timeout: peer did not send his wire format.
    at org.springframework.jms.support.JmsUtils.convertJmsAccessException(JmsUtils.java:316)
    at org.springframework.jms.support.JmsAccessor.convertJmsAccessException(JmsAccessor.java:168)
    at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:469)
    at org.springframework.jms.core.JmsTemplate.send(JmsTemplate.java:534)
    at org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:612)
    at org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:604)
    at com.xxxxxxxxxxx.mobilepush.client.RealClientTest.main(RealClientTest.java:29)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:120)
Caused by: javax.jms.JMSException: Wire format negotiation timeout: peer did not send his wire format.
    at org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:62)
    at org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1380)
    at org.apache.activemq.ActiveMQConnection.ensureConnectionInfoSent(ActiveMQConnection.java:1466)
    at org.apache.activemq.ActiveMQConnection.createSession(ActiveMQConnection.java:308)
    at org.springframework.jms.support.JmsAccessor.createSession(JmsAccessor.java:196)
    at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:457)
    ... 9 more
Caused by: java.io.IOException: Wire format negotiation timeout: peer did not send his wire format.
    at org.apache.activemq.transport.WireFormatNegotiator.oneway(WireFormatNegotiator.java:98)
    at org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:68)
    at org.apache.activemq.transport.ResponseCorrelator.asyncRequest(ResponseCorrelator.java:81)
    at org.apache.activemq.transport.ResponseCorrelator.request(ResponseCorrelator.java:86)
    at org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1351)
    ... 13 more

I don't understand why my producer would be blocked when my consumer is in my breakpoint.

My broker uri is: mobilepush.activemq.broker.transport.connector.uri=tcp://0.0.0.0:61616

The producer connects through tcp to the broker. The consumer, colocated with the broker, connects through vm://localhost.

My configuration is pretty simple:

SERVER:

    <!--  lets create an embedded ActiveMQ Broker -->
    <amq:broker useJmx="false" persistent="true">
        <amq:transportConnectors>
            <amq:transportConnector uri="${mobilepush.activemq.broker.transport.connector.uri}" />
        </amq:transportConnectors>
        <amq:persistenceAdapter>
            <amq:kahaPersistenceAdapter directory="${mobilepush.activemq.broker.queue.persistence.directory}" maxDataFileLength="100 Mb"/>
        </amq:persistenceAdapter>
    </amq:broker>

CONSUMER:
(management namespace and xebia class it only a JMX decorator)

<bean id="connectionFactory" class="fr.xebia.management.jms.SpringManagedConnectionFactory">
        <property name="connectionFactory">
            <bean class="org.apache.activemq.ActiveMQConnectionFactory" >
                <property name="brokerURL" value="${mobilepush.activemq.broker.uri}"/>
            </bean>
        </property>
    </bean>

    <bean id="pushConsumer" class="com.xxxxxxxxxxxxxxx.mobilepush.messaging.jms.PushConsumer">
        <property name="jmsPushMessageConverter" ref="jmsPushMessageConverter"/>
        <property name="pushDelegate" ref="directPushDelegate"/>
    </bean>

    <management:executor-service id="pushConsumerExecutor"
                                 pool-size="${mobilepush.consumer.thread.min}-${mobilepush.consumer.thread.max}" keep-alive="60" />

    <jms:listener-container
            task-executor="pushConsumerExecutor"
            connection-factory="connectionFactory"
            acknowledge="auto"
            container-class="fr.xebia.springframework.jms.ManagedDefaultMessageListenerContainer">
        <jms:listener destination="mobilepush.queue" ref="pushConsumer" method="onMessage" />
    </jms:listener-container>

PRODUCER:

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"  >
        <property name="brokerURL" value="${mobilepush.activemq.broker.uri}"/>
    </bean>


    <bean id="mobilePushJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="defaultDestination" ref="mobilePushQueue"/>
        <property name="messageConverter" ref="jmsPushMessageConverter"/>
        <property name="connectionFactory">
            <!-- lets wrap in a pool to avoid creating a connection per send -->
            <bean class="org.springframework.jms.connection.SingleConnectionFactory">
                <property name="targetConnectionFactory">
                    <ref local="connectionFactory" />
                </property>
            </bean>
        </property>
    </bean>

Answer

Sebastien Lorber picture Sebastien Lorber · Sep 5, 2012

I found the problem!

The remote breakpoint i put on my embedded ActiveMQ consumer was a default breakpoint with suspend-policty=all.

And as the consumer and the server runs on the same JVM, i was also suspending all the ActiveMQ server threads!

The solution is to use a breakpoint suspend-policy=thread so that only the consumer thread is suspended and the server threads can continue to run.

Related questions

High Performance DefaultMessageListenerContainer config for thousands of topics

for my current project i need to consume messages from many destinations (from hundreds up to 20 or 30k) all destinations are topics. currently (for initial load tests) all messages are created localy on the same server, in a thread pool. …

Read More
Spring JMS(ActiveMQ) delayed delivery of messages

We're trying to set a delay on some JMS messages, so that a message will only be added to the queue/ received by the listener after x time. So far we've tried 2 approaches that didn't work. 1) According to the spring …

Read More
jms producer performance with spring

i created a simple producer consumer simulation based on spring, jms and activemq, i'm trying to reach high performance from both sides, producers and consumers, Connection settings : <tx:annotation-driven /> <bean id="transactionManager" class="org.springframework.jms.connection.…

Read More