HornetQ Topic becomes unresponsive

[parmstrong]

I have a standalone hornetq with several clients listening to a topic.  Everything will be working great for a while like a day or so and then all of a sudden the topic will become unresponsive.  I can't add more messages and the clients get time outs from hornetQ.  I have attached an example of the error the client gets when this happens as well as a thread dump for hornetQ.  Like I say everything will be working fine but seemingly randomly this will happen and I have to restart hornetQ.  Any idea what is going on here?

[clebert.suconic]

http://hornetq.sourceforge.net/docs/hornetq-2.1.2.Final/user-manual/en/html/paging.html#d0e4868

Which is being changed by:

https://issues.jboss.org/browse/HORNETQ-498

[parmstrong]

I don't think I am using message selectors on my topic.  Also This seems to happen when traffic is pretty low.  I don't imagine paging is happening.

[parmstrong]

How can I fix this issue now without waiting for the bug to be fixed?  My config for hornetQ's address settings is below.  Would you suggest that I change the address-full-policy to BLOCK and just increase the max-size-bytes.  I am sure that the topic wouldn't have had more than a few hundred messages a minute flowing to the consumers(numbered around 40).  Anyway, let me know what you think.  This seems to happen once or twice each day.

<address-setting match="#">
         <dead-letter-address>jms.queue.DLQ</dead-letter-address>
         <expiry-address>jms.queue.ExpiryQueue</expiry-address>
         <redelivery-delay>0</redelivery-delay>
         <max-size-bytes>104857600</max-size-bytes>
         <page-size-bytes>10485760</page-size-bytes>
         <message-counter-history-day-limit>10</message-counter-history-day-limit>
         <address-full-policy>PAGE</address-full-policy>
      </address-setting>

[clebert.suconic]

I'm referring to 24.9 (unacked messages).

[parmstrong]

I think that makes sense.  It does however look like we are acknowledging the messages right away.  Here is the message listener for the clients.  Am I missing something here.  The other interesting thing is that when this happens I pull up Java VisualVM and go to look at the topic and I can't.  I see that the topic is there but when I click on it to look at the attributes it won't show anything just a blank page and it won't let me invoke any jmx operations I can still see the queues like normal but not the topic.  Is there a way to test your theory that I am running into this issue?:

package com.ca.perseus.client.remoteejb;

import com.ca.perseus.client.PerseusException;
import com.ca.perseus.client.remoteejb.config.LoginSessionDetails;
import com.ca.perseus.client.remoteejb.config.NamingContextFactory;
import com.ca.perseus.client.remoteejb.config.NamingContextFactory.ContextDetails;
import com.ca.perseus.client.remoteejb.config.defaults.JnpConfigurations;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import org.apache.log4j.Logger;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnectionFactory;

@Singleton
public class JMSPerseusUpdateClient implements MessageListener
{
    private Map<Class,TopicCallBack> callBackMap;
    private TopicConnection connection;
    private TopicSession topicSession;
    private TopicSubscriber subscriber;

    @Inject
    public JMSPerseusUpdateClient(LoginSessionDetails loginSessionDetails, ExceptionListener handler)
    {
        callBackMap = new HashMap<Class, TopicCallBack>();
        try
        {

            ContextDetails contextDetails = loginSessionDetails.getMessagingServerContextDetails();

            TopicConnectionFactory factory = (TopicConnectionFactory) contextDetails.getContext().lookup(loginSessionDetails.getAppServerContextDetails().getConfig().getConnectionFactory());
            connection = factory.createTopicConnection();
            topicSession = connection.createTopicSession(false, Session.CLIENT_ACKNOWLEDGE);

            //Lookup a topic
            Topic topic = (Topic) contextDetails.getContext().lookup("topic/PerseusUpdateTopic");
            subscriber = topicSession.createSubscriber(topic);
       
            subscriber.setMessageListener(this);
            connection.start();
            connection.setExceptionListener(handler);
        }
        catch (Exception ex)
        {
            throw new PerseusException("Unable to create remote connection to model server.", ex);
        }
    }

    public void addCallBackHandler(Class clas, TopicCallBack callback)
    {
        callBackMap.put(clas, callback);
    }

    @Override
    public void onMessage(Message message)
    {
        try
        {
            message.acknowledge();//acknowleging the message as soon as I can.... (If I have issues with processing them ohh well)
            ObjectMessage objMessage = (ObjectMessage) message;
            Serializable object = objMessage.getObject();
            TopicCallBack callBackHandler = callBackMap.get(object.getClass());
            if(callBackHandler == null)
                return;
            callBackHandler.callBack(object);
        }
        catch (JMSException e)
        {
            Logger.getLogger(this.getClass()).error("Error getting and processing message", e);
        }
    }

    public void closeConnections()
    {
        try
        {
            subscriber.close();
            topicSession.close();
            connection.close();
        }
        catch (Exception e)
        {
        }
    }

    public static interface TopicCallBack<T>
    {
        public void callBack(T t);
    }
}

[clebert.suconic]

Since you're using ClientAcknowledge.. ACKs will be batched accordingly to the TransactionBatchSize.

Try using Auto-ACK for a test or Transactions on the consumer (but don't consume the whole thing in a single batch), or you may still starve (having more messages in memory than what would fit in your max-size on the address settings).

[parmstrong]

Are you saying that if we use topicSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); the acks will not be batched together? I can change that.  I am not sure I understand what you mean when you say  "or Transactions on the consumer (but don't consume the whole thing in a  single batch), or you may still starve (having more messages in memory  than what would fit in your max-size on the address settings)"  Do I have to explicitly say that the session is non-transacted?  I don't care if I have a transactional session I can lose some messages.  What can I do to elimnate the transactional behavior of the session?  I thought I hadd to create the session with Session.SESSION_TRANSACTED passed in.

[clebert.suconic]

not using a whole batch I mean:

If you consume 200 MB of messages... you will starve as the memory will  MaxSizes.

You can't just ACK messages that are requivalent > max-size, or you would starve. The same principle will be valid to batch-size or Transactions.

[parmstrong]

Okay I changed the acknowledgment mode to be pre-acknowledge to try and avoid this issue that way the server should hold onto the message at all.  The topic still froze after a day and a half.  Any other ideas?

[clebert.suconic]

Can you provide a self contained test?

Even if it's not a bug, it would help us identify what you're doing wrong.

[parmstrong]

I don't know exactly how to reproduce this seeing as I don't know what is causing it exactly and it takes a half a day or a day for it to show up.  I have however found that setting the ttl on the messages seems to help significantly.  I don't know if messages are hanging around for clients that have already disconnected or something.

[clebert.suconic]

Maybe you are creating new subscriptions, and the old subscriptions are still about.