Index: src/main/org/jboss/mq/pm/jdbc2/PersistenceManager.java
===================================================================
--- src/main/org/jboss/mq/pm/jdbc2/PersistenceManager.java	(revision 85208)
+++ src/main/org/jboss/mq/pm/jdbc2/PersistenceManager.java	(working copy)
@@ -418,7 +418,7 @@
 
          // Find out what the next TXID should be
          stmt = c.prepareStatement(SELECT_MAX_TX);
-         rs = stmt.executeQuery();
+         rs = executeQueryRetrying(stmt);
          if (rs.next())
             nextTransactionId.set(rs.getLong(1) + 1);
          rs.close();
@@ -428,7 +428,7 @@
 
          // Delete all the temporary messages.
          stmt = c.prepareStatement(DELETE_TEMPORARY_MESSAGES);
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
          stmt.close();
          stmt = null;
 
@@ -438,7 +438,7 @@
             deleteMarkedMessagesWithTx = DELETE_MARKED_MESSAGES_WITH_TX_XARECOVERY;
          stmt = c.prepareStatement(deleteMarkedMessagesWithTx);
          stmt.setString(1, "A");
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
          stmt.close();
          stmt = null;
 
@@ -450,7 +450,7 @@
          stmt.setNull(1, java.sql.Types.BIGINT);
          stmt.setString(2, "A");
          stmt.setString(3, "D");
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
          stmt.close();
          stmt = null;
 
@@ -459,7 +459,7 @@
          if (xaRecovery)
             deleteAllTx = DELETE_ALL_TX_XARECOVERY;
          stmt = c.prepareStatement(deleteAllTx);
-         stmt.execute();
+         executeUpdateRetrying(stmt);
          stmt.close();
          stmt = null;
 
@@ -467,7 +467,7 @@
          if (xaRecovery)
          {
             stmt = c.prepareStatement(SELECT_ALL_TX_XARECOVERY);
-            rs = stmt.executeQuery();
+            rs = executeQueryRetrying(stmt);
             while (rs.next())
             {
                long txid = rs.getLong(1);
@@ -618,7 +618,7 @@
 
          long txid = 0;
          String txop = null;
-         rs = stmt.executeQuery();
+         rs = executeQueryRetrying(stmt);
          int counter = 0;
          int recovery = 0;
          while (rs.next())
@@ -641,7 +641,7 @@
                {
                   stmt2.setLong(1, msgid);
                   stmt2.setString(2, dest.toString());
-                  rs2 = stmt2.executeQuery();
+                  rs2 = executeQueryRetrying(stmt2);
                   if (rs2.next())
                   {
                      message = extractMessage(rs2);
@@ -916,7 +916,7 @@
          stmt = c.prepareStatement(DELETE_MARKED_MESSAGES);
          stmt.setLong(1, txid.longValue());
          stmt.setString(2, mark);
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
       }
       finally
       {
@@ -949,7 +949,7 @@
             else
                stmt.setNull(2, java.sql.Types.BLOB);
          }
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
       }
       finally
       {
@@ -971,7 +971,7 @@
       {
          stmt = c.prepareStatement(DELETE_TX);
          stmt.setLong(1, txid);
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
       }
       finally
       {
@@ -1014,7 +1014,7 @@
          stmt.setString(2, "A");
          stmt.setString(3, "D");
          stmt.setLong(4, txId.longValue());
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
          stmt.close();
          stmt = null;
       }
@@ -1169,7 +1169,7 @@
             stmt.setNull(4, java.sql.Types.BIGINT);
          stmt.setString(5, mark);
 
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
       }
       finally
       {
@@ -1203,7 +1203,7 @@
          stmt.setString(2, mark);
          stmt.setLong(3, messageid);
          stmt.setString(4, destination);
-         stmt.executeUpdate();
+         executeUpdateRetrying(stmt);
       }
       finally
       {
@@ -1326,7 +1326,7 @@
             setBlob(stmt, 1, messageRef.getMessage());
             stmt.setLong(2, messageRef.messageId);
             stmt.setString(3, messageRef.getPersistentKey());
-            int rc = stmt.executeUpdate();
+            int rc = executeUpdateRetrying(stmt);
             if (rc != 1)
                throw new SpyJMSException(
                   "Could not update the message in the database: update affected " + rc + " rows");
@@ -1455,9 +1455,12 @@
                       log.error("Retried " + tries + " times, now giving up");
                       throw new IllegalStateException("Could not remove message after " +tries + "attempts");
                    }
-                   log.warn("Trying again after a pause");
                    //Now we wait for a random amount of time to minimise risk of deadlock
-                   Thread.sleep((long)(Math.random() * 500));
+                   final long randomSleepInterval = getRandomSleepInterval();
+                   log.warn("SQLException! Try: " + tries + " of " + statementRetries +
+                                   ". Trying again after a pause of " + randomSleepInterval + "ms. " +
+                                   e.toString());
+                   Thread.sleep(randomSleepInterval);
                 }
              }
 
