log.info("Received:" + numMessages);
   public void run()   {                  //Small pause so as not to miss any messages in a topic      try      {         Thread.sleep(1000);      }      catch (InterruptedException e)      {               }            try      {                      while (count < numMessages)         {            Message m = getMessage();                        if (m == null)            {               log.error("Message is null");               failed = true;               processingDone();               return;            }            String prodName = m.getStringProperty("PROD_NAME");            Integer msgCount = new Integer(m.getIntProperty("MSG_NUMBER"));                                  Integer prevCount = (Integer)counts.get(prodName);            if (prevCount == null)            {               if (msgCount.intValue() != 0)               {                  log.error("First message received not zero");                  failed = true;                  processingDone();                  return;               }                           }            else            {               if (prevCount.intValue() != msgCount.intValue() - 1)               {                  log.error("Message out of sequence");                  failed = true;                  processingDone();                  return;               }            }            counts.put(prodName, msgCount);                        count++;                        processingDone();         }                  log.info("Received:" + numMessages);      }      catch (Exception e)      {         log.error("Failed to receive message", e);         failed = true;      }   }