protected IQueueMessage < ID , DATA > _takeWithRetries ( Connection conn , int numRetries , int maxRetries ) { try { jdbcHelper . startTransaction ( conn ) ; conn . setTransactionIsolation ( transactionIsolationLevel ) ; boolean result = true ; IQueueMessage < ID , DATA > msg = readFromQueueStorage ( conn ) ; if ( msg != null ) { result = result && removeFromQueueStorage ( conn , msg ) ; if ( ! isEphemeralDisabled ( ) ) { try { result = result && putToEphemeralStorage ( conn , msg ) ; } catch ( DuplicatedValueException dve ) { LOGGER . warn ( dve . getMessage ( ) , dve ) ; } catch ( DaoException de ) { if ( de . getCause ( ) instanceof DuplicatedValueException ) { LOGGER . warn ( de . getMessage ( ) , de ) ; } else { throw de ; } } } } if ( result ) { jdbcHelper . commitTransaction ( conn ) ; return msg ; } else { jdbcHelper . rollbackTransaction ( conn ) ; return null ; } } catch ( DaoException de ) { if ( de . getCause ( ) instanceof ConcurrencyFailureException ) { jdbcHelper . rollbackTransaction ( conn ) ; if ( numRetries > maxRetries ) { throw new QueueException ( de ) ; } else { incRetryCounter ( "_takeWithRetries" ) ; return _takeWithRetries ( conn , numRetries + 1 , maxRetries ) ; } } throw de ; } catch ( Exception e ) { jdbcHelper . rollbackTransaction ( conn ) ; throw e instanceof QueueException ? ( QueueException ) e : new QueueException ( e ) ; } }
Take a message from queue retry if deadlock .