MDB Message consumption is failing under heavy load

[nnanda]

Hi,

I am using HornetQ 2.0 in my project for JMS. I have deployed a MDB which consumes messages from a local queue. The messages in the queue are populated by two remote applications; a web service and a web application.

Everything is working fine; but in case of load testing I noticed many messages are not consumed. The pattern is not significant. It happens when the web service and web app gets too many concurrent requests and hence, put a lot of message in the queue.

Is it because of too many messages? Is there any configuration I need to do to recover from this?

Please provide some help.

Thanks,

Niranjan

[timfox]

Firstly please see this: http://community.jboss.org/wiki/Howtoreportabugissue

Secondly, it's almost impossible to answer your question since you've given very little information.

Thirdly please try with TRUNK

[nnanda]

Hi Tim,

Thank you very much for this quick response.

By all means I respect the process of reporting a bug to JBoss. I am clueless if there is any issue with HornetQ or my configurations; so it was impossible for me to raise a JIRA. I didn't provide any details because I wanted someone to ask me what information is needed.

I beg my pardon, but I didn't understand the TRUNK point. If you mean taking latest build from trunk, since, I am using HornetQ for a critical application of my client, so I thought I should use the GA version that you guys release.

Please find below the configuration details that I am using in my application (both client and server side).

Server Side:

MDB:

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.interceptor.Interceptors;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.ejb3.annotation.ResourceAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.ejb.interceptor.SpringBeanAutowiringInterceptor;

import com.demo.db.DatabaseLogPublisherImpl;
import com.demo.dto.common.LogDTO;

@MessageDriven (
        name = "LoggerMDB",
        description = "Puts the data in the DB log table from LoggerQueue",
        activationConfig = {
                @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
                @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/LoggerQueue")
        }
)
@ResourceAdapter("hornetq-ra.rar")
@Interceptors(SpringBeanAutowiringInterceptor.class)
public class LoggerMDB implements MessageListener
{
    /**
     * Logger
     */
    private final static Log logger = LogFactory.getLog(LoggerMDB.class);
    
    @Autowired
    private DatabaseLogPublisherImpl databaseLogPublisher;
    
    /**
     * This is a call-back method and can be used to initialize all helper stuffs for this MDB
     */
    @PostConstruct
    public void initializeResources() {
        logger.debug("initializeResources : START");
        
        logger.debug("initializeResources : END");
    }
    
    /**
     * This call-back method can be used to free any resources that this MDB uses.
     */
    @PreDestroy
    public void cleanupResources() {
        logger.debug("cleanupResources : START");
        
        logger.debug("cleanupResources : END");
    }

    /* (non-Javadoc)
     * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
     */
    public void onMessage(Message message) {
        logger.debug("onMessage : START");
        
        if(message instanceof ObjectMessage) {
            try {
                final ObjectMessage receivedMessage = (ObjectMessage) message;
                final Object receivedObject = receivedMessage.getObject();
                if(receivedObject instanceof LogDTO) {
                    databaseLogPublisher.publishLog((LogDTO) receivedObject);
                }
            } catch(Exception e) {
                // For any exception, let this publishing fail.
                logger.error("Exception while publishing log data... ", e);
            }
        }       
        
        logger.debug("onMessage : END");
    }

}

HornetQ Configuration

1. hornetq-jms.xml

<queue name="LoggerQueue">
      <entry name="queue/LoggerQueue"/>
</queue>

2. hornetq-configuration.xml

