try to make the case more clear.

we are running our application on 4 jboss svr, one as producer and other 3 as consumer, and try to replicate the message from producer to consumer,

here is the code we are using for create session and messagehandler

public class ReplConnectorHornetQImpl implements ReplConnector {

    private static final Logger LOG = Logger.getLogger(ReplConnectorHornetQImpl.class);

    private static final MessageHandler msgHandler = new MessageHandlerImpl();

    private static final int RECONNECT_ATTEMPTS = 600;

    private static final long RETRY_INTERVAL = 100l;

    private static final int BUFFER_SIZE = 10240;

    private ReplServiceConfig replConfig;

    private ClientSession session;

    private ClientProducer producer;

    private ClientConsumer consumer;

    private static class LazyReplConnectorHolder

    {

        private static final ReplConnectorHornetQImpl instance = new ReplConnectorHornetQImpl();

    }

    private ReplConnectorHornetQImpl() {}

    public static ReplConnectorHornetQImpl getInstance()

    {

        return LazyReplConnectorHolder.instance;

    }

    @Override

    public void init() throws Exception

    {

        final SimpleString queueName = new SimpleString("queuename" + "-" + System.getProperty("java.rmi.server.hostname") + "-" + System.getProperty("jboss.server.name"));

        LOG.info("init: queueName = " + queueName.toString());

        Map connectionParams = new HashMap();

        connectionParams.put("host", IP_ADDRESS);

        connectionParams.put("port", PORT));

        TransportConfiguration config = new TransportConfiguration("org.hornetq.core.remoting.impl.netty.NettyConnectorFactory", connectionParams);

        ServerLocator locator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration[] {

                config

        });

        locator.setReconnectAttempts(RECONNECT_ATTEMPTS);

        locator.setRetryInterval(RETRY_INTERVAL);

        String role = (config.isWriter()) ? "writer" : "reader";

        LOG.debug("Connecting to " + locator.toString() + " as " + role);

        try {

        ClientSessionFactory factory = locator.createSessionFactory();

        session = factory.createSession(true, true,0);

        if(!session.queueQuery(queueName).isExists())

        {

            session.createQueue("topicname", "queuename", false);

            LOG.info((new StringBuilder("Initialized queue because queue '")).append(queueName).append("' has not been setup in the server").toString());

        }     

        if(config.isWriter()) {

            producer = session.createProducer(replConfig.getTopicName());

        } else {

            consumer = session.createConsumer(queueName);

            consumer.setMessageHandler(msgHandler);

        }

        session.start();

        } catch (Exception e) {

            HornetQFailureBeanImpl.getInstance().setStateToFalse();

            throw new Exception(e);

        }

    }

    public void cleanup()

    {

        try

        {

            session.close();

            producer.close();

            consumer.close();

        }

        catch(Exception e)

        {

            LOG.error((new StringBuilder("cleanup: ")).append(e).toString());

        }

    }

    public boolean send(Object toSend)

    {

        if(!replConfig.isEnable())

            return true;

        // If this is not a writer config, this should not have been called

        if(!replConfig.isWriter())

        {

            LOG.error("send: called for a non-writer configuration");

            return false;

        }

        // Create json string

        ObjectMapper mapper = new ObjectMapper();

        String jsonStr;

        try {

            jsonStr = mapper.writeValueAsString(toSend);

        } catch (Exception e) {

            LOG.error("send: caught exception " + e);

            return false;

        }

        LOG.debug("send: called to send " + jsonStr);

        ClientMessage message = session.createMessage(false);

        message.getBodyBuffer().writeString(jsonStr);

        try

        {

            producer.send(message);

            LOG.debug((new StringBuilder("send: producer sent message ")).append(message.toString()).toString());

        }

        catch(HornetQException e)

        {

            LOG.error((new StringBuilder("send ")).append(e).toString());

            return false;

        }

        if (producer.isClosed()) {

            // check if the session is closed after  reconnect

            HornetQFailureBeanImpl.getInstance().setStateToFalse();

            HornetQFailureBeanImpl.getInstance().incrementCount();

            LOG.error((new StringBuilder("send fail: ")).append(message).toString());

            return false;

        }

        return true;

    }

}