@@ -1532,7 +1535,7 @@
          stmt.setLong(1, messageRef.messageId);
          stmt.setString(2, messageRef.getPersistentKey());
 
-         rs = stmt.executeQuery();
+         rs = executeQueryRetrying(stmt);
          if (rs.next())
             return extractMessage(rs);
          else
@@ -1994,4 +1997,146 @@
          statementRetries = 0;
       this.statementRetries = statementRetries;
    }
+
+
+   /**
+    * Calls executeUpdate() on the given statement object
+    * retrying at most <code>statementRetries</code>
+    * times if an SQLException is thrown.
+    *
+    * @param stmt
+    * @return the modified record count, as returned by the executeUpdate() method
+    * @throws SQLException if the update fails <code>statementRetries</code> times
+    */
+   protected int executeUpdateRetrying(PreparedStatement stmt)
+       throws SQLException
+   {
+       int tries = 0;
+       boolean threadWasInterrupted = Thread.interrupted();
+       try
+       {
+           while (true)
+           {
+               try
+               {
+                   int rc = stmt.executeUpdate();
+                   if (tries > 0)
+                   {
+                       log.warn("JDBC operation worked after " + tries + " retries");
+                   }
+                   return rc;
+               }
+               catch (SQLException sqlEx)
+               {
+                   tries++;
+                   log.warn("SQLException caught, try: " + tries, sqlEx);
+                   if (tries >= statementRetries)
+                   {
+                       log.error("Giving up on SQLException after " + tries + " tries", sqlEx);
+                       throw sqlEx;
+                   }
+
+                   final long randomSleepInterval = getRandomSleepInterval();
+                   log.warn("SQLException! try: " + tries + " of " + statementRetries +
+                                   ". Trying again after a pause of " + randomSleepInterval + "ms. " +
+                                   sqlEx.toString());
+                   try
+                   {
+                       Thread.sleep(randomSleepInterval);
+                   }
+                   catch (InterruptedException ie)
+                   {
+                       threadWasInterrupted = true;
+                       log.warn("Thread interrupted while waiting for next retry, rethrowing SQLException", ie);
+                       throw sqlEx;
+                   }
+               }
+           }
+       }
+       finally
+       {
+           //restore interrupted status
+           if (threadWasInterrupted)
+           {
+               Thread.currentThread().interrupt();
+           }
+       }
+   }
+
+
+   /**
+    * Calls executeQuery() on the given statement object,
+    * retrying at most <code>statementRetries</code>
+    * times if an SQLException is thrown.
+    *
+    * @param stmt
+    * @return the result set, as returned by the executeQuery() method
+    * @throws SQLException if the query fails <code>statementRetries</code> times
+    */
+   protected ResultSet executeQueryRetrying(PreparedStatement stmt)
+       throws SQLException
+   {
+       int tries = 0;
+       boolean threadWasInterrupted = Thread.interrupted();
+       try
+       {
+           while (true)
+           {
+               try
+               {
+                   ResultSet rs = stmt.executeQuery();
+                   if (tries > 0)
+                   {
+                       log.warn("JDBC operation worked after " + tries + " retries");
+                   }
+                   return rs;
+               }
+               catch (SQLException sqlEx)
+               {
+                   tries++;
+                   log.warn("SQLException caught, try: " + tries, sqlEx);
+                   if (tries >= statementRetries)
+                   {
+                       log.error("Giving up on SQLException after " + tries + " tries", sqlEx);
+                       throw sqlEx;
+                   }
+
+                   final long randomSleepInterval = getRandomSleepInterval();
+                   log.warn("SQLException! Try: " + tries + " of " + statementRetries +
+                                   ". Trying again after a pause of " + randomSleepInterval + "ms. " +
+                                   sqlEx.toString());
+                   try
+                   {
+                       Thread.sleep(randomSleepInterval);
+                   }
+                   catch (InterruptedException ie)
+                   {
+                       threadWasInterrupted = true;
+                       log.warn("Thread interrupted while waiting for next retry, rethrowing SQLException", ie);
+                       throw sqlEx;
+                   }
+               }
+           }
+       }
+       finally
+       {
+           //restore interrupted status
+           if (threadWasInterrupted)
+           {
+               Thread.currentThread().interrupt();
+           }
+       }
+   }
+
+
+   /**
+    * @return a pseudorandom long greater than or equal to 0 and less than 500.
+    */
+   protected static long getRandomSleepInterval()
+   {
+       // wait for a random amount of time to minimise risk of deadlock,
+       // in case the Exception was caused by a deadlock
+       return (long) (Math.random() * 500);
+   }
+
 }