<connectors>
      <connector name="netty">
         <factory-class>org.hornetq.integration.transports.netty.NettyConnectorFactory</factory-class>
         <param key="host"  value="${jboss.bind.address}"/>
         <param key="port"  value="56789"/>
      </connector>

      <connector name="in-vm">
         <factory-class>org.hornetq.core.remoting.impl.invm.InVMConnectorFactory</factory-class>
      </connector>
   </connectors>

   <acceptors>   
      <acceptor name="netty">
         <factory-class>org.hornetq.integration.transports.netty.NettyAcceptorFactory</factory-class>
         <param key="host"  value="${jboss.bind.address}"/>
         <param key="port"  value="56789"/>
      </acceptor>

      <acceptor name="in-vm">
        <factory-class>org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory</factory-class>
        <param key="server-id" value="0"/>
      </acceptor>
   </acceptors>

   <security-settings>
      <security-setting match="#">
         <permission type="createTempQueue" roles="guest"/>
         <permission type="deleteTempQueue" roles="guest"/>
         <permission type="consume" roles="guest"/>
         <permission type="send" roles="guest"/>
      </security-setting>
   </security-settings>

   <address-settings>
      <!--default for catch all-->
      <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>-1</max-size-bytes>
         <page-size-bytes>10485760</page-size-bytes>         
         <message-counter-history-day-limit>10</message-counter-history-day-limit>
      </address-setting>
   </address-settings>

Client Side

public void publishDataInQueue(final LogDTO logDTO) {
    logger.debug("publishData : START");

    Context initialContext = null;
    QueueConnectionFactory queueConFactory = null;
    QueueConnection queueConnection = null;
    QueueSession queueSession = null;
    Queue jmsQueue = null;

    // Get JNDI environment properties
    final Properties jndiProperties = new Properties();
    jndiProperties.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
    jndiProperties.put(Context.URL_PKG_PREFIXES, "java.naming.factory.url.pkgs");
    jndiProperties.put(Context.PROVIDER_URL, "jnp://my.host:port");

    try {
        // Get the JNDI context 
        initialContext = new InitialContext(jndiProperties);

        // Get Queue Connection Factory by JNDI lookup
        queueConFactory = (QueueConnectionFactory) initialContext.lookup("/ConnectionFactory");
        
        try {
            // Create a Queue Session
            queueConnection = queueConFactory.createQueueConnection();
            queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

            // Get the JMS Queue
            jmsQueue = (Queue) initialContext.lookup("queue/LoggerQueue");
            
            // Create a message sender
            final MessageProducer producer = queueSession.createSender(jmsQueue);

            // Create an ObjectMessage from LogDTO
            final StringBuilder idBuilder = new StringBuilder("ID:");
            idBuilder.append(logDTO.getCorrelationId()); // This will be used for JMS message ID and JMS correlation ID.

            final ObjectMessage objectMessage = queueSession
                    .createObjectMessage(logDTO);
            objectMessage.setJMSCorrelationID(idBuilder.toString());
            objectMessage.setJMSMessageID(idBuilder.toString());

            // Publish the message in queue
            producer.send(objectMessage);
            logger.debug("Message sent...");
        } finally {
            if(null != queueSession) {
                queueSession.close();
            }

            if(null != queueConnection) {
                queueConnection.close();
            }
        }
    } catch(Exception e) {
        logger.error("Exception in getting queue connection factory from current server. Trying with next available server...");
    }
}

Please let me know if more information is needed.

If you could find some issues in my settings/coding/configuration, please let me know.

Thanks,

Niranjan

[clebert.suconic]

I guess it's important you looking at your logs. You don't see anything suspicious?

My bet is you're missing some reconnection reconfiguration between your client and the server (since your client is another AS Container with a JCA connector) and maybe you're having some network issue?

[timfox]

A few points:

1) You need to verify any issue against TRUNK not a previous release. We don't have the resources to debug older versions

2) Looking at your program you are doing a JNDI lookup, creating session, creating producer, sending message for *every* message you send.

This is a very common and serious anti-pattern - it will be very slow! This has been discussed in many, many forum threads and is also discussed in the performance tuning section of the user manual (which I assume you haven't read).

[timfox]

The point is not that you should run TRUNK in production, the point is you should verify any errors occur with TRUNK too before reporting them. Things might have been fixed already and we only fix stuff in TRUNK.