<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.0 Transitional//EN">
<html>
<body link="#355491" alink="#4262a1" vlink="#355491" style="background: #e2e2e2; margin: 0; padding: 20px;">
<div>
<table cellpadding="0" bgcolor="#FFFFFF" border="0" cellspacing="0" style="border: 1px solid #dadada; margin-bottom: 30px; width: 100%; -moz-border-radius: 6px; -webkit-border-radius: 6px;">
<tbody>
<tr>
<td>
<table border="0" cellpadding="0" cellspacing="0" bgcolor="#FFFFFF" style="border: solid 2px #ccc; background: #dadada; width: 100%; -moz-border-radius: 6px; -webkit-border-radius: 6px;">
<tbody>
<tr>
<td bgcolor="#000000" valign="middle" height="58px" style="border-bottom: 1px solid #ccc; padding: 20px; -moz-border-radius-topleft: 3px; -moz-border-radius-topright: 3px; -webkit-border-top-right-radius: 5px; -webkit-border-top-left-radius: 5px;">
<h1 style="color: #333333; font: bold 22px Arial, Helvetica, sans-serif; margin: 0; display: block !important;">
<!-- To have a header image/logo replace the name below with your img tag -->
<!-- Email clients will render the images when the message is read so any image -->
<!-- must be made available on a public server, so that all recipients can load the image. -->
<a href="http://community.jboss.org/index.jspa" style="text-decoration: none; color: #E1E1E1">JBoss Community</a></h1>
</td>
</tr>
<tr>
<td bgcolor="#FFFFFF" style="font: normal 12px Arial, Helvetica, sans-serif; color:#333333; padding: 20px; -moz-border-radius-bottomleft: 4px; -moz-border-radius-bottomright: 4px; -webkit-border-bottom-right-radius: 5px; -webkit-border-bottom-left-radius: 5px;"><h3 style="margin: 10px 0 5px; font-size: 17px; font-weight: normal;">
Only one subscriber sees a topic message in a clustered topic setup
</h3>
<span style="margin-bottom: 10px;">
created by <a href="http://community.jboss.org/people/poesys">Bob Muller</a> in <i>JBoss Messaging</i> - <a href="http://community.jboss.org/message/595433#595433">View the full discussion</a>
</span>
<hr style="margin: 20px 0; border: none; background-color: #dadada; height: 1px;">
<div class="jive-rendered-content"><p>I have created a straightforward Topic setup on a two-node JBoss 5.1 cluster (using the all server with Messaging defaults). Here's the topic configuration I added in destinations-service.xml on all nodes:</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><pre class="jive-pre"><code class="jive-code jive-xml">   <span class="jive-xml-tag"><mbean
      code="org.jboss.jms.server.destination.TopicService"
      name="jboss.messaging.destination:service=Topic,name=PoesysCacheDelete"
      xmbean-dd="xmdesc/Topic-xmbean.xml"></span>
      <span class="jive-xml-tag"><depends optional-attribute-name="ServerPeer"></span>jboss.messaging:service=ServerPeer<span class="jive-xml-tag"></depends></span>
      <span class="jive-xml-tag"><depends></span>jboss.messaging:service=PostOffice<span class="jive-xml-tag"></depends></span>
   <span class="jive-xml-tag"></mbean></span>