public class MessageHandlerImpl implements MessageHandler {

    private static final Logger LOG = Logger.getLogger(MessageHandlerImpl.class);

    @Override

    public void onMessage(ClientMessage message) {

        String jsonStr = message.getBodyBuffer().readString();

        LOG.debug("received message: " + jsonStr);

        // Map to ReplWrapper object

        ObjectMapper mapper = new ObjectMapper();

        JsonNode jsonObj = null;

        try {            

            jsonObj = mapper.readValue(jsonStr, JsonNode.class);

            @SuppressWarnings("rawtypes")

            Class c = Class.forName(jsonObj.get("method").getTextValue()); // Reserve line repl callback

            LOG.debug("onMessage: callback method on " + c.getName());

            ReplCallback callback = (ReplCallback)(c.newInstance());

            callback.onMessage(jsonObj);

            message.acknowledge();

        } catch (JsonParseException e) {

            LOG.error("onMessage: could not parse " + jsonStr + e);

            return;

        } catch (JsonMappingException e) {

            LOG.error("onMessage: could not map " + jsonStr + e);

            return;

        } catch (IOException e) {

            LOG.error("onMessage: error receiving " + jsonStr + e);

            return;

        } catch (ClassNotFoundException e) {

            LOG.error("onMessage: error receiving " + jsonStr + e);

            return;

        } catch (Exception e) {

            LOG.error("onMessage: error processing " + jsonStr + e);

            return;

        }

    }

}

the config for hornetq is listed as following

<configuration xmlns="urn:hornetq"

               xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

               xsi:schemaLocation="urn:hornetq /schema/hornetq-configuration.xsd">

   <paging-directory>${data.dir:../data}/paging</paging-directory>

   <bindings-directory>${data.dir:../data}/bindings</bindings-directory>

   <journal-directory>${data.dir:../data}/journal</journal-directory>

   <journal-min-files>10</journal-min-files>

   <large-messages-directory>${data.dir:../data}/large-messages</large-messages-directory>

   <connectors>

      <connector name="netty">

         <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class>

         <param key="host"  value="${hornetq.remoting.netty.host:localhost}"/>

         <param key="port"  value="${hornetq.remoting.netty.port:5445}"/>

      </connector>

      <connector name="netty-throughput">

         <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class>

         <param key="host"  value="${hornetq.remoting.netty.host:localhost}"/>

         <param key="port"  value="${hornetq.remoting.netty.batch.port:5455}"/>

         <param key="batch-delay" value="50"/>

      </connector>

   </connectors>

   <acceptors>

      <acceptor name="netty">

         <factory-class>org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory</factory-class>

         <param key="host"  value="${hornetq.remoting.netty.host:localhost}"/>

         <param key="port"  value="${hornetq.remoting.netty.port:5445}"/>

      </acceptor>

      <acceptor name="netty-throughput">

         <factory-class>org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory</factory-class>

         <param key="host"  value="${hornetq.remoting.netty.host:localhost}"/>

         <param key="port"  value="${hornetq.remoting.netty.batch.port:5455}"/>

         <param key="batch-delay" value="50"/>

         <param key="direct-deliver" value="false"/>

      </acceptor>

   </acceptors>

   <security-enabled>false</security-enabled>

   <security-settings>

      <security-setting match="#">

         <permission type="createNonDurableQueue" roles="guest"/>

         <permission type="deleteNonDurableQueue" 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>10485760</max-size-bytes>

         <message-counter-history-day-limit>10</message-counter-history-day-limit>

         <address-full-policy>BLOCK</address-full-policy>

      </address-setting>

   </address-settings>

</configuration>

sorry, pretty new to hornetq, not sure what do you mean by "consume".

the message is send to consumer and consumed by the MessageHandlerImpl which I listed above, after that we did a   message.acknowledge();

all the consumer queue message count is 0 after 10k message delivered and the "messaged added" is equal to producer producing message, so

I thought all the messages are consumed already.

is not this the full cycle. or there is any step to "consume" the message?