failed = true; return;
   public void run()   {                  //Small pause so as not to miss any messages in a topic      try      {         Thread.sleep(1000);      }      catch (InterruptedException e)      {               }            try      {               int iterations = numMessages / commitSize;         for (int outerCount = 0; outerCount < iterations; outerCount++)         {                        XidImpl xid = null;            if (commitSize > 0)            {               xid = new XidImpl("bq1".getBytes(), 1, new GUID().toString().getBytes());               xaResource.start(xid, XAResource.TMNOFLAGS);            }                                    for (int innerCount = 0; innerCount < commitSize; innerCount++)            {               Message m = getMessage();                                       if (m == null)               {                  log.error("Message is null");                  failed = true;                  return;               }               String prodName = m.getStringProperty("PROD_NAME");               Integer msgCount = new Integer(m.getIntProperty("MSG_NUMBER"));                              Count count = (Count)counts.get(prodName);               if (count == null)               {                  //First time                  if (msgCount.intValue() != 0)                  {                     log.error("First message from " + prodName + " is not 0, it is " + msgCount);                     failed = true;                     return;                  }                  else                  {                     count = new Count();                     counts.put(prodName, count);                  }               }               else               {                  if (count.lastCommitted != msgCount.intValue() - 1)                  {                     log.error("Message out of sequence for " + prodName + ", expected " + (count.lastCommitted + 1) + ", actual " + msgCount);                     //failed = true;                     //return;                  }               }               count.lastCommitted = msgCount.intValue();                              count.lastReceived = msgCount.intValue();                              if (innerCount == commitSize -1)               {                  xaResource.end(xid, XAResource.TMSUCCESS);                  xaResource.prepare(xid);                  xaResource.commit(xid, false);                                    if (rollbackSize > 0 && outerCount != iterations - 1)                  {                     //Starting new tx                     xid = new XidImpl("bq1".getBytes(), 1, new GUID().toString().getBytes());                     xaResource.start(xid, XAResource.TMNOFLAGS);                  }               }                              processingDone();                        }                                                if (outerCount == iterations - 1)            {               break;            }                                                   for (int innerCount = 0; innerCount < rollbackSize; innerCount++)            {               Message m = getMessage();                              if (m == null)               {                  log.error("Message is null (rollback)");                  failed = true;                  return;               }               String prodName = m.getStringProperty("PROD_NAME");                              Integer msgCount = new Integer(m.getIntProperty("MSG_NUMBER"));                              Count count = (Count)counts.get(prodName);               if (count == null)               {                  // First time                  if (msgCount.intValue() != 0)                  {                     log.error("First message from " + prodName + " is not 0, it is " + msgCount);                     failed = true;                     return;                  }                  else                  {                     count = new Count();                     count.lastCommitted = -1;                     counts.put(prodName, count);                  }                                }               else               {                  if (count.lastReceived != msgCount.intValue() - 1)                  {                     log.error("Message out of sequence");                     failed = true;                     return;                  }               }               count.lastReceived = msgCount.intValue();                                        if (innerCount == rollbackSize -1)               {                  xaResource.end(xid, XAResource.TMSUCCESS);                  xaResource.prepare(xid);                  xaResource.rollback(xid);               }               processingDone();            }                     }             }      catch (Exception e)      {         log.error("Failed to receive message", e);         failed = true;      }   }