</code></pre><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p>Our web app has a simple pojo cache (a Java Map) that holds data objects queried from a database. Each node has a separate singleton cache. I am using messaging to have the cache on each node remove an object on demand. So the behavior I want is:</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><ol><li>The user updates an object.</li><li>The database transaction happens.</li><li>The data access code sends a message to the topic to remove the object from the cache (the payload is the key in the Map).</li><li>The registered listeners (MessageListener implementations) in the instance of the web app running on each node receive the published topic message and respond by removing the cached object as requested.</li></ol><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p>This works fine as long as I have only one node; I'm logging all this and I see the message sent and received and the cache deleted.</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p>However, when I have two nodes, everything stops working. Looking at the logs, it appears that when a node receives the message and acts on it, the other node(s) don't ever see that message. Since I want the action to happen on all nodes, this means the cache operation(s) don't happen on all the nodes. In particular, the sticky-session setup means that the originating/publishing node only sees some of the messages and may or may not remove the object from its own cache, which then causes the next page to display old data.</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p>This looks an awful lot like a race condition between the nodes. My understanding was (and seems to be verified by looking through the topics on this discussion forum) that a clustered topic means that ALL listeners get the message. I can verify that isn't true here--the instant a message is received on one node, the other node(s) don't see it.</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p>I did verify that the mbean is clustered by examining it in the jmx-console for the nodes.</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p>This is the Java code that implements MessageListener:</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><pre class="jive-pre"><code class="jive-code jive-java"><font color="darkgreen">/**
* A thread-based class that listens for messages about the Poesys/DB cache.
*
* @author Robert J. Muller
*/</font>
<font color="navy"><b>public</b></font> <font color="navy"><b>class</b></font> CacheMessageListener <font color="navy"><b>implements</b></font> Runnable, MessageListener <font color="navy">{</font>
 
  <font color="darkgreen">/**
   * Logger for this class
   */</font>
  <font color="navy"><b>private</b></font> <font color="navy"><b>static</b></font> <font color="navy"><b>final</b></font> Logger logger =
    Logger.getLogger(CacheMessageListener.class);
 
  <font color="navy"><b>private</b></font> <font color="navy"><b>static</b></font> <font color="navy"><b>final</b></font> String LISTENER_MSG =
    <font color="red">"com.poesys.db.dao.msg.listener_problem"</font>;
  <font color="navy"><b>private</b></font> <font color="navy"><b>static</b></font> <font color="navy"><b>final</b></font> String DELETE_MSG =
    <font color="red">"com.poesys.db.dao.msg.delete_problem"</font>;
  <font color="navy"><b>private</b></font> <font color="navy"><b>static</b></font> <font color="navy"><b>final</b></font> String INTERRUPT_MSG =
    <font color="red">"com.poesys.db.dao.msg.interrupted"</font>;
 
  <font color="darkgreen">/** JMS topic name for the Poesys/DB delete topic */</font>
  <font color="navy"><b>public</b></font> <font color="navy"><b>static</b></font> <font color="navy"><b>final</b></font> String DELETE_TOPIC = <font color="red">"topic/PoesysCacheDelete"</font>;
  <font color="darkgreen">/** JMS connection factory name */</font>
  <font color="navy"><b>public</b></font> <font color="navy"><b>static</b></font> <font color="navy"><b>final</b></font> String CONNECTION_FACTORY = <font color="red">"ClusteredConnectionFactory"</font>;
  <font color="darkgreen">/** JMS ObjectMessage property name for cache name property */</font>
  <font color="navy"><b>public</b></font> <font color="navy"><b>static</b></font> <font color="navy"><b>final</b></font> String CACHE_NAME_PROPERTY = <font color="red">"CacheName"</font>;
 
  <font color="navy"><b>private</b></font> Connection connection;
  <font color="navy"><b>private</b></font> Session sessionConsumer;
  <font color="navy"><b>private</b></font> MessageConsumer consumer;
 
  <font color="darkgreen">/**
   * Runs the message listener.
   */</font>
  <font color="navy"><b>public</b></font> <font color="navy"><b>void</b></font> run() <font color="navy">{</font>
    <font color="navy"><b>try</b></font> <font color="navy">{</font>
      <font color="darkgreen">// Look up the connection factory using JNDI.</font>
      Context initial = <font color="navy"><b>new</b></font> InitialContext();
      ConnectionFactory cf =
        (ConnectionFactory)initial.lookup(CONNECTION_FACTORY);
 
      <font color="darkgreen">// Set this object to be a message listener for delete requests.</font>
      Destination deleteTopic = (Destination)initial.lookup(DELETE_TOPIC);
      connection = cf.createConnection();
      sessionConsumer =
        connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      consumer = sessionConsumer.createConsumer(deleteTopic);
      consumer.setMessageListener(<font color="navy"><b>this</b></font>);
      connection.start();
 
      logger.info(<font color="red">"Cache message listener started, listening for cache removal requests"</font>);
 
      <font color="darkgreen">// Sleep indefinitely until interruption.</font>
      <font color="navy"><b>while</b></font> (!Thread.currentThread().isInterrupted()) <font color="navy">{</font>
        <font color="darkgreen">// Sleeps for 10 seconds</font>
        Thread.sleep(10 * 1000);
      <font color="navy">}</font>
    <font color="navy">}</font> <font color="navy"><b>catch</b></font> (InterruptedException e) <font color="navy">{</font>
      String message = com.poesys.db.Message.getMessage(INTERRUPT_MSG, <font color="navy"><b>null</b></font>);
      logger.info(message);
    <font color="navy">}</font> <font color="navy"><b>catch</b></font> (Exception e) <font color="navy">{</font>
      String message = com.poesys.db.Message.getMessage(LISTENER_MSG, <font color="navy"><b>null</b></font>);
      logger.error(message, e);
    <font color="navy">}</font> <font color="navy"><b>finally</b></font> <font color="navy">{</font>
      <font color="navy"><b>if</b></font> (connection != <font color="navy"><b>null</b></font>) <font color="navy">{</font>
        <font color="navy"><b>try</b></font> <font color="navy">{</font>
          connection.close();
        <font color="navy">}</font> <font color="navy"><b>catch</b></font> (JMSException e) <font color="navy">{</font>
          String message = com.poesys.db.Message.getMessage(LISTENER_MSG, <font color="navy"><b>null</b></font>);
          logger.error(message, e);
        <font color="navy">}</font>
      <font color="navy">}</font>
    <font color="navy">}</font>
  <font color="navy">}</font>
 
  <font color="darkgreen">/*
   * (non-Javadoc)
   *
   * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
   */</font>
  @Override
  <font color="navy"><b>public</b></font> <font color="navy"><b>void</b></font> onMessage(Message message) <font color="navy">{</font>
    IPrimaryKey key = <font color="navy"><b>null</b></font>;
    String cacheName = <font color="navy"><b>null</b></font>;
 
    <font color="navy"><b>if</b></font> (message == <font color="navy"><b>null</b></font>) <font color="navy">{</font>
      logger.error(<font color="red">"Cache message listener received null message"</font>);
    <font color="navy">}</font> <font color="navy"><b>else</b></font> <font color="navy">{</font>
      <font color="navy"><b>try</b></font> <font color="navy">{</font>
        logger.debug(<font color="red">"Received cache removal request"</font>);
        <font color="darkgreen">// Get the message and extract the key and the cache name.</font>
        ObjectMessage objectMessage = (ObjectMessage)message;
        <font color="navy"><b>if</b></font> (objectMessage != <font color="navy"><b>null</b></font>) <font color="navy">{</font>
          <font color="darkgreen">// Message key is the object payload.</font>
          Serializable object = objectMessage.getObject();
          <font color="navy"><b>if</b></font> (object <font color="navy"><b>instanceof</b></font> com.poesys.ms.pk.IPrimaryKey) <font color="navy">{</font>
            com.poesys.ms.pk.IPrimaryKey messageKey =
              (com.poesys.ms.pk.IPrimaryKey)objectMessage.getObject();
            <font color="darkgreen">// Translate into database primary key.</font>
            key = MessageKeyFactory.getKey(messageKey);
            <font color="darkgreen">// Cache name is a property.</font>
            cacheName = objectMessage.getStringProperty(CACHE_NAME_PROPERTY);
            IDtoCache<?> cache = DaoManager.getCache(cacheName);
            <font color="darkgreen">// Remove the object from the local cache only if it's there; if</font>
            <font color="darkgreen">// it's not there, move on since there's nothing to do.</font>
            <font color="navy"><b>if</b></font> (cache != <font color="navy"><b>null</b></font>) <font color="navy">{</font>
              logger.debug(<font color="red">"Removing key "</font> + key.getValueList() + <font color="red">" from cache "</font>
                           + cacheName);
              cache.removeLocally(key);
            <font color="navy">}</font> <font color="navy"><b>else</b></font> <font color="navy">{</font>
              logger.debug(<font color="red">"No cache from which to remove object"</font>);
            <font color="navy">}</font>
          <font color="navy">}</font> <font color="navy"><b>else</b></font> <font color="navy">{</font>
            logger.error(<font color="red">"Cache message listener received message with a payload that was not a primary key"</font>);
          <font color="navy">}</font>
        <font color="navy">}</font>
      <font color="navy">}</font> <font color="navy"><b>catch</b></font> (JMSException e) <font color="navy">{</font>
        <font color="darkgreen">// log full information and ignore</font>
        Object[] objects = <font color="navy">{</font> cacheName, key.getValueList() <font color="navy">}</font>;
        String errorMsg = com.poesys.db.Message.getMessage(DELETE_MSG, objects);
        logger.error(errorMsg, e);
      <font color="navy">}</font> <font color="navy"><b>catch</b></font> (RuntimeException e) <font color="navy">{</font>
        <font color="darkgreen">// log and ignore</font>
        logger.error(<font color="red">"Runtime exception in onMessage: "</font>, e);
      <font color="navy">}</font>
    <font color="navy">}</font>
  <font color="navy">}</font>
<font color="navy">}</font>
</code></pre><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p>This is the Java code that sends the message:</p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><p style="min-height: 8pt; height: 8pt; padding: 0px;"> </p><pre class="jive-pre"><code class="jive-code jive-java">  @Override
  <font color="navy"><b>public</b></font> <font color="navy"><b>void</b></font> remove(IPrimaryKey key) <font color="navy">{</font>
    <font color="darkgreen">// Send a message to listeners asking to remove there. This will remove</font>
    <font color="darkgreen">// the object from all listening caches with the cache name of this cache,</font>
    <font color="darkgreen">// including THIS one.</font>
    Connection connection = <font color="navy"><b>null</b></font>;
    <font color="navy"><b>try</b></font> <font color="navy">{</font>
      Context initial = <font color="navy"><b>new</b></font> InitialContext();
      ConnectionFactory cf =
        (ConnectionFactory)initial.lookup(CacheMessageListener.CONNECTION_FACTORY);
      Destination deleteTopic =
        (Destination)initial.lookup(CacheMessageListener.DELETE_TOPIC);
      connection = cf.createConnection();
      Session session =
        connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageProducer producer = session.createProducer(deleteTopic);
      connection.start();
      ObjectMessage om = session.createObjectMessage(key.getMessageObject());
      om.setStringProperty(CacheMessageListener.CACHE_NAME_PROPERTY,
                           getCacheName());
      producer.send(om);
      logger.debug(<font color="red">"Sent message to remove "</font> + key.getValueList()
                   + <font color="red">" from cache "</font> + getCacheName());
    <font color="navy">}</font> <font color="navy"><b>catch</b></font> (Exception e) <font color="navy">{</font>
      Object[] objects = <font color="navy">{</font> getCacheName() <font color="navy">}</font>;
      String message = com.poesys.db.Message.getMessage(PRODUCER_MSG, objects);
      logger.error(message, e);
    <font color="navy">}</font> <font color="navy"><b>finally</b></font> <font color="navy">{</font>
      <font color="navy"><b>if</b></font> (connection != <font color="navy"><b>null</b></font>) <font color="navy">{</font>
        <font color="navy"><b>try</b></font> <font color="navy">{</font>
          connection.close();
        <font color="navy">}</font> <font color="navy"><b>catch</b></font> (JMSException e) <font color="navy">{</font>
          Object[] objects = <font color="navy">{</font> getCacheName() <font color="navy">}</font>;
          String message =
            com.poesys.db.Message.getMessage(PRODUCER_MSG, objects);
          logger.error(message, e);
        <font color="navy">}</font>
      <font color="navy">}</font>
    <font color="navy">}</font>
  <font color="navy">}</font>
</code></pre></div>
<div style="background-color: #f4f4f4; padding: 10px; margin-top: 20px;">
<p style="margin: 0;">Reply to this message by <a href="http://community.jboss.org/message/595433#595433">going to Community</a></p>
<p style="margin: 0;">Start a new discussion in JBoss Messaging at <a href="http://community.jboss.org/choose-container!input.jspa?contentType=1&containerType=14&container=2042">Community</a></p>
</div></td>
</tr>
</tbody>
</table>
</td>
</tr>
</tbody>
</table>
</div>
</body>
</html